(self)
| 299 | self.assertTrue(RGX_REPR.match(repr(ev))) |
| 300 | |
| 301 | async def test_wait(self): |
| 302 | ev = asyncio.Event() |
| 303 | self.assertFalse(ev.is_set()) |
| 304 | |
| 305 | result = [] |
| 306 | |
| 307 | async def c1(result): |
| 308 | if await ev.wait(): |
| 309 | result.append(1) |
| 310 | |
| 311 | async def c2(result): |
| 312 | if await ev.wait(): |
| 313 | result.append(2) |
| 314 | |
| 315 | async def c3(result): |
| 316 | if await ev.wait(): |
| 317 | result.append(3) |
| 318 | |
| 319 | t1 = asyncio.create_task(c1(result)) |
| 320 | t2 = asyncio.create_task(c2(result)) |
| 321 | |
| 322 | await asyncio.sleep(0) |
| 323 | self.assertEqual([], result) |
| 324 | |
| 325 | t3 = asyncio.create_task(c3(result)) |
| 326 | |
| 327 | ev.set() |
| 328 | await asyncio.sleep(0) |
| 329 | self.assertEqual([3, 1, 2], result) |
| 330 | |
| 331 | self.assertTrue(t1.done()) |
| 332 | self.assertIsNone(t1.result()) |
| 333 | self.assertTrue(t2.done()) |
| 334 | self.assertIsNone(t2.result()) |
| 335 | self.assertTrue(t3.done()) |
| 336 | self.assertIsNone(t3.result()) |
| 337 | |
| 338 | async def test_wait_on_set(self): |
| 339 | ev = asyncio.Event() |
nothing calls this directly
no test coverage detected