diff --git a/Lib/test/test_timeout.py b/Lib/test/test_timeout.py index 70a0175d77..967d4ff7e1 100644 --- a/Lib/test/test_timeout.py +++ b/Lib/test/test_timeout.py @@ -5,9 +5,6 @@ from test import support from test.support import socket_helper -# This requires the 'network' resource as given on the regrtest command line. -skip_expected = not support.is_resource_enabled('network') - import time import errno import socket @@ -29,10 +26,8 @@ class CreationTestCase(unittest.TestCase): """Test case for socket.gettimeout() and socket.settimeout()""" def setUp(self): - self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - - def tearDown(self): - self.sock.close() + self.sock = self.enterContext( + socket.socket(socket.AF_INET, socket.SOCK_STREAM)) def testObjectCreation(self): # Test Socket creation @@ -53,10 +48,10 @@ def testFloatReturnValue(self): def testReturnType(self): # Test return type of gettimeout() self.sock.settimeout(1) - self.assertEqual(type(self.sock.gettimeout()), type(1.0)) + self.assertIs(type(self.sock.gettimeout()), float) self.sock.settimeout(3.9) - self.assertEqual(type(self.sock.gettimeout()), type(1.0)) + self.assertIs(type(self.sock.gettimeout()), float) def testTypeCheck(self): # Test type checking by settimeout() @@ -116,8 +111,6 @@ class TimeoutTestCase(unittest.TestCase): def setUp(self): raise NotImplementedError() - tearDown = setUp - def _sock_operation(self, count, timeout, method, *args): """ Test the specified socket method. @@ -145,19 +138,16 @@ class TCPTimeoutTestCase(TimeoutTestCase): """TCP test case for socket.socket() timeout functions""" def setUp(self): - self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock = self.enterContext( + socket.socket(socket.AF_INET, socket.SOCK_STREAM)) self.addr_remote = resolve_address('www.python.org.', 80) - def tearDown(self): - self.sock.close() - - @unittest.skipIf(True, 'need to replace these hosts; see bpo-35518') def testConnectTimeout(self): # Testing connect timeout is tricky: we need to have IP connectivity # to a host that silently drops our packets. We can't simulate this # from Python because it's a function of the underlying TCP/IP stack. - # So, the following Snakebite host has been defined: - blackhole = resolve_address('blackhole.snakebite.net', 56666) + # So, the following port on the pythontest.net host has been defined: + blackhole = resolve_address('pythontest.net', 56666) # Blackhole has been configured to silently drop any incoming packets. # No RSTs (for TCP) or ICMP UNREACH (for UDP/ICMP) will be sent back @@ -169,7 +159,7 @@ def testConnectTimeout(self): # to firewalling or general network configuration. In order to improve # our confidence in testing the blackhole, a corresponding 'whitehole' # has also been set up using one port higher: - whitehole = resolve_address('whitehole.snakebite.net', 56667) + whitehole = resolve_address('pythontest.net', 56667) # This address has been configured to immediately drop any incoming # packets as well, but it does it respectfully with regards to the @@ -183,35 +173,27 @@ def testConnectTimeout(self): # timeframe). # For the records, the whitehole/blackhole configuration has been set - # up using the 'pf' firewall (available on BSDs), using the following: + # up using the 'iptables' firewall, using the following rules: # - # ext_if="bge0" - # - # blackhole_ip="35.8.247.6" - # whitehole_ip="35.8.247.6" - # blackhole_port="56666" - # whitehole_port="56667" - # - # block return in log quick on $ext_if proto { tcp udp } \ - # from any to $whitehole_ip port $whitehole_port - # block drop in log quick on $ext_if proto { tcp udp } \ - # from any to $blackhole_ip port $blackhole_port + # -A INPUT -p tcp --destination-port 56666 -j DROP + # -A INPUT -p udp --destination-port 56666 -j DROP + # -A INPUT -p tcp --destination-port 56667 -j REJECT + # -A INPUT -p udp --destination-port 56667 -j REJECT # + # See https://github.com/python/psf-salt/blob/main/pillar/base/firewall/snakebite.sls + # for the current configuration. skip = True - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - timeout = support.LOOPBACK_TIMEOUT - sock.settimeout(timeout) - try: - sock.connect((whitehole)) - except TimeoutError: - pass - except OSError as err: - if err.errno == errno.ECONNREFUSED: - skip = False - finally: - sock.close() - del sock + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + try: + timeout = support.LOOPBACK_TIMEOUT + sock.settimeout(timeout) + sock.connect((whitehole)) + except TimeoutError: + pass + except OSError as err: + if err.errno == errno.ECONNREFUSED: + skip = False if skip: self.skipTest( @@ -278,10 +260,8 @@ class UDPTimeoutTestCase(TimeoutTestCase): """UDP test case for socket.socket() timeout functions""" def setUp(self): - self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - - def tearDown(self): - self.sock.close() + self.sock = self.enterContext( + socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) def testRecvfromTimeout(self): # Test recvfrom() timeout @@ -292,6 +272,7 @@ def testRecvfromTimeout(self): def setUpModule(): support.requires('network') + support.requires_working_socket(module=True) if __name__ == "__main__":