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()