(self)
| 1378 | self.assertFalse(barrier.broken) |
| 1379 | |
| 1380 | async def test_filling_task_by_task(self): |
| 1381 | self.N = 3 |
| 1382 | barrier = asyncio.Barrier(self.N) |
| 1383 | |
| 1384 | t1 = asyncio.create_task(barrier.wait()) |
| 1385 | await asyncio.sleep(0) |
| 1386 | self.assertEqual(barrier.n_waiting, 1) |
| 1387 | self.assertIn("filling", repr(barrier)) |
| 1388 | |
| 1389 | t2 = asyncio.create_task(barrier.wait()) |
| 1390 | await asyncio.sleep(0) |
| 1391 | self.assertEqual(barrier.n_waiting, 2) |
| 1392 | self.assertIn("filling", repr(barrier)) |
| 1393 | |
| 1394 | t3 = asyncio.create_task(barrier.wait()) |
| 1395 | await asyncio.sleep(0) |
| 1396 | |
| 1397 | await asyncio.wait([t1, t2, t3]) |
| 1398 | |
| 1399 | self.assertEqual(barrier.n_waiting, 0) |
| 1400 | self.assertFalse(barrier.broken) |
| 1401 | |
| 1402 | async def test_filling_tasks_wait_twice(self): |
| 1403 | barrier = asyncio.Barrier(self.N) |
nothing calls this directly
no test coverage detected