diff options
Diffstat (limited to 'Lib/test/test_asyncio/test_tasks.py')
| -rw-r--r-- | Lib/test/test_asyncio/test_tasks.py | 75 |
1 files changed, 25 insertions, 50 deletions
diff --git a/Lib/test/test_asyncio/test_tasks.py b/Lib/test/test_asyncio/test_tasks.py index 05a822ba45..b08068e617 100644 --- a/Lib/test/test_asyncio/test_tasks.py +++ b/Lib/test/test_asyncio/test_tasks.py @@ -210,10 +210,8 @@ class BaseTaskTests: a = notmuch() self.addCleanup(a.close) - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaisesRegex(RuntimeError, 'There is no current event loop'): - asyncio.ensure_future(a) - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'no current event loop'): + asyncio.ensure_future(a) async def test(): return asyncio.ensure_future(notmuch()) @@ -223,12 +221,10 @@ class BaseTaskTests: self.assertTrue(t.done()) self.assertEqual(t.result(), 'ok') - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 asyncio.set_event_loop(self.loop) self.addCleanup(asyncio.set_event_loop, None) - with self.assertWarns(DeprecationWarning) as cm: - t = asyncio.ensure_future(notmuch()) - self.assertEqual(cm.warnings[0].filename, __file__) + t = asyncio.ensure_future(notmuch()) self.assertIs(t._loop, self.loop) self.loop.run_until_complete(t) self.assertTrue(t.done()) @@ -247,10 +243,8 @@ class BaseTaskTests: a = notmuch() self.addCleanup(a.close) - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaisesRegex(RuntimeError, 'There is no current event loop'): - asyncio.ensure_future(a) - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'There is no current event loop'): + asyncio.ensure_future(a) async def test(): return asyncio.ensure_future(notmuch()) @@ -260,12 +254,10 @@ class BaseTaskTests: self.assertTrue(t.done()) self.assertEqual(t.result(), 'ok') - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 asyncio.set_event_loop(self.loop) self.addCleanup(asyncio.set_event_loop, None) - with self.assertWarns(DeprecationWarning) as cm: - t = asyncio.ensure_future(notmuch()) - self.assertEqual(cm.warnings[0].filename, __file__) + t = asyncio.ensure_future(notmuch()) self.assertIs(t._loop, self.loop) self.loop.run_until_complete(t) self.assertTrue(t.done()) @@ -1488,10 +1480,8 @@ class BaseTaskTests: self.addCleanup(a.close) futs = asyncio.as_completed([a]) - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaisesRegex(RuntimeError, 'There is no current event loop'): - list(futs) - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'no current event loop'): + list(futs) def test_as_completed_coroutine_use_running_loop(self): loop = self.new_test_loop() @@ -1507,17 +1497,14 @@ class BaseTaskTests: loop.run_until_complete(test()) def test_as_completed_coroutine_use_global_loop(self): - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 async def coro(): return 42 loop = self.new_test_loop() asyncio.set_event_loop(loop) self.addCleanup(asyncio.set_event_loop, None) - futs = asyncio.as_completed([coro()]) - with self.assertWarns(DeprecationWarning) as cm: - futs = list(futs) - self.assertEqual(cm.warnings[0].filename, __file__) + futs = list(asyncio.as_completed([coro()])) self.assertEqual(len(futs), 1) self.assertEqual(loop.run_until_complete(futs[0]), 42) @@ -1987,10 +1974,8 @@ class BaseTaskTests: inner = coro() self.addCleanup(inner.close) - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaisesRegex(RuntimeError, 'There is no current event loop'): - asyncio.shield(inner) - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'no current event loop'): + asyncio.shield(inner) def test_shield_coroutine_use_running_loop(self): async def coro(): @@ -2004,15 +1989,13 @@ class BaseTaskTests: self.assertEqual(res, 42) def test_shield_coroutine_use_global_loop(self): - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 async def coro(): return 42 asyncio.set_event_loop(self.loop) self.addCleanup(asyncio.set_event_loop, None) - with self.assertWarns(DeprecationWarning) as cm: - outer = asyncio.shield(coro()) - self.assertEqual(cm.warnings[0].filename, __file__) + outer = asyncio.shield(coro()) self.assertEqual(outer._loop, self.loop) res = self.loop.run_until_complete(outer) self.assertEqual(res, 42) @@ -2950,7 +2933,7 @@ class BaseCurrentLoopTests: self.assertIsNone(asyncio.current_task(loop=self.loop)) def test_current_task_no_running_loop_implicit(self): - with self.assertRaises(RuntimeError): + with self.assertRaisesRegex(RuntimeError, 'no running event loop'): asyncio.current_task() def test_current_task_with_implicit_loop(self): @@ -3114,10 +3097,8 @@ class FutureGatherTests(GatherTestsBase, test_utils.TestCase): return asyncio.gather(*args, **kwargs) def test_constructor_empty_sequence_without_loop(self): - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaises(RuntimeError): - asyncio.gather() - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'no current event loop'): + asyncio.gather() def test_constructor_empty_sequence_use_running_loop(self): async def gather(): @@ -3130,12 +3111,10 @@ class FutureGatherTests(GatherTestsBase, test_utils.TestCase): self.assertEqual(fut.result(), []) def test_constructor_empty_sequence_use_global_loop(self): - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 asyncio.set_event_loop(self.one_loop) self.addCleanup(asyncio.set_event_loop, None) - with self.assertWarns(DeprecationWarning) as cm: - fut = asyncio.gather() - self.assertEqual(cm.warnings[0].filename, __file__) + fut = asyncio.gather() self.assertIsInstance(fut, asyncio.Future) self.assertIs(fut._loop, self.one_loop) self._run_loop(self.one_loop) @@ -3223,10 +3202,8 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase): self.addCleanup(gen1.close) gen2 = coro() self.addCleanup(gen2.close) - with self.assertWarns(DeprecationWarning) as cm: - with self.assertRaises(RuntimeError): - asyncio.gather(gen1, gen2) - self.assertEqual(cm.warnings[0].filename, __file__) + with self.assertRaisesRegex(RuntimeError, 'no current event loop'): + asyncio.gather(gen1, gen2) def test_constructor_use_running_loop(self): async def coro(): @@ -3240,16 +3217,14 @@ class CoroutineGatherTests(GatherTestsBase, test_utils.TestCase): self.one_loop.run_until_complete(fut) def test_constructor_use_global_loop(self): - # Deprecated in 3.10 + # Deprecated in 3.10.0, undeprecated in 3.10.9 async def coro(): return 'abc' asyncio.set_event_loop(self.other_loop) self.addCleanup(asyncio.set_event_loop, None) gen1 = coro() gen2 = coro() - with self.assertWarns(DeprecationWarning) as cm: - fut = asyncio.gather(gen1, gen2) - self.assertEqual(cm.warnings[0].filename, __file__) + fut = asyncio.gather(gen1, gen2) self.assertIs(fut._loop, self.other_loop) self.other_loop.run_until_complete(fut) |