diff options
Diffstat (limited to 'lib/python2.7/test/test_dummy_threading.py')
-rw-r--r-- | lib/python2.7/test/test_dummy_threading.py | 63 |
1 files changed, 63 insertions, 0 deletions
diff --git a/lib/python2.7/test/test_dummy_threading.py b/lib/python2.7/test/test_dummy_threading.py new file mode 100644 index 0000000..2f59d5f --- /dev/null +++ b/lib/python2.7/test/test_dummy_threading.py @@ -0,0 +1,63 @@ +from test import test_support +import unittest +import dummy_threading as _threading +import time + +class DummyThreadingTestCase(unittest.TestCase): + + class TestThread(_threading.Thread): + + def run(self): + global running + global sema + global mutex + # Uncomment if testing another module, such as the real 'threading' + # module. + #delay = random.random() * 2 + delay = 0 + if test_support.verbose: + print 'task', self.name, 'will run for', delay, 'sec' + sema.acquire() + mutex.acquire() + running += 1 + if test_support.verbose: + print running, 'tasks are running' + mutex.release() + time.sleep(delay) + if test_support.verbose: + print 'task', self.name, 'done' + mutex.acquire() + running -= 1 + if test_support.verbose: + print self.name, 'is finished.', running, 'tasks are running' + mutex.release() + sema.release() + + def setUp(self): + self.numtasks = 10 + global sema + sema = _threading.BoundedSemaphore(value=3) + global mutex + mutex = _threading.RLock() + global running + running = 0 + self.threads = [] + + def test_tasks(self): + for i in range(self.numtasks): + t = self.TestThread(name="<thread %d>"%i) + self.threads.append(t) + t.start() + + if test_support.verbose: + print 'waiting for all tasks to complete' + for t in self.threads: + t.join() + if test_support.verbose: + print 'all tasks done' + +def test_main(): + test_support.run_unittest(DummyThreadingTestCase) + +if __name__ == '__main__': + test_main() |