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.
  14 """
  15
  16 NUM_THREADS = 20
  17 FILES_PER_THREAD = 50
  18
  19 import thread # If this fails, we can't test this module
  20 import threading
  21 import tempfile
  22
  23 from test.test_support import threading_setup, threading_cleanup, run_unittest
  24 import unittest
  25 import StringIO
  26 from traceback import print_exc
  27
  28 startEvent = threading.Event()
  29
  30 class TempFileGreedy(threading.Thread):
  31     error_count = 0
  32     ok_count = 0
  33
  34     def run(self):
  35         self.errors = StringIO.StringIO()
  36         startEvent.wait()
  37         for i in range(FILES_PER_THREAD):
  38             try:
  39                 f = tempfile.TemporaryFile("w+b")
  40                 f.close()
  41             except:
  42                 self.error_count += 1
  43                 print_exc(file=self.errors)
  44             else:
  45                 self.ok_count += 1
  46
  47
  48 class ThreadedTempFileTest(unittest.TestCase):
  49     def test_main(self):
  50         threads = []
  51         thread_info = threading_setup()
  52
  53         for i in range(NUM_THREADS):
  54             t = TempFileGreedy()
  55             threads.append(t)
  56             t.start()
  57
  58         startEvent.set()
  59
  60         ok = 0
  61         errors = []
  62         for t in threads:
  63             t.join()
  64             ok += t.ok_count
  65             if t.error_count:
  66                 errors.append(str(t.getName()) + str(t.errors.getvalue()))
  67
  68         threading_cleanup(*thread_info)
  69
  70         msg = "Errors: errors %d ok %d\n%s" % (len(errors), ok,
  71             '\n'.join(errors))
  72         self.assertEquals(errors, [], msg)
  73         self.assertEquals(ok, NUM_THREADS * FILES_PER_THREAD)
  74
  75 def test_main():
  76     run_unittest(ThreadedTempFileTest)
  77
  78 if __name__ == "__main__":
  79     test_main()