[Python-checkins] bpo-47104: Rewrite asyncio.to_thread tests to use IsolatedAsyncioTestCase (GH-32086)
asvetlov
webhook-mailer at python.org
Wed Mar 23 17:17:18 EDT 2022
https://github.com/python/cpython/commit/ff619c7dfe8dcb0e4c8dc655abc3acc7dc586d0d
commit: ff619c7dfe8dcb0e4c8dc655abc3acc7dc586d0d
branch: main
author: Andrew Svetlov <andrew.svetlov at gmail.com>
committer: asvetlov <andrew.svetlov at gmail.com>
date: 2022-03-23T23:17:07+02:00
summary:
bpo-47104: Rewrite asyncio.to_thread tests to use IsolatedAsyncioTestCase (GH-32086)
files:
A Misc/NEWS.d/next/Tests/2022-03-23-22-45-51.bpo-47104._esUq8.rst
M Lib/test/test_asyncio/test_threads.py
diff --git a/Lib/test/test_asyncio/test_threads.py b/Lib/test/test_asyncio/test_threads.py
index 2af322421dacf..1138a93e0f78e 100644
--- a/Lib/test/test_asyncio/test_threads.py
+++ b/Lib/test/test_asyncio/test_threads.py
@@ -5,87 +5,58 @@
from contextvars import ContextVar
from unittest import mock
-from test.test_asyncio import utils as test_utils
def tearDownModule():
asyncio.set_event_loop_policy(None)
-class ToThreadTests(test_utils.TestCase):
- def setUp(self):
- super().setUp()
- self.loop = asyncio.new_event_loop()
- asyncio.set_event_loop(self.loop)
-
- def tearDown(self):
- self.loop.run_until_complete(
- self.loop.shutdown_default_executor())
- self.loop.close()
- asyncio.set_event_loop(None)
- self.loop = None
- super().tearDown()
-
- def test_to_thread(self):
- async def main():
- return await asyncio.to_thread(sum, [40, 2])
-
- result = self.loop.run_until_complete(main())
+class ToThreadTests(unittest.IsolatedAsyncioTestCase):
+ async def test_to_thread(self):
+ result = await asyncio.to_thread(sum, [40, 2])
self.assertEqual(result, 42)
- def test_to_thread_exception(self):
+ async def test_to_thread_exception(self):
def raise_runtime():
raise RuntimeError("test")
- async def main():
- await asyncio.to_thread(raise_runtime)
-
with self.assertRaisesRegex(RuntimeError, "test"):
- self.loop.run_until_complete(main())
+ await asyncio.to_thread(raise_runtime)
- def test_to_thread_once(self):
+ async def test_to_thread_once(self):
func = mock.Mock()
- async def main():
- await asyncio.to_thread(func)
-
- self.loop.run_until_complete(main())
+ await asyncio.to_thread(func)
func.assert_called_once()
- def test_to_thread_concurrent(self):
+ async def test_to_thread_concurrent(self):
func = mock.Mock()
- async def main():
- futs = []
- for _ in range(10):
- fut = asyncio.to_thread(func)
- futs.append(fut)
- await asyncio.gather(*futs)
+ futs = []
+ for _ in range(10):
+ fut = asyncio.to_thread(func)
+ futs.append(fut)
+ await asyncio.gather(*futs)
- self.loop.run_until_complete(main())
self.assertEqual(func.call_count, 10)
- def test_to_thread_args_kwargs(self):
+ async def test_to_thread_args_kwargs(self):
# Unlike run_in_executor(), to_thread() should directly accept kwargs.
func = mock.Mock()
- async def main():
- await asyncio.to_thread(func, 'test', something=True)
+ await asyncio.to_thread(func, 'test', something=True)
- self.loop.run_until_complete(main())
func.assert_called_once_with('test', something=True)
- def test_to_thread_contextvars(self):
+ async def test_to_thread_contextvars(self):
test_ctx = ContextVar('test_ctx')
def get_ctx():
return test_ctx.get()
- async def main():
- test_ctx.set('parrot')
- return await asyncio.to_thread(get_ctx)
+ test_ctx.set('parrot')
+ result = await asyncio.to_thread(get_ctx)
- result = self.loop.run_until_complete(main())
self.assertEqual(result, 'parrot')
diff --git a/Misc/NEWS.d/next/Tests/2022-03-23-22-45-51.bpo-47104._esUq8.rst b/Misc/NEWS.d/next/Tests/2022-03-23-22-45-51.bpo-47104._esUq8.rst
new file mode 100644
index 0000000000000..1369bc227f5f2
--- /dev/null
+++ b/Misc/NEWS.d/next/Tests/2022-03-23-22-45-51.bpo-47104._esUq8.rst
@@ -0,0 +1,2 @@
+Rewrite :func:`asyncio.to_thread` tests to use
+:class:`unittest.IsolatedAsyncioTestCase`.
More information about the Python-checkins
mailing list