(self)
| 952 | self.assertRaises(ValueError, asyncio.Semaphore, -1) |
| 953 | |
| 954 | async def test_acquire(self): |
| 955 | sem = asyncio.Semaphore(3) |
| 956 | result = [] |
| 957 | |
| 958 | self.assertTrue(await sem.acquire()) |
| 959 | self.assertTrue(await sem.acquire()) |
| 960 | self.assertFalse(sem.locked()) |
| 961 | |
| 962 | async def c1(result): |
| 963 | await sem.acquire() |
| 964 | result.append(1) |
| 965 | return True |
| 966 | |
| 967 | async def c2(result): |
| 968 | await sem.acquire() |
| 969 | result.append(2) |
| 970 | return True |
| 971 | |
| 972 | async def c3(result): |
| 973 | await sem.acquire() |
| 974 | result.append(3) |
| 975 | return True |
| 976 | |
| 977 | async def c4(result): |
| 978 | await sem.acquire() |
| 979 | result.append(4) |
| 980 | return True |
| 981 | |
| 982 | t1 = asyncio.create_task(c1(result)) |
| 983 | t2 = asyncio.create_task(c2(result)) |
| 984 | t3 = asyncio.create_task(c3(result)) |
| 985 | |
| 986 | await asyncio.sleep(0) |
| 987 | self.assertEqual([1], result) |
| 988 | self.assertTrue(sem.locked()) |
| 989 | self.assertEqual(2, len(sem._waiters)) |
| 990 | self.assertEqual(0, sem._value) |
| 991 | |
| 992 | t4 = asyncio.create_task(c4(result)) |
| 993 | |
| 994 | sem.release() |
| 995 | sem.release() |
| 996 | self.assertEqual(0, sem._value) |
| 997 | |
| 998 | await asyncio.sleep(0) |
| 999 | self.assertEqual(0, sem._value) |
| 1000 | self.assertEqual(3, len(result)) |
| 1001 | self.assertTrue(sem.locked()) |
| 1002 | self.assertEqual(1, len(sem._waiters)) |
| 1003 | self.assertEqual(0, sem._value) |
| 1004 | |
| 1005 | self.assertTrue(t1.done()) |
| 1006 | self.assertTrue(t1.result()) |
| 1007 | race_tasks = [t2, t3, t4] |
| 1008 | done_tasks = [t for t in race_tasks if t.done() and t.result()] |
| 1009 | self.assertEqual(2, len(done_tasks)) |
| 1010 | |
| 1011 | # cleanup locked semaphore |
nothing calls this directly
no test coverage detected