Issue #850728: Add a *timeout* parameter to the `acquire()` method of
`threading.Semaphore` objects. Original patch by Torsten Landschoff.
diff --git a/Lib/test/lock_tests.py b/Lib/test/lock_tests.py
index 74db3e4..a288176 100644
--- a/Lib/test/lock_tests.py
+++ b/Lib/test/lock_tests.py
@@ -521,6 +521,19 @@
# ordered.
self.assertEqual(sorted(results), [False] * 7 + [True] * 3 )
+ def test_acquire_timeout(self):
+ sem = self.semtype(2)
+ self.assertRaises(ValueError, sem.acquire, False, timeout=1.0)
+ self.assertTrue(sem.acquire(timeout=0.005))
+ self.assertTrue(sem.acquire(timeout=0.005))
+ self.assertFalse(sem.acquire(timeout=0.005))
+ sem.release()
+ self.assertTrue(sem.acquire(timeout=0.005))
+ t = time.time()
+ self.assertFalse(sem.acquire(timeout=0.5))
+ dt = time.time() - t
+ self.assertTimeout(dt, 0.5)
+
def test_default_value(self):
# The default initial value is 1.
sem = self.semtype()