(self)
| 1444 | self.assertFalse(barrier.broken) |
| 1445 | |
| 1446 | async def test_draining_state(self): |
| 1447 | barrier = asyncio.Barrier(self.N) |
| 1448 | results = [] |
| 1449 | |
| 1450 | async def coro(): |
| 1451 | async with barrier: |
| 1452 | # barrier state change to filling for the last task release |
| 1453 | results.append("draining" in repr(barrier)) |
| 1454 | |
| 1455 | await self.gather_tasks(self.N, coro) |
| 1456 | |
| 1457 | self.assertEqual(len(results), self.N) |
| 1458 | self.assertEqual(results[-1], False) |
| 1459 | self.assertTrue(all(results[:self.N-1])) |
| 1460 | |
| 1461 | self.assertEqual(barrier.n_waiting, 0) |
| 1462 | self.assertFalse(barrier.broken) |
| 1463 | |
| 1464 | async def test_blocking_tasks_while_draining(self): |
| 1465 | rewait = 2 |
nothing calls this directly
no test coverage detected