(self)
| 1013 | await asyncio.gather(*race_tasks) |
| 1014 | |
| 1015 | async def test_acquire_cancel(self): |
| 1016 | sem = asyncio.Semaphore() |
| 1017 | await sem.acquire() |
| 1018 | |
| 1019 | acquire = asyncio.create_task(sem.acquire()) |
| 1020 | asyncio.get_running_loop().call_soon(acquire.cancel) |
| 1021 | with self.assertRaises(asyncio.CancelledError): |
| 1022 | await acquire |
| 1023 | self.assertTrue((not sem._waiters) or |
| 1024 | all(waiter.done() for waiter in sem._waiters)) |
| 1025 | |
| 1026 | async def test_acquire_cancel_before_awoken(self): |
| 1027 | sem = asyncio.Semaphore(value=0) |
nothing calls this directly
no test coverage detected