(self)
| 915 | self.assertTrue(sem.locked()) |
| 916 | |
| 917 | async def test_repr(self): |
| 918 | sem = asyncio.Semaphore() |
| 919 | self.assertEndsWith(repr(sem), '[unlocked, value:1]>') |
| 920 | self.assertTrue(RGX_REPR.match(repr(sem))) |
| 921 | |
| 922 | await sem.acquire() |
| 923 | self.assertEndsWith(repr(sem), '[locked]>') |
| 924 | self.assertTrue('waiters' not in repr(sem)) |
| 925 | self.assertTrue(RGX_REPR.match(repr(sem))) |
| 926 | |
| 927 | if sem._waiters is None: |
| 928 | sem._waiters = collections.deque() |
| 929 | |
| 930 | sem._waiters.append(mock.Mock()) |
| 931 | self.assertTrue('waiters:1' in repr(sem)) |
| 932 | self.assertTrue(RGX_REPR.match(repr(sem))) |
| 933 | |
| 934 | sem._waiters.append(mock.Mock()) |
| 935 | self.assertTrue('waiters:2' in repr(sem)) |
| 936 | self.assertTrue(RGX_REPR.match(repr(sem))) |
| 937 | |
| 938 | async def test_semaphore(self): |
| 939 | sem = asyncio.Semaphore() |
nothing calls this directly
no test coverage detected