Convert test_dummy_threading and test_dbm to unittest.
diff --git a/Lib/test/test_dbm.py b/Lib/test/test_dbm.py
index 6b57efb..562b14f 100755
--- a/Lib/test/test_dbm.py
+++ b/Lib/test/test_dbm.py
@@ -1,60 +1,40 @@
-#! /usr/bin/env python
-"""Test script for the dbm module
- Roger E. Masse
-"""
+from test import test_support
+import unittest
import os
+import random
import dbm
from dbm import error
-from test.test_support import verbose, verify, TestSkipped, TESTFN
-# make filename unique to allow multiple concurrent tests
-# and to minimize the likelihood of a problem from an old file
-filename = TESTFN
+class DbmTestCase(unittest.TestCase):
-def cleanup():
- for suffix in ['', '.pag', '.dir', '.db']:
- try:
- os.unlink(filename + suffix)
- except OSError, (errno, strerror):
- # if we can't delete the file because of permissions,
- # nothing will work, so skip the test
- if errno == 1:
- raise TestSkipped, 'unable to remove: ' + filename + suffix
+ def setUp(self):
+ self.filename = test_support.TESTFN
+ self.d = dbm.open(self.filename, 'c')
+ self.d.close()
-def test_keys():
- d = dbm.open(filename, 'c')
- verify(d.keys() == [])
- d['a'] = 'b'
- d['12345678910'] = '019237410982340912840198242'
- d.keys()
- if d.has_key('a'):
- if verbose:
- print 'Test dbm keys: ', d.keys()
+ def tearDown(self):
+ for suffix in ['', '.pag', '.dir', '.db']:
+ test_support.unlink(self.filename + suffix)
- d.close()
+ def test_keys(self):
+ self.d = dbm.open(self.filename, 'c')
+ self.assert_(self.d.keys() == [])
+ self.d['a'] = 'b'
+ self.d['12345678910'] = '019237410982340912840198242'
+ self.d.keys()
+ self.assert_(self.d.has_key('a'))
+ self.d.close()
-def test_modes():
- d = dbm.open(filename, 'r')
- d.close()
- d = dbm.open(filename, 'rw')
- d.close()
- d = dbm.open(filename, 'w')
- d.close()
- d = dbm.open(filename, 'n')
- d.close()
+ def test_modes(self):
+ for mode in ['r', 'rw', 'w', 'n']:
+ try:
+ self.d = dbm.open(self.filename, mode)
+ self.d.close()
+ except dbm.error:
+ self.fail()
def test_main():
- cleanup()
- try:
- test_keys()
- test_modes()
- except:
- cleanup()
- raise
-
- cleanup()
-
-
+ test_support.run_unittest(DbmTestCase)
if __name__ == '__main__':
test_main()
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 DummyThreadingTestCase(unittest.TestCase):
-class TestThread(_threading.Thread):
+ class TestThread(_threading.Thread):
- def run(self):
+ 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()
+ running = 0
+ self.threads = []
-def starttasks():
- for i in range(numtasks):
- t = TestThread(name="<thread %d>"%i)
- threads.append(t)
- t.start()
+ 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()