self.assertEqual(p.exitcode, 0)
@classmethod
- def _test_waitfor_timeout_f(cls, cond, state, success):
+ def _test_waitfor_timeout_f(cls, cond, state, success, sem):
+ sem.release()
with cond:
expected = 0.1
dt = time.time()
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))
+ args=(cond, state, success, sem))
p.daemon = True
p.start()
+ self.assertTrue(sem.acquire(timeout=10))
# Only increment 3 times, so state == 4 is never reached.
for i in range(3):
delta = time.time() - start
self.assertEqual(res, [])
- self.assertLess(delta, expected + 1)
- self.assertGreater(delta, expected - 1)
+ self.assertLess(delta, expected * 2)
+ self.assertGreater(delta, expected * 0.5)
b.send(None)