[Python-checkins] bpo-38614: Use support timeout constants (GH-17572)
Victor Stinner
webhook-mailer at python.org
Wed Dec 11 16:17:29 EST 2019
https://github.com/python/cpython/commit/7772b1af5ebc9d72d0cfc8332aea6b2143eafa27
commit: 7772b1af5ebc9d72d0cfc8332aea6b2143eafa27
branch: master
author: Victor Stinner <vstinner at python.org>
committer: GitHub <noreply at github.com>
date: 2019-12-11T22:17:04+01:00
summary:
bpo-38614: Use support timeout constants (GH-17572)
files:
M Lib/test/_test_multiprocessing.py
M Lib/test/test_embed.py
M Lib/test/test_poplib.py
M Lib/test/test_smtplib.py
M Lib/test/test_socket.py
diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 399be13f3a260..c0da7d8912e64 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -1144,7 +1144,7 @@ def __reduce__(self):
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 @@ def __reduce__(self):
# 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 @@ def _on_queue_feeder_error(e, obj):
# 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 60f7f7a93ea1a..73ef96265b711 100644
--- a/Lib/test/test_embed.py
+++ b/Lib/test/test_embed.py
@@ -1303,10 +1303,14 @@ def test_audit_subinterpreter(self):
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 @@ def test_audit_run_interactivehook(self):
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 @@ def test_audit_run_startup(self):
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 7b1d854d1c086..911cba1f1dcd4 100644
--- a/Lib/test/test_poplib.py
+++ b/Lib/test/test_poplib.py
@@ -255,7 +255,8 @@ def assertOK(self, resp):
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 @@ def test_stls_context(self):
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 a3e5b3f3774c6..faf013ac9a66e 100644
--- a/Lib/test/test_smtplib.py
+++ b/Lib/test/test_smtplib.py
@@ -964,11 +964,13 @@ def tearDown(self):
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 @@ def testEHLO(self):
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 @@ def testVRFY(self):
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 @@ def testEXPN(self):
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 @@ def testAUTH_LOGIN(self):
@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 @@ def testAUTH_CRAM_MD5(self):
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 @@ def test_auth_function(self):
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 @@ def test_auth_function(self):
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 @@ def test_with_statement_QUIT_failure(self):
# 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 @@ def test__rest_from_mail_cmd(self):
# 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 @@ def test_421_from_mail_cmd(self):
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 @@ def found_terminator(self):
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 at foo.org', ['Sally at foo.org'], 'test message')
@@ -1393,15 +1406,15 @@ def tearDown(self):
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 184c67c56ebd0..64e95ea34b997 100755
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -5063,14 +5063,16 @@ def _justAccept(self):
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 @@ def testOffset(self):
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 @@ def testCount(self):
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 @@ def testNonBlocking(self):
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)
More information about the Python-checkins
mailing list