Home | History | Annotate | Download | only in test 
      1 """
      2 Create and delete FILES_PER_THREAD temp files (via tempfile.TemporaryFile)
      3 in each of NUM_THREADS threads, recording the number of successes and
      4 failures.  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 
     10 By default, NUM_THREADS == 20 and FILES_PER_THREAD == 50.  This is enough to
     11 create about 150 failures per run under Win98SE in 2.0, and runs pretty
     12 quickly. Guido reports needing to boost FILES_PER_THREAD to 500 before
     13 provoking a 2.0 failure under Linux.
     14 """
     15 
     16 NUM_THREADS = 20
     17 FILES_PER_THREAD = 50
     18 
     19 import tempfile
     20 
     21 from test.test_support import threading_setup, threading_cleanup, run_unittest, import_module
     22 threading = import_module('threading')
     23 import unittest
     24 import StringIO
     25 from traceback import print_exc
     26 
     27 startEvent = threading.Event()
     28 
     29 class 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 
     47 class ThreadedTempFileTest(unittest.TestCase):
     48     def test_main(self):
     49         threads = []
     50         thread_info = threading_setup()
     51 
     52         for i in range(NUM_THREADS):
     53             t = TempFileGreedy()
     54             threads.append(t)
     55             t.start()
     56 
     57         startEvent.set()
     58 
     59         ok = 0
     60         errors = []
     61         for t in threads:
     62             t.join()
     63             ok += t.ok_count
     64             if t.error_count:
     65                 errors.append(str(t.getName()) + str(t.errors.getvalue()))
     66 
     67         threading_cleanup(*thread_info)
     68 
     69         msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok,
     70             '\n'.join(errors))
     71         self.assertEqual(errors, [], msg)
     72         self.assertEqual(ok, NUM_THREADS * FILES_PER_THREAD)
     73 
     74 def test_main():
     75     run_unittest(ThreadedTempFileTest)
     76 
     77 if __name__ == "__main__":
     78     test_main()
     79