bpo-38614: Use support timeout constants (GH-17572)

diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 399be13f3a2601f6758f7805cacb4cebced3ac4c..c0da7d8912e64f3f4f79305feb90733bbf6e33f7 100644 (file)
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -1144,7 +1144,7 @@ class _TestQueue(BaseTestCase):
             q = self.Queue()
             q.put(NotSerializable())
             q.put(True)
-            self.assertTrue(q.get(timeout=support.LONG_TIMEOUT))
+            self.assertTrue(q.get(timeout=support.SHORT_TIMEOUT))
             close_queue(q)
 
         with test.support.captured_stderr():
@@ -1159,8 +1159,7 @@ class _TestQueue(BaseTestCase):
                 # qsize is not available on all platform as it
                 # relies on sem_getvalue
                 pass
-            # bpo-30595: use a timeout of 1 second for slow buildbots
-            self.assertTrue(q.get(timeout=1.0))
+            self.assertTrue(q.get(timeout=support.SHORT_TIMEOUT))
             # Check that the size of the queue is correct
             self.assertTrue(q.empty())
             close_queue(q)
@@ -1197,7 +1196,7 @@ class _TestQueue(BaseTestCase):
 
             # Verify that q is still functioning correctly
             q.put(True)
-            self.assertTrue(q.get(timeout=1.0))
+            self.assertTrue(q.get(timeout=support.SHORT_TIMEOUT))
 
         # Assert that the serialization and the hook have been called correctly
         self.assertTrue(not_serializable_obj.reduce_was_called)

diff --git a/Lib/test/test_embed.py b/Lib/test/test_embed.py
index 60f7f7a93ea1ad0bb3e31bc5a1c5ee4611aa36df..73ef96265b7119f0c3cd5097506d17c54fd0e91f 100644 (file)
--- a/Lib/test/test_embed.py
+++ b/Lib/test/test_embed.py
@@ -1303,10 +1303,14 @@ class AuditingTests(EmbeddingTestsMixin, unittest.TestCase):
         self.run_embedded_interpreter("test_audit_subinterpreter")
 
     def test_audit_run_command(self):
-        self.run_embedded_interpreter("test_audit_run_command", timeout=3, returncode=1)
+        self.run_embedded_interpreter("test_audit_run_command",
+                                      timeout=support.SHORT_TIMEOUT,
+                                      returncode=1)
 
     def test_audit_run_file(self):
-        self.run_embedded_interpreter("test_audit_run_file", timeout=3, returncode=1)
+        self.run_embedded_interpreter("test_audit_run_file",
+                                      timeout=support.SHORT_TIMEOUT,
+                                      returncode=1)
 
     def test_audit_run_interactivehook(self):
         startup = os.path.join(self.oldcwd, support.TESTFN) + ".py"
@@ -1315,7 +1319,8 @@ class AuditingTests(EmbeddingTestsMixin, unittest.TestCase):
             print("sys.__interactivehook__ = lambda: None", file=f)
         try:
             env = {**remove_python_envvars(), "PYTHONSTARTUP": startup}
-            self.run_embedded_interpreter("test_audit_run_interactivehook", timeout=5,
+            self.run_embedded_interpreter("test_audit_run_interactivehook",
+                                          timeout=support.SHORT_TIMEOUT,
                                           returncode=10, env=env)
         finally:
             os.unlink(startup)
@@ -1326,13 +1331,16 @@ class AuditingTests(EmbeddingTestsMixin, unittest.TestCase):
             print("pass", file=f)
         try:
             env = {**remove_python_envvars(), "PYTHONSTARTUP": startup}
-            self.run_embedded_interpreter("test_audit_run_startup", timeout=5,
+            self.run_embedded_interpreter("test_audit_run_startup",
+                                          timeout=support.SHORT_TIMEOUT,
                                           returncode=10, env=env)
         finally:
             os.unlink(startup)
 
     def test_audit_run_stdin(self):
-        self.run_embedded_interpreter("test_audit_run_stdin", timeout=3, returncode=1)
+        self.run_embedded_interpreter("test_audit_run_stdin",
+                                      timeout=support.SHORT_TIMEOUT,
+                                      returncode=1)
 
 if __name__ == "__main__":
     unittest.main()

diff --git a/Lib/test/test_poplib.py b/Lib/test/test_poplib.py
index 7b1d854d1c08666c0bdbcbe4040380f12e6fe8ab..911cba1f1dcd499c23a7bf72a4be2c83d36255c8 100644 (file)
--- a/Lib/test/test_poplib.py
+++ b/Lib/test/test_poplib.py
@@ -255,7 +255,8 @@ class TestPOP3Class(TestCase):
     def setUp(self):
         self.server = DummyPOP3Server((HOST, PORT))
         self.server.start()
-        self.client = poplib.POP3(self.server.host, self.server.port, timeout=3)
+        self.client = poplib.POP3(self.server.host, self.server.port,
+                                  timeout=test_support.LOOPBACK_TIMEOUT)
 
     def tearDown(self):
         self.client.close()
@@ -376,7 +377,8 @@ class TestPOP3Class(TestCase):
         self.assertEqual(ctx.check_hostname, True)
         with self.assertRaises(ssl.CertificateError):
             resp = self.client.stls(context=ctx)
-        self.client = poplib.POP3("localhost", self.server.port, timeout=3)
+        self.client = poplib.POP3("localhost", self.server.port,
+                                  timeout=test_support.LOOPBACK_TIMEOUT)
         resp = self.client.stls(context=ctx)
         self.assertEqual(resp, expected)
 
@@ -445,7 +447,8 @@ class TestPOP3_TLSClass(TestPOP3Class):
     def setUp(self):
         self.server = DummyPOP3Server((HOST, PORT))
         self.server.start()
-        self.client = poplib.POP3(self.server.host, self.server.port, timeout=3)
+        self.client = poplib.POP3(self.server.host, self.server.port,
+                                  timeout=test_support.LOOPBACK_TIMEOUT)
         self.client.stls()
 
     def tearDown(self):

diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py
index a3e5b3f3774c6ebe7252cef18ca67ada3f4a03c9..faf013ac9a66e0e56070e3ba5f78ec36db1aea8d 100644 (file)
--- a/Lib/test/test_smtplib.py
+++ b/Lib/test/test_smtplib.py
@@ -964,11 +964,13 @@ class SMTPSimTests(unittest.TestCase):
 
     def testBasic(self):
         # smoke test
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.quit()
 
     def testEHLO(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
 
         # no features should be present before the EHLO
         self.assertEqual(smtp.esmtp_features, {})
@@ -989,7 +991,8 @@ class SMTPSimTests(unittest.TestCase):
         smtp.quit()
 
     def testVRFY(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
 
         for addr_spec, name in sim_users.items():
             expected_known = (250, bytes('%s %s' %
@@ -1003,7 +1006,8 @@ class SMTPSimTests(unittest.TestCase):
         smtp.quit()
 
     def testEXPN(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
 
         for listname, members in sim_lists.items():
             users = []
@@ -1019,14 +1023,16 @@ class SMTPSimTests(unittest.TestCase):
 
     def testAUTH_PLAIN(self):
         self.serv.add_feature("AUTH PLAIN")
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         resp = smtp.login(sim_auth[0], sim_auth[1])
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
 
     def testAUTH_LOGIN(self):
         self.serv.add_feature("AUTH LOGIN")
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         resp = smtp.login(sim_auth[0], sim_auth[1])
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
@@ -1034,7 +1040,8 @@ class SMTPSimTests(unittest.TestCase):
     @requires_hashdigest('md5')
     def testAUTH_CRAM_MD5(self):
         self.serv.add_feature("AUTH CRAM-MD5")
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         resp = smtp.login(sim_auth[0], sim_auth[1])
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
@@ -1042,7 +1049,8 @@ class SMTPSimTests(unittest.TestCase):
     def testAUTH_multiple(self):
         # Test that multiple authentication methods are tried.
         self.serv.add_feature("AUTH BOGUS PLAIN LOGIN CRAM-MD5")
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         resp = smtp.login(sim_auth[0], sim_auth[1])
         self.assertEqual(resp, (235, b'Authentication Succeeded'))
         smtp.close()
@@ -1060,7 +1068,8 @@ class SMTPSimTests(unittest.TestCase):
         for mechanism in supported:
             with self.subTest(mechanism=mechanism):
                 smtp = smtplib.SMTP(HOST, self.port,
-                                    local_hostname='localhost', timeout=15)
+                                    local_hostname='localhost',
+                                    timeout=support.LOOPBACK_TIMEOUT)
                 smtp.ehlo('foo')
                 smtp.user, smtp.password = sim_auth[0], sim_auth[1]
                 method = 'auth_' + mechanism.lower().replace('-', '_')
@@ -1071,7 +1080,7 @@ class SMTPSimTests(unittest.TestCase):
     def test_quit_resets_greeting(self):
         smtp = smtplib.SMTP(HOST, self.port,
                             local_hostname='localhost',
-                            timeout=15)
+                            timeout=support.LOOPBACK_TIMEOUT)
         code, message = smtp.ehlo()
         self.assertEqual(code, 250)
         self.assertIn('size', smtp.esmtp_features)
@@ -1105,7 +1114,8 @@ class SMTPSimTests(unittest.TestCase):
 
     # Issue 17498: make sure _rset does not raise SMTPServerDisconnected exception
     def test__rest_from_mail_cmd(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.noop()
         self.serv._SMTPchannel.mail_response = '451 Requested action aborted'
         self.serv._SMTPchannel.disconnect = True
@@ -1115,7 +1125,8 @@ class SMTPSimTests(unittest.TestCase):
 
     # Issue 5713: make sure close, not rset, is called if we get a 421 error
     def test_421_from_mail_cmd(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.noop()
         self.serv._SMTPchannel.mail_response = '421 closing connection'
         with self.assertRaises(smtplib.SMTPSenderRefused):
@@ -1124,7 +1135,8 @@ class SMTPSimTests(unittest.TestCase):
         self.assertEqual(self.serv._SMTPchannel.rset_count, 0)
 
     def test_421_from_rcpt_cmd(self):
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.noop()
         self.serv._SMTPchannel.rcpt_response = ['250 accepted', '421 closing']
         with self.assertRaises(smtplib.SMTPRecipientsRefused) as r:
@@ -1141,7 +1153,8 @@ class SMTPSimTests(unittest.TestCase):
                 else:
                     super().found_terminator()
         self.serv.channel_class = MySimSMTPChannel
-        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.noop()
         with self.assertRaises(smtplib.SMTPDataError):
             smtp.sendmail('John@foo.org', ['Sally@foo.org'], 'test message')
@@ -1393,15 +1406,15 @@ class SMTPAUTHInitialResponseSimTests(unittest.TestCase):
 
     def testAUTH_PLAIN_initial_response_login(self):
         self.serv.add_feature('AUTH PLAIN')
-        smtp = smtplib.SMTP(HOST, self.port,
-                            local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.login('psu', 'doesnotexist')
         smtp.close()
 
     def testAUTH_PLAIN_initial_response_auth(self):
         self.serv.add_feature('AUTH PLAIN')
-        smtp = smtplib.SMTP(HOST, self.port,
-                            local_hostname='localhost', timeout=15)
+        smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost',
+                            timeout=support.LOOPBACK_TIMEOUT)
         smtp.user = 'psu'
         smtp.password = 'doesnotexist'
         code, response = smtp.auth('plain', smtp.auth_plain)

diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index 184c67c56ebd08698bdae970c31ad4c3195daf50..64e95ea34b99730688eff599520cebb1709ed482 100755 (executable)
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -5063,14 +5063,16 @@ class NetworkConnectionAttributesTest(SocketTCPTest, ThreadableTest):
 
     testFamily = _justAccept
     def _testFamily(self):
-        self.cli = socket.create_connection((HOST, self.port), timeout=30)
+        self.cli = socket.create_connection((HOST, self.port),
+                            timeout=support.LOOPBACK_TIMEOUT)
         self.addCleanup(self.cli.close)
         self.assertEqual(self.cli.family, 2)
 
     testSourceAddress = _justAccept
     def _testSourceAddress(self):
-        self.cli = socket.create_connection((HOST, self.port), timeout=30,
-                source_address=('', self.source_port))
+        self.cli = socket.create_connection((HOST, self.port),
+                            timeout=support.LOOPBACK_TIMEOUT,
+                            source_address=('', self.source_port))
         self.addCleanup(self.cli.close)
         self.assertEqual(self.cli.getsockname()[1], self.source_port)
         # The port number being used is sufficient to show that the bind()
@@ -5959,7 +5961,9 @@ class SendfileUsingSendTest(ThreadedTCPSocketTest):
     def _testCount(self):
         address = self.serv.getsockname()
         file = open(support.TESTFN, 'rb')
-        with socket.create_connection(address, timeout=2) as sock, file as file:
+        sock = socket.create_connection(address,
+                                        timeout=support.LOOPBACK_TIMEOUT)
+        with sock, file:
             count = 5000007
             meth = self.meth_from_sock(sock)
             sent = meth(file, count=count)
@@ -5978,7 +5982,9 @@ class SendfileUsingSendTest(ThreadedTCPSocketTest):
     def _testCountSmall(self):
         address = self.serv.getsockname()
         file = open(support.TESTFN, 'rb')
-        with socket.create_connection(address, timeout=2) as sock, file as file:
+        sock = socket.create_connection(address,
+                                        timeout=support.LOOPBACK_TIMEOUT)
+        with sock, file:
             count = 1
             meth = self.meth_from_sock(sock)
             sent = meth(file, count=count)
@@ -6032,7 +6038,9 @@ class SendfileUsingSendTest(ThreadedTCPSocketTest):
     def _testWithTimeout(self):
         address = self.serv.getsockname()
         file = open(support.TESTFN, 'rb')
-        with socket.create_connection(address, timeout=2) as sock, file as file:
+        sock = socket.create_connection(address,
+                                        timeout=support.LOOPBACK_TIMEOUT)
+        with sock, file:
             meth = self.meth_from_sock(sock)
             sent = meth(file)
             self.assertEqual(sent, self.FILESIZE)