Lib/test/test_threadedtempfile.py

   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.  Run the test alone to boost either
  14 via cmdline switches:
  15
  16 -f  FILES_PER_THREAD (int)
  17 -t  NUM_THREADS (int)
  18 """
  19
  20 NUM_THREADS = 20        # change w/ -t option
  21 FILES_PER_THREAD = 50   # change w/ -f option
  22
  23 import thread # If this fails, we can't test this module
  24 import threading
  25 from test.test_support import TestFailed
  26 import StringIO
  27 from traceback import print_exc
  28 import tempfile
  29
  30 startEvent = threading.Event()
  31
  32 class TempFileGreedy(threading.Thread):
  33     error_count = 0
  34     ok_count = 0
  35
  36     def run(self):
  37         self.errors = StringIO.StringIO()
  38         startEvent.wait()
  39         for i in range(FILES_PER_THREAD):
  40             try:
  41                 f = tempfile.TemporaryFile("w+b")
  42                 f.close()
  43             except:
  44                 self.error_count += 1
  45                 print_exc(file=self.errors)
  46             else:
  47                 self.ok_count += 1
  48
  49 def test_main():
  50     threads = []
  51
  52     print "Creating"
  53     for i in range(NUM_THREADS):
  54         t = TempFileGreedy()
  55         threads.append(t)
  56         t.start()
  57
  58     print "Starting"
  59     startEvent.set()
  60
  61     print "Reaping"
  62     ok = errors = 0
  63     for t in threads:
  64         t.join()
  65         ok += t.ok_count
  66         errors += t.error_count
  67         if t.error_count:
  68             print '%s errors:\n%s' % (t.getName(), t.errors.getvalue())
  69
  70     msg = "Done: errors %d ok %d" % (errors, ok)
  71     print msg
  72     if errors:
  73         raise TestFailed(msg)
  74
  75
  76 if __name__ == "__main__":
  77     import sys, getopt
  78     opts, args = getopt.getopt(sys.argv[1:], "t:f:")
  79     for o, v in opts:
  80         if o == "-f":
  81             FILES_PER_THREAD = int(v)
  82         elif o == "-t":
  83             NUM_THREADS = int(v)
  84     test_main()