python (3.12.0)
1 # IsolatedAsyncioTestCase based tests
2 import asyncio
3 import contextvars
4 import traceback
5 import unittest
6 from asyncio import tasks
7
8
9 def tearDownModule():
10 asyncio.set_event_loop_policy(None)
11
12
13 class ESC[4;38;5;81mFutureTests:
14
15 async def test_future_traceback(self):
16
17 async def raise_exc():
18 raise TypeError(42)
19
20 future = self.cls(raise_exc())
21
22 for _ in range(5):
23 try:
24 await future
25 except TypeError as e:
26 tb = ''.join(traceback.format_tb(e.__traceback__))
27 self.assertEqual(tb.count("await future"), 1)
28 else:
29 self.fail('TypeError was not raised')
30
31 async def test_task_exc_handler_correct_context(self):
32 # see https://github.com/python/cpython/issues/96704
33 name = contextvars.ContextVar('name', default='foo')
34 exc_handler_called = False
35
36 def exc_handler(*args):
37 self.assertEqual(name.get(), 'bar')
38 nonlocal exc_handler_called
39 exc_handler_called = True
40
41 async def task():
42 name.set('bar')
43 1/0
44
45 loop = asyncio.get_running_loop()
46 loop.set_exception_handler(exc_handler)
47 self.cls(task())
48 await asyncio.sleep(0)
49 self.assertTrue(exc_handler_called)
50
51 async def test_handle_exc_handler_correct_context(self):
52 # see https://github.com/python/cpython/issues/96704
53 name = contextvars.ContextVar('name', default='foo')
54 exc_handler_called = False
55
56 def exc_handler(*args):
57 self.assertEqual(name.get(), 'bar')
58 nonlocal exc_handler_called
59 exc_handler_called = True
60
61 def callback():
62 name.set('bar')
63 1/0
64
65 loop = asyncio.get_running_loop()
66 loop.set_exception_handler(exc_handler)
67 loop.call_soon(callback)
68 await asyncio.sleep(0)
69 self.assertTrue(exc_handler_called)
70
71 @unittest.skipUnless(hasattr(tasks, '_CTask'),
72 'requires the C _asyncio module')
73 class ESC[4;38;5;81mCFutureTests(ESC[4;38;5;149mFutureTests, ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mIsolatedAsyncioTestCase):
74 cls = tasks._CTask
75
76 class ESC[4;38;5;81mPyFutureTests(ESC[4;38;5;149mFutureTests, ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mIsolatedAsyncioTestCase):
77 cls = tasks._PyTask
78
79 class ESC[4;38;5;81mFutureReprTests(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mIsolatedAsyncioTestCase):
80
81 async def test_recursive_repr_for_pending_tasks(self):
82 # The call crashes if the guard for recursive call
83 # in base_futures:_future_repr_info is absent
84 # See Also: https://bugs.python.org/issue42183
85
86 async def func():
87 return asyncio.all_tasks()
88
89 # The repr() call should not raise RecursionError at first.
90 waiter = await asyncio.wait_for(asyncio.Task(func()),timeout=10)
91 self.assertIn('...', repr(waiter))
92
93
94 if __name__ == '__main__':
95 unittest.main()
Browse
Usage
Versions