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 16import tempfile 17 18from test.support import threading_helper 19import unittest 20import io 21import threading 22from traceback import print_exc 23 24threading_helper.requires_working_threading(module=True) 25 26NUM_THREADS = 20 27FILES_PER_THREAD = 50 28 29 30startEvent = threading.Event() 31 32 33class TempFileGreedy(threading.Thread): 34 error_count = 0 35 ok_count = 0 36 37 def run(self): 38 self.errors = io.StringIO() 39 startEvent.wait() 40 for i in range(FILES_PER_THREAD): 41 try: 42 f = tempfile.TemporaryFile("w+b") 43 f.close() 44 except: 45 self.error_count += 1 46 print_exc(file=self.errors) 47 else: 48 self.ok_count += 1 49 50 51class ThreadedTempFileTest(unittest.TestCase): 52 def test_main(self): 53 threads = [TempFileGreedy() for i in range(NUM_THREADS)] 54 with threading_helper.start_threads(threads, startEvent.set): 55 pass 56 ok = sum(t.ok_count for t in threads) 57 errors = [str(t.name) + str(t.errors.getvalue()) 58 for t in threads if t.error_count] 59 60 msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok, 61 '\n'.join(errors)) 62 self.assertEqual(errors, [], msg) 63 self.assertEqual(ok, NUM_THREADS * FILES_PER_THREAD) 64 65if __name__ == "__main__": 66 unittest.main() 67