(self)
| 1063 | self.assertTrue(t2.done()) |
| 1064 | |
| 1065 | async def test_acquire_no_hang(self): |
| 1066 | |
| 1067 | sem = asyncio.Semaphore(1) |
| 1068 | |
| 1069 | async def c1(): |
| 1070 | async with sem: |
| 1071 | await asyncio.sleep(0) |
| 1072 | t2.cancel() |
| 1073 | |
| 1074 | async def c2(): |
| 1075 | async with sem: |
| 1076 | self.assertFalse(True) |
| 1077 | |
| 1078 | t1 = asyncio.create_task(c1()) |
| 1079 | t2 = asyncio.create_task(c2()) |
| 1080 | |
| 1081 | r1, r2 = await asyncio.gather(t1, t2, return_exceptions=True) |
| 1082 | self.assertTrue(r1 is None) |
| 1083 | self.assertTrue(isinstance(r2, asyncio.CancelledError)) |
| 1084 | |
| 1085 | await asyncio.wait_for(sem.acquire(), timeout=1.0) |
| 1086 | |
| 1087 | def test_release_not_acquired(self): |
| 1088 | sem = asyncio.BoundedSemaphore() |
nothing calls this directly
no test coverage detected