def __del__(self):
pass
- async def coro():
+ async def corofn():
await asyncio.sleep(0.01)
- task = Subclass(coro(), loop = self.loop)
+ coro = corofn()
+ task = Subclass(coro, loop = self.loop)
task._log_destroy_pending = False
del task
support.gc_collect()
+ coro.close()
+
@mock.patch('asyncio.base_events.logger')
def test_tb_logger_not_called_after_cancel(self, m_log):
loop = asyncio.new_event_loop()
coro = coroutine_function()
with contextlib.closing(asyncio.EventLoop()) as loop:
task = asyncio.Task.__new__(asyncio.Task)
-
for _ in range(5):
with self.assertRaisesRegex(RuntimeError, 'break'):
task.__init__(coro, loop=loop, context=obj, name=Break())
coro.close()
+ task._log_destroy_pending = False
del task
self.assertEqual(sys.getrefcount(obj), initial_refcount)