(self)
| 362 | self.assertFalse(ev.is_set()) |
| 363 | |
| 364 | async def test_clear_with_waiters(self): |
| 365 | ev = asyncio.Event() |
| 366 | result = [] |
| 367 | |
| 368 | async def c1(result): |
| 369 | if await ev.wait(): |
| 370 | result.append(1) |
| 371 | return True |
| 372 | |
| 373 | t = asyncio.create_task(c1(result)) |
| 374 | await asyncio.sleep(0) |
| 375 | self.assertEqual([], result) |
| 376 | |
| 377 | ev.set() |
| 378 | ev.clear() |
| 379 | self.assertFalse(ev.is_set()) |
| 380 | |
| 381 | ev.set() |
| 382 | ev.set() |
| 383 | self.assertEqual(1, len(ev._waiters)) |
| 384 | |
| 385 | await asyncio.sleep(0) |
| 386 | self.assertEqual([1], result) |
| 387 | self.assertEqual(0, len(ev._waiters)) |
| 388 | |
| 389 | self.assertTrue(t.done()) |
| 390 | self.assertTrue(t.result()) |
| 391 | |
| 392 | |
| 393 | class ConditionTests(unittest.IsolatedAsyncioTestCase): |
nothing calls this directly
no test coverage detected