diff options
author | Antoine Pitrou <solipsis@pitrou.net> | 2009-11-06 22:41:14 (GMT) |
---|---|---|
committer | Antoine Pitrou <solipsis@pitrou.net> | 2009-11-06 22:41:14 (GMT) |
commit | 557934f5a9a23908603365e483af3434f3c98284 (patch) | |
tree | ecb14756332276c5bf294b9362e3da62fcdbedcc /Lib/test/test_threading.py | |
parent | 1766b9d10e2b0d657bbadde664aeeab5ae9b3966 (diff) | |
download | cpython-557934f5a9a23908603365e483af3434f3c98284.zip cpython-557934f5a9a23908603365e483af3434f3c98284.tar.gz cpython-557934f5a9a23908603365e483af3434f3c98284.tar.bz2 |
Merged revisions 76137 via svnmerge from
svn+ssh://pythondev@svn.python.org/python/trunk
........
r76137 | antoine.pitrou | 2009-11-06 23:34:35 +0100 (ven., 06 nov. 2009) | 4 lines
Issue #7270: Add some dedicated unit tests for multi-thread synchronization
primitives such as Lock, RLock, Condition, Event and Semaphore.
........
Diffstat (limited to 'Lib/test/test_threading.py')
-rw-r--r-- | Lib/test/test_threading.py | 52 |
1 files changed, 32 insertions, 20 deletions
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index b5b05c3..8f7c676 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -12,6 +12,8 @@ import unittest import weakref import os +from test import lock_tests + # A trivial mutable counter. class Counter(object): def __init__(self): @@ -487,22 +489,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.maxsize) - def test_joining_current_thread(self): current_thread = threading.current_thread() self.assertRaises(RuntimeError, current_thread.join); @@ -517,11 +503,37 @@ 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.support.run_unittest(ThreadTests, - ThreadJoinOnShutdown, - ThreadingExceptionTests, - ) + test.support.run_unittest(LockTests, RLockTests, EventTests, + ConditionAsRLockTests, ConditionTests, + SemaphoreTests, BoundedSemaphoreTests, + ThreadTests, + ThreadJoinOnShutdown, + ThreadingExceptionTests, + ) if __name__ == "__main__": test_main() |