1""" 2Create and delete FILES_PER_THREAD temp files (via tempfile.TemporaryFile) 3in each of NUM_THREADS threads, recording the number of successes and 4failures. A failure is a bug in tempfile, and may be due to: 5 6+ Trying to create more than one tempfile with the same name. 7+ Trying to delete a tempfile that doesn't still exist. 8+ Something we've never seen before. 9 10By default, NUM_THREADS == 20 and FILES_PER_THREAD == 50. This is enough to 11create about 150 failures per run under Win98SE in 2.0, and runs pretty 12quickly. Guido reports needing to boost FILES_PER_THREAD to 500 before 13provoking a 2.0 failure under Linux. 14""" 15 16NUM_THREADS = 20 17FILES_PER_THREAD = 50 18 19import tempfile 20 21from test.test_support import start_threads, run_unittest, import_module 22threading = import_module('threading') 23import unittest 24import StringIO 25from traceback import print_exc 26 27startEvent = threading.Event() 28 29class TempFileGreedy(threading.Thread): 30 error_count = 0 31 ok_count = 0 32 33 def run(self): 34 self.errors = StringIO.StringIO() 35 startEvent.wait() 36 for i in range(FILES_PER_THREAD): 37 try: 38 f = tempfile.TemporaryFile("w+b") 39 f.close() 40 except: 41 self.error_count += 1 42 print_exc(file=self.errors) 43 else: 44 self.ok_count += 1 45 46 47class ThreadedTempFileTest(unittest.TestCase): 48 def test_main(self): 49 threads = [TempFileGreedy() for i in range(NUM_THREADS)] 50 with start_threads(threads, startEvent.set): 51 pass 52 ok = sum(t.ok_count for t in threads) 53 errors = [str(t.getName()) + str(t.errors.getvalue()) 54 for t in threads if t.error_count] 55 56 msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok, 57 '\n'.join(errors)) 58 self.assertEqual(errors, [], msg) 59 self.assertEqual(ok, NUM_THREADS * FILES_PER_THREAD) 60 61def test_main(): 62 run_unittest(ThreadedTempFileTest) 63 64if __name__ == "__main__": 65 test_main() 66