(self)
| 1049 | await asyncio.sleep(0) |
| 1050 | |
| 1051 | async def test_acquire_hang(self): |
| 1052 | sem = asyncio.Semaphore(value=0) |
| 1053 | |
| 1054 | t1 = asyncio.create_task(sem.acquire()) |
| 1055 | t2 = asyncio.create_task(sem.acquire()) |
| 1056 | await asyncio.sleep(0) |
| 1057 | |
| 1058 | t1.cancel() |
| 1059 | sem.release() |
| 1060 | await asyncio.sleep(0) |
| 1061 | await asyncio.sleep(0) |
| 1062 | self.assertTrue(sem.locked()) |
| 1063 | self.assertTrue(t2.done()) |
| 1064 | |
| 1065 | async def test_acquire_no_hang(self): |
| 1066 |
nothing calls this directly
no test coverage detected