diff options
Diffstat (limited to 'Lib/test/test_threading.py')
-rw-r--r-- | Lib/test/test_threading.py | 46 |
1 files changed, 29 insertions, 17 deletions
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index 656aee6..c38eda7 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -11,6 +11,8 @@ import time import unittest import weakref +from test import lock_tests + # A trivial mutable counter. class Counter(object): def __init__(self): @@ -482,22 +484,6 @@ class ThreadingExceptionTests(BaseTestCase): thread.start() self.assertRaises(RuntimeError, thread.start) - def test_releasing_unacquired_rlock(self): - rlock = threading.RLock() - self.assertRaises(RuntimeError, rlock.release) - - def test_waiting_on_unacquired_condition(self): - cond = threading.Condition() - self.assertRaises(RuntimeError, cond.wait) - - def test_notify_on_unacquired_condition(self): - cond = threading.Condition() - self.assertRaises(RuntimeError, cond.notify) - - def test_semaphore_with_negative_value(self): - self.assertRaises(ValueError, threading.Semaphore, value = -1) - self.assertRaises(ValueError, threading.Semaphore, value = -sys.maxint) - def test_joining_current_thread(self): current_thread = threading.current_thread() self.assertRaises(RuntimeError, current_thread.join); @@ -512,8 +498,34 @@ class ThreadingExceptionTests(BaseTestCase): self.assertRaises(RuntimeError, setattr, thread, "daemon", True) +class LockTests(lock_tests.LockTests): + locktype = staticmethod(threading.Lock) + +class RLockTests(lock_tests.RLockTests): + locktype = staticmethod(threading.RLock) + +class EventTests(lock_tests.EventTests): + eventtype = staticmethod(threading.Event) + +class ConditionAsRLockTests(lock_tests.RLockTests): + # An Condition uses an RLock by default and exports its API. + locktype = staticmethod(threading.Condition) + +class ConditionTests(lock_tests.ConditionTests): + condtype = staticmethod(threading.Condition) + +class SemaphoreTests(lock_tests.SemaphoreTests): + semtype = staticmethod(threading.Semaphore) + +class BoundedSemaphoreTests(lock_tests.BoundedSemaphoreTests): + semtype = staticmethod(threading.BoundedSemaphore) + + def test_main(): - test.test_support.run_unittest(ThreadTests, + test.test_support.run_unittest(LockTests, RLockTests, EventTests, + ConditionAsRLockTests, ConditionTests, + SemaphoreTests, BoundedSemaphoreTests, + ThreadTests, ThreadJoinOnShutdown, ThreadingExceptionTests, ) |