Mercurial > cpython-lru_cache-weakref
diff Lib/test/test_threading.py @ 57129:cd360ac58cd2 legacy-trunk
Issue #7270: Add some dedicated unit tests for multi-thread synchronization
primitives such as Lock, RLock, Condition, Event and Semaphore.
| author | Antoine Pitrou <solipsis@pitrou.net> |
|---|---|
| date | Fri, 06 Nov 2009 22:34:35 +0000 |
| parents | 5f61bd3aa427 |
| children | 6f5f481d619c |
line wrap: on
line diff
--- a/Lib/test/test_threading.py Thu Nov 05 23:54:42 2009 +0000 +++ b/Lib/test/test_threading.py Fri Nov 06 22:34:35 2009 +0000 @@ -11,6 +11,8 @@ import unittest import weakref +from test import lock_tests + # A trivial mutable counter. class Counter(object): def __init__(self): @@ -482,22 +484,6 @@ 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 @@ 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, )
