diff -r ffa851df0825 -r 2fb8b9db1c86 symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_threadedtempfile.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_threadedtempfile.py Fri Jul 31 15:01:17 2009 +0100 @@ -0,0 +1,79 @@ +""" +Create and delete FILES_PER_THREAD temp files (via tempfile.TemporaryFile) +in each of NUM_THREADS threads, recording the number of successes and +failures. A failure is a bug in tempfile, and may be due to: + ++ Trying to create more than one tempfile with the same name. ++ Trying to delete a tempfile that doesn't still exist. ++ Something we've never seen before. + +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. +""" + +NUM_THREADS = 20 +FILES_PER_THREAD = 50 + +import thread # If this fails, we can't test this module +import threading +import tempfile + +from test.test_support import threading_setup, threading_cleanup, run_unittest +import unittest +import StringIO +from traceback import print_exc + +startEvent = threading.Event() + +class TempFileGreedy(threading.Thread): + error_count = 0 + ok_count = 0 + + def run(self): + self.errors = StringIO.StringIO() + startEvent.wait() + for i in range(FILES_PER_THREAD): + try: + f = tempfile.TemporaryFile("w+b") + f.close() + except: + self.error_count += 1 + print_exc(file=self.errors) + else: + self.ok_count += 1 + + +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())) + + threading_cleanup(*thread_info) + + msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok, + '\n'.join(errors)) + self.assertEquals(errors, [], msg) + self.assertEquals(ok, NUM_THREADS * FILES_PER_THREAD) + +def test_main(): + run_unittest(ThreadedTempFileTest) + +if __name__ == "__main__": + test_main()