Index: test_threadedtempfile.py =================================================================== --- test_threadedtempfile.py (revision 54282) +++ test_threadedtempfile.py (working copy) @@ -10,25 +10,24 @@ By default, NUM_THREADS == 20 and FILES_PER_THREAD == 50. This is enough to create about 150 failures per run under Win98SE in 2.0, and runs pretty quickly. Guido reports needing to boost FILES_PER_THREAD to 500 before -provoking a 2.0 failure under Linux. Run the test alone to boost either -via cmdline switches: - --f FILES_PER_THREAD (int) --t NUM_THREADS (int) +provoking a 2.0 failure under Linux. """ -NUM_THREADS = 20 # change w/ -t option -FILES_PER_THREAD = 50 # change w/ -f option +NUM_THREADS = 20 +FILES_PER_THREAD = 50 import thread # If this fails, we can't test this module import threading -from test.test_support import TestFailed, threading_setup, threading_cleanup +from test.test_support import threading_setup, threading_cleanup, verbose, \ + run_unittest +import unittest import StringIO from traceback import print_exc import tempfile startEvent = threading.Event() + class TempFileGreedy(threading.Thread): error_count = 0 ok_count = 0 @@ -46,41 +45,40 @@ else: self.ok_count += 1 -def test_main(): - threads = [] - thread_info = threading_setup() - print "Creating" - for i in range(NUM_THREADS): - t = TempFileGreedy() - threads.append(t) - t.start() +class ThreadedTempFileTest(unittest.TestCase): + def test_main(self): + threads = [] + thread_info = threading_setup() + + for i in range(NUM_THREADS): + t = TempFileGreedy() + threads.append(t) + t.start() + + startEvent.set() + + ok = 0 + errors = [] + for t in threads: + t.join() + ok += t.ok_count + if t.error_count: + errors.append(str(t.getName()) + str(t.errors.getvalue())) - print "Starting" - startEvent.set() + threading_cleanup(*thread_info) + + msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok, + '\n'.join(errors)) + self.assertEquals([], errors, msg) - print "Reaping" - ok = errors = 0 - for t in threads: - t.join() - ok += t.ok_count - errors += t.error_count - if t.error_count: - print '%s errors:\n%s' % (t.getName(), t.errors.getvalue()) - msg = "Done: errors %d ok %d" % (errors, ok) - print msg - if errors: - raise TestFailed(msg) +def test_main(): + run_unittest( + TempFileGreedy, + ThreadedTempFileTest, + ) - threading_cleanup(*thread_info) if __name__ == "__main__": - import sys, getopt - opts, args = getopt.getopt(sys.argv[1:], "t:f:") - for o, v in opts: - if o == "-f": - FILES_PER_THREAD = int(v) - elif o == "-t": - NUM_THREADS = int(v) test_main()