mirror of https://github.com/python/cpython
Issue #19589: Use specific asserts in asyncio tests.
This commit is contained in:
parent
f38f7f3c7b
commit
e048addedc
|
@ -472,8 +472,8 @@ class EventLoopTestsMixin:
|
|||
f = self.loop.create_connection(
|
||||
lambda: MyProto(loop=self.loop), *httpd.address)
|
||||
tr, pr = self.loop.run_until_complete(f)
|
||||
self.assertTrue(isinstance(tr, transports.Transport))
|
||||
self.assertTrue(isinstance(pr, protocols.Protocol))
|
||||
self.assertIsInstance(tr, transports.Transport)
|
||||
self.assertIsInstance(pr, protocols.Protocol)
|
||||
self.loop.run_until_complete(pr.done)
|
||||
self.assertGreater(pr.nbytes, 0)
|
||||
tr.close()
|
||||
|
@ -500,8 +500,8 @@ class EventLoopTestsMixin:
|
|||
f = self.loop.create_connection(
|
||||
lambda: MyProto(loop=self.loop), sock=sock)
|
||||
tr, pr = self.loop.run_until_complete(f)
|
||||
self.assertTrue(isinstance(tr, transports.Transport))
|
||||
self.assertTrue(isinstance(pr, protocols.Protocol))
|
||||
self.assertIsInstance(tr, transports.Transport)
|
||||
self.assertIsInstance(pr, protocols.Protocol)
|
||||
self.loop.run_until_complete(pr.done)
|
||||
self.assertGreater(pr.nbytes, 0)
|
||||
tr.close()
|
||||
|
@ -513,8 +513,8 @@ class EventLoopTestsMixin:
|
|||
lambda: MyProto(loop=self.loop), *httpd.address,
|
||||
ssl=test_utils.dummy_ssl_context())
|
||||
tr, pr = self.loop.run_until_complete(f)
|
||||
self.assertTrue(isinstance(tr, transports.Transport))
|
||||
self.assertTrue(isinstance(pr, protocols.Protocol))
|
||||
self.assertIsInstance(tr, transports.Transport)
|
||||
self.assertIsInstance(pr, protocols.Protocol)
|
||||
self.assertTrue('ssl' in tr.__class__.__name__.lower())
|
||||
self.assertIsNotNone(tr.get_extra_info('sockname'))
|
||||
self.loop.run_until_complete(pr.done)
|
||||
|
@ -926,7 +926,8 @@ class EventLoopTestsMixin:
|
|||
r.setblocking(False)
|
||||
f = self.loop.sock_recv(r, 1)
|
||||
ov = getattr(f, 'ov', None)
|
||||
self.assertTrue(ov is None or ov.pending)
|
||||
if ov is not None:
|
||||
self.assertTrue(ov.pending)
|
||||
|
||||
@tasks.coroutine
|
||||
def main():
|
||||
|
@ -949,7 +950,8 @@ class EventLoopTestsMixin:
|
|||
self.assertLess(elapsed, 0.1)
|
||||
self.assertEqual(t.result(), 'cancelled')
|
||||
self.assertRaises(futures.CancelledError, f.result)
|
||||
self.assertTrue(ov is None or not ov.pending)
|
||||
if ov is not None:
|
||||
self.assertFalse(ov.pending)
|
||||
self.loop._stop_serving(r)
|
||||
|
||||
r.close()
|
||||
|
|
|
@ -779,7 +779,7 @@ class TaskTests(unittest.TestCase):
|
|||
self.assertEqual(len(res), 2, res)
|
||||
self.assertEqual(res[0], (1, 'a'))
|
||||
self.assertEqual(res[1][0], 2)
|
||||
self.assertTrue(isinstance(res[1][1], futures.TimeoutError))
|
||||
self.assertIsInstance(res[1][1], futures.TimeoutError)
|
||||
self.assertAlmostEqual(0.12, loop.time())
|
||||
|
||||
# move forward to close generator
|
||||
|
|
|
@ -67,7 +67,7 @@ class SelectorEventLoopTests(unittest.TestCase):
|
|||
cb = lambda: True
|
||||
self.loop.add_signal_handler(signal.SIGHUP, cb)
|
||||
h = self.loop._signal_handlers.get(signal.SIGHUP)
|
||||
self.assertTrue(isinstance(h, events.Handle))
|
||||
self.assertIsInstance(h, events.Handle)
|
||||
self.assertEqual(h._callback, cb)
|
||||
|
||||
@unittest.mock.patch('asyncio.unix_events.signal')
|
||||
|
|
Loading…
Reference in New Issue