(self)
| 403 | self.assertEqual(2, runner.run(get_context()).get(cvar)) |
| 404 | |
| 405 | def test_recursive_run(self): |
| 406 | async def g(): |
| 407 | pass |
| 408 | |
| 409 | async def f(): |
| 410 | runner.run(g()) |
| 411 | |
| 412 | with asyncio.Runner() as runner: |
| 413 | with self.assertWarnsRegex( |
| 414 | RuntimeWarning, |
| 415 | "coroutine .+ was never awaited", |
| 416 | ): |
| 417 | with self.assertRaisesRegex( |
| 418 | RuntimeError, |
| 419 | re.escape( |
| 420 | "Runner.run() cannot be called from a running event loop" |
| 421 | ), |
| 422 | ): |
| 423 | runner.run(f()) |
| 424 | |
| 425 | def test_interrupt_call_soon(self): |
| 426 | # The only case when task is not suspended by waiting a future |
nothing calls this directly
no test coverage detected