diff options
author | Brett Cannon <bcannon@gmail.com> | 2008-03-18 01:50:25 (GMT) |
---|---|---|
committer | Brett Cannon <bcannon@gmail.com> | 2008-03-18 01:50:25 (GMT) |
commit | ba17cfc66c3ea90613a934704766147c7b8284a0 (patch) | |
tree | 606341b54f0d97921170681a6269bd50c8a2a374 /Lib/test/test_dummy_threading.py | |
parent | 5de250e823f03e45cee3558b699f9e8b4404c347 (diff) | |
download | cpython-ba17cfc66c3ea90613a934704766147c7b8284a0.zip cpython-ba17cfc66c3ea90613a934704766147c7b8284a0.tar.gz cpython-ba17cfc66c3ea90613a934704766147c7b8284a0.tar.bz2 |
Convert test_dummy_threading and test_dbm to unittest.
Diffstat (limited to 'Lib/test/test_dummy_threading.py')
-rw-r--r-- | Lib/test/test_dummy_threading.py | 117 |
1 files changed, 54 insertions, 63 deletions
diff --git a/Lib/test/test_dummy_threading.py b/Lib/test/test_dummy_threading.py index dcb9729..df7df3c 100644 --- a/Lib/test/test_dummy_threading.py +++ b/Lib/test/test_dummy_threading.py @@ -1,72 +1,63 @@ -# Very rudimentary test of threading module - -# Create a bunch of threads, let each do some work, wait until all are done - -from test.test_support import verbose +from test import test_support +import unittest import dummy_threading as _threading import time - -class TestThread(_threading.Thread): - - def run(self): +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.getName(), '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.getName(), 'done' + mutex.acquire() + running -= 1 + if test_support.verbose: + print self.getName(), '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 - # Uncomment if testing another module, such as the real 'threading' - # module. - #delay = random.random() * 2 - delay = 0 - if verbose: - print 'task', self.getName(), 'will run for', delay, 'sec' - sema.acquire() - mutex.acquire() - running = running + 1 - if verbose: - print running, 'tasks are running' - mutex.release() - time.sleep(delay) - if verbose: - print 'task', self.getName(), 'done' - mutex.acquire() - running = running - 1 - if verbose: - print self.getName(), 'is finished.', running, 'tasks are running' - mutex.release() - sema.release() - -def starttasks(): - for i in range(numtasks): - t = TestThread(name="<thread %d>"%i) - threads.append(t) - t.start() - + 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(): - # This takes about n/3 seconds to run (about n/3 clumps of tasks, times - # about 1 second per clump). - global numtasks - numtasks = 10 - - # no more than 3 of the 10 can run at once - global sema - sema = _threading.BoundedSemaphore(value=3) - global mutex - mutex = _threading.RLock() - global running - running = 0 - - global threads - threads = [] - - starttasks() - - if verbose: - print 'waiting for all tasks to complete' - for t in threads: - t.join() - if verbose: - print 'all tasks done' - - + test_support.run_unittest(DummyThreadingTestCase) if __name__ == '__main__': test_main() |