(self, evt)
| 504 | self.assertFalse(evt.is_set()) |
| 505 | |
| 506 | def _check_notify(self, evt): |
| 507 | # All threads get notified |
| 508 | N = 5 |
| 509 | results1 = [] |
| 510 | results2 = [] |
| 511 | def f(): |
| 512 | results1.append(evt.wait()) |
| 513 | results2.append(evt.wait()) |
| 514 | |
| 515 | with Bunch(f, N): |
| 516 | # Threads blocked on first evt.wait() |
| 517 | wait_threads_blocked(N) |
| 518 | self.assertEqual(len(results1), 0) |
| 519 | |
| 520 | # Threads unblocked |
| 521 | evt.set() |
| 522 | |
| 523 | self.assertEqual(results1, [True] * N) |
| 524 | self.assertEqual(results2, [True] * N) |
| 525 | |
| 526 | def test_notify(self): |
| 527 | evt = self.eventtype() |
no test coverage detected