Issue #850728: Add a *timeout* parameter to the `acquire()` method of

`threading.Semaphore` objects.  Original patch by Torsten Landschoff.

1 files changed, 13 insertions, 0 deletions

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 @@ class BaseSemaphoreTests(BaseTestCase):
         # 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()