(self)
| 735 | self._check_notify(cond) |
| 736 | |
| 737 | def test_timeout(self): |
| 738 | cond = self.condtype() |
| 739 | timeout = 0.5 |
| 740 | results = [] |
| 741 | def f(): |
| 742 | cond.acquire() |
| 743 | t1 = time.monotonic() |
| 744 | result = cond.wait(timeout) |
| 745 | t2 = time.monotonic() |
| 746 | cond.release() |
| 747 | results.append((t2 - t1, result)) |
| 748 | |
| 749 | N = 5 |
| 750 | with Bunch(f, N): |
| 751 | pass |
| 752 | self.assertEqual(len(results), N) |
| 753 | |
| 754 | for dt, result in results: |
| 755 | self.assertTimeout(dt, timeout) |
| 756 | # Note that conceptually (that"s the condition variable protocol) |
| 757 | # a wait() may succeed even if no one notifies us and before any |
| 758 | # timeout occurs. Spurious wakeups can occur. |
| 759 | # This makes it hard to verify the result value. |
| 760 | # In practice, this implementation has no spurious wakeups. |
| 761 | self.assertFalse(result) |
| 762 | |
| 763 | def test_waitfor(self): |
| 764 | cond = self.condtype() |
nothing calls this directly
no test coverage detected