(self)
| 69 | gc.collect() |
| 70 | |
| 71 | def test_leaked_coroutine(self): |
| 72 | # This test verifies that "leaked" coroutines are shut down |
| 73 | # without triggering warnings like "task was destroyed but it |
| 74 | # is pending". If this test were to fail, it would fail |
| 75 | # because runtests.py detected unexpected output to stderr. |
| 76 | event = Event() |
| 77 | |
| 78 | async def callback(): |
| 79 | try: |
| 80 | await event.wait() |
| 81 | except asyncio.CancelledError: |
| 82 | pass |
| 83 | |
| 84 | self.io_loop.add_callback(callback) |
| 85 | self.io_loop.add_callback(self.stop) |
| 86 | self.wait() |
| 87 | |
| 88 | |
| 89 | class AsyncHTTPTestCaseTest(AsyncHTTPTestCase): |
nothing calls this directly
no test coverage detected