mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 08:59:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 11

Merged revisions 76137 via svnmerge from

Browse source 
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.
........
This commit is contained in:
Antoine Pitrou 2009-11-06 22:41:14 +00:00
parent 1766b9d10e
commit 557934f5a9
4 changed files with 574 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

52
Lib/test/test_threading.py
View file

@ -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()