mirrors/cpython
0
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2024-12-01 05:45:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

gh-113980: Fix resource warnings in test_asyncgen (GH-113984)

Browse Source
This commit is contained in:
Serhiy Storchaka 2024-01-12 17:30:26 +02:00 committed by GitHub
parent ed066481c7
commit e02c15b3f1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 19 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
Lib/test/test_asyncgen.py
View File

@ -379,7 +379,10 @@ class AsyncGenTest(unittest.TestCase):
def test_async_gen_exception_12(self):
async def gen():
await anext(me)
with self.assertWarnsRegex(RuntimeWarning,
f"coroutine method 'asend' of '{gen.__qualname__}' "
f"was never awaited"):
await anext(me)
yield 123
me = gen()
@ -395,7 +398,12 @@ class AsyncGenTest(unittest.TestCase):
yield 123
with self.assertWarns(DeprecationWarning):
gen().athrow(GeneratorExit, GeneratorExit(), None)
x = gen().athrow(GeneratorExit, GeneratorExit(), None)
with self.assertWarnsRegex(RuntimeWarning,
f"coroutine method 'athrow' of '{gen.__qualname__}' "
f"was never awaited"):
del x
gc_collect()
def test_async_gen_api_01(self):
async def gen():
@ -1564,6 +1572,11 @@ class AsyncGenAsyncioTest(unittest.TestCase):
self.assertIsInstance(message['exception'], ZeroDivisionError)
self.assertIn('unhandled exception during asyncio.run() shutdown',
message['message'])
with self.assertWarnsRegex(RuntimeWarning,
f"coroutine method 'aclose' of '{async_iterate.__qualname__}' "
f"was never awaited"):
del message, messages
gc_collect()
def test_async_gen_expression_01(self):
async def arange(n):
@ -1617,6 +1630,10 @@ class AsyncGenAsyncioTest(unittest.TestCase):
asyncio.run(main())
self.assertEqual([], messages)
with self.assertWarnsRegex(RuntimeWarning,
f"coroutine method 'aclose' of '{async_iterate.__qualname__}' "
f"was never awaited"):
gc_collect()
def test_async_gen_await_same_anext_coro_twice(self):
async def async_iterate():