mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-31 14:07:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

gh-71339: Use new assertion methods in test_asyncio (#129051)

Browse source
This commit is contained in:
Serhiy Storchaka 2025-01-20 13:32:39 +02:00 • committed by GitHub
parent 6f167d7134
commit c6b570e5e3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 36 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

18
Lib/test/test_asyncio/test_streams.py
View file

@ -50,7 +50,7 @@ class StreamTests(test_utils.TestCase):
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
f = reader.read()
data = self.loop.run_until_complete(f)
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
writer.close()
self.assertEqual(messages, [])
@ -75,7 +75,7 @@ class StreamTests(test_utils.TestCase):
writer.write(b'GET / HTTP/1.0\r\n\r\n')
f = reader.read()
data = self.loop.run_until_complete(f)
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
writer.close()
self.assertEqual(messages, [])
@ -1002,7 +1002,7 @@ class StreamTests(test_utils.TestCase):
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
f = rd.read()
data = self.loop.run_until_complete(f)
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
self.assertFalse(wr.is_closing())
wr.close()
self.assertTrue(wr.is_closing())
@ -1028,7 +1028,7 @@ class StreamTests(test_utils.TestCase):
data = await rd.readline()
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
data = await rd.read()
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
wr.close()
await wr.wait_closed()
@ -1048,7 +1048,7 @@ class StreamTests(test_utils.TestCase):
data = await rd.readline()
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
data = await rd.read()
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
wr.close()
with self.assertRaises(ConnectionResetError):
wr.write(b'data')
@ -1089,12 +1089,12 @@ class StreamTests(test_utils.TestCase):
data = await rd.readline()
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
data = await rd.read()
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
with self.assertWarns(ResourceWarning) as cm:
del wr
gc.collect()
self.assertEqual(len(cm.warnings), 1)
self.assertTrue(str(cm.warnings[0].message).startswith("unclosed <StreamWriter"))
self.assertStartsWith(str(cm.warnings[0].message), "unclosed <StreamWriter")
messages = []
self.loop.set_exception_handler(lambda loop, ctx: messages.append(ctx))
@ -1112,7 +1112,7 @@ class StreamTests(test_utils.TestCase):
data = await rd.readline()
self.assertEqual(data, b'HTTP/1.0 200 OK\r\n')
data = await rd.read()
self.assertTrue(data.endswith(b'\r\n\r\nTest message'))
self.assertEndsWith(data, b'\r\n\r\nTest message')
# Make "loop is closed" occur first before "del wr" for this test.
self.loop.stop()
@ -1144,7 +1144,7 @@ class StreamTests(test_utils.TestCase):
del wr
gc.collect()
self.assertEqual(len(cm.warnings), 1)
self.assertTrue(str(cm.warnings[0].message).startswith("unclosed <StreamWriter"))
self.assertStartsWith(str(cm.warnings[0].message), "unclosed <StreamWriter")
async def outer():
srv = await asyncio.start_server(inner, socket_helper.HOSTv4, 0)