home *** CD-ROM | disk | FTP | other *** search
/ Maximum CD 2011 June / maximum-cd-2011-06.iso / DiscContents / LibO_3.3.1_Win_x86_install_multi.exe / libreoffice1.cab / test_threadedtempfile.py < prev    next >
Encoding:
Python Source  |  2011-02-15  |  2.2 KB  |  80 lines
  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()
  80.