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  import tempfile
      17  
      18  from test.support import threading_helper
      19  import unittest
      20  import io
      21  import threading
      22  from traceback import print_exc
      23  
      24  threading_helper.requires_working_threading(module=True)
      25  
      26  NUM_THREADS = 20
      27  FILES_PER_THREAD = 50
      28  
      29  
      30  startEvent = threading.Event()
      31  
      32  
      33  class ESC[4;38;5;81mTempFileGreedy(ESC[4;38;5;149mthreadingESC[4;38;5;149m.ESC[4;38;5;149mThread):
      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  
      51  class ESC[4;38;5;81mThreadedTempFileTest(ESC[4;38;5;149munittestESC[4;38;5;149m.ESC[4;38;5;149mTestCase):
      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  
      65  if __name__ == "__main__":
      66      unittest.main()