Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

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

This commit is contained in:
Victor Stinner 2019-12-11 22:17:04 +01:00 committed by GitHub
parent 0d63bacefd
commit 7772b1af5e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 67 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
Lib/test/_test_multiprocessing.py
View File

@ -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)

18
Lib/test/test_embed.py
View File

@ -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()

9
Lib/test/test_poplib.py
View File

@ -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):

49
Lib/test/test_smtplib.py
View File

@ -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)

20
Lib/test/test_socket.py
View File

@ -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)