Method test_waitfor_timeout

Lib/test/_test_multiprocessing.py:2037–2058

(self)

    @warnings_helper.ignore_fork_in_thread_deprecation_warnings()
    @unittest.skipUnless(HAS_SHAREDCTYPES, 'needs sharedctypes')
    def test_waitfor_timeout(self):
        # based on test in test/lock_tests.py
        cond = self.Condition()
        state = self.Value('i', 0)
        success = self.Value('i', False)
        sem = self.Semaphore(0)

        p = self.Process(target=self._test_waitfor_timeout_f,
                         args=(cond, state, success, sem))
        p.daemon = True
        p.start()
        self.assertTrue(sem.acquire(timeout=support.LONG_TIMEOUT))

        # Only increment 3 times, so state == 4 is never reached.
        for i in range(3):
            time.sleep(0.010)
            with cond:
                state.value += 1
                cond.notify()

        join_process(p)
        self.assertTrue(success.value)

    @classmethod
    def _test_wait_result(cls, c, pid):