(self)
| 1334 | self.assertIsInstance(asyncio.Barrier(self.N), asyncio.Barrier) |
| 1335 | |
| 1336 | async def test_context_manager(self): |
| 1337 | self.N = 3 |
| 1338 | barrier = asyncio.Barrier(self.N) |
| 1339 | results = [] |
| 1340 | |
| 1341 | async def coro(): |
| 1342 | async with barrier as i: |
| 1343 | results.append(i) |
| 1344 | |
| 1345 | await self.gather_tasks(self.N, coro) |
| 1346 | |
| 1347 | self.assertListEqual(sorted(results), list(range(self.N))) |
| 1348 | self.assertEqual(barrier.n_waiting, 0) |
| 1349 | self.assertFalse(barrier.broken) |
| 1350 | |
| 1351 | async def test_filling_one_task(self): |
| 1352 | barrier = asyncio.Barrier(1) |
nothing calls this directly
no test coverage detected