(self)
| 135 | self.assertTrue(t3.result()) |
| 136 | |
| 137 | async def test_acquire_cancel(self): |
| 138 | lock = asyncio.Lock() |
| 139 | self.assertTrue(await lock.acquire()) |
| 140 | |
| 141 | task = asyncio.create_task(lock.acquire()) |
| 142 | asyncio.get_running_loop().call_soon(task.cancel) |
| 143 | with self.assertRaises(asyncio.CancelledError): |
| 144 | await task |
| 145 | self.assertFalse(lock._waiters) |
| 146 | |
| 147 | async def test_cancel_race(self): |
| 148 | # Several tasks: |
nothing calls this directly
no test coverage detected