(self)
| 2096 | |
| 2097 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
| 2098 | def test_event(self): |
| 2099 | event = self.Event() |
| 2100 | wait = TimingWrapper(event.wait) |
| 2101 | |
| 2102 | # Removed temporarily, due to API shear, this does not |
| 2103 | # work with threading._Event objects. is_set == isSet |
| 2104 | self.assertEqual(event.is_set(), False) |
| 2105 | |
| 2106 | # Removed, threading.Event.wait() will return the value of the __flag |
| 2107 | # instead of None. API Shear with the semaphore backed mp.Event |
| 2108 | self.assertEqual(wait(0.0), False) |
| 2109 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 2110 | self.assertEqual(wait(TIMEOUT1), False) |
| 2111 | self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) |
| 2112 | |
| 2113 | event.set() |
| 2114 | |
| 2115 | # See note above on the API differences |
| 2116 | self.assertEqual(event.is_set(), True) |
| 2117 | self.assertEqual(wait(), True) |
| 2118 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 2119 | self.assertEqual(wait(TIMEOUT1), True) |
| 2120 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 2121 | # self.assertEqual(event.is_set(), True) |
| 2122 | |
| 2123 | event.clear() |
| 2124 | |
| 2125 | #self.assertEqual(event.is_set(), False) |
| 2126 | |
| 2127 | p = self.Process(target=self._test_event, args=(event,)) |
| 2128 | p.daemon = True |
| 2129 | p.start() |
| 2130 | self.assertEqual(wait(), True) |
| 2131 | p.join() |
| 2132 | |
| 2133 | def test_repr(self) -> None: |
| 2134 | event = self.Event() |
nothing calls this directly
no test coverage detected