symbian-qemu-0.9.1-12/python-2.6.1/Lib/test/test_threadedtempfile.py
changeset 1 2fb8b9db1c86
--- /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()