(self)
| 1892 | event.set() |
| 1893 | |
| 1894 | def test_event(self): |
| 1895 | event = self.Event() |
| 1896 | wait = TimingWrapper(event.wait) |
| 1897 | |
| 1898 | # Removed temporarily, due to API shear, this does not |
| 1899 | # work with threading._Event objects. is_set == isSet |
| 1900 | self.assertEqual(event.is_set(), False) |
| 1901 | |
| 1902 | # Removed, threading.Event.wait() will return the value of the __flag |
| 1903 | # instead of None. API Shear with the semaphore backed mp.Event |
| 1904 | self.assertEqual(wait(0.0), False) |
| 1905 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 1906 | self.assertEqual(wait(TIMEOUT1), False) |
| 1907 | self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) |
| 1908 | |
| 1909 | event.set() |
| 1910 | |
| 1911 | # See note above on the API differences |
| 1912 | self.assertEqual(event.is_set(), True) |
| 1913 | self.assertEqual(wait(), True) |
| 1914 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 1915 | self.assertEqual(wait(TIMEOUT1), True) |
| 1916 | self.assertTimingAlmostEqual(wait.elapsed, 0.0) |
| 1917 | # self.assertEqual(event.is_set(), True) |
| 1918 | |
| 1919 | event.clear() |
| 1920 | |
| 1921 | #self.assertEqual(event.is_set(), False) |
| 1922 | |
| 1923 | p = self.Process(target=self._test_event, args=(event,)) |
| 1924 | p.daemon = True |
| 1925 | p.start() |
| 1926 | self.assertEqual(wait(), True) |
| 1927 | p.join() |
| 1928 | |
| 1929 | def test_repr(self) -> None: |
| 1930 | event = self.Event() |
nothing calls this directly
no test coverage detected