PyXR

c:\python24\lib \ test \ test_threadedtempfile.py



0001 """
0002 Create and delete FILES_PER_THREAD temp files (via tempfile.TemporaryFile)
0003 in each of NUM_THREADS threads, recording the number of successes and
0004 failures.  A failure is a bug in tempfile, and may be due to:
0005 
0006 + Trying to create more than one tempfile with the same name.
0007 + Trying to delete a tempfile that doesn't still exist.
0008 + Something we've never seen before.
0009 
0010 By default, NUM_THREADS == 20 and FILES_PER_THREAD == 50.  This is enough to
0011 create about 150 failures per run under Win98SE in 2.0, and runs pretty
0012 quickly. Guido reports needing to boost FILES_PER_THREAD to 500 before
0013 provoking a 2.0 failure under Linux.  Run the test alone to boost either
0014 via cmdline switches:
0015 
0016 -f  FILES_PER_THREAD (int)
0017 -t  NUM_THREADS (int)
0018 """
0019 
0020 NUM_THREADS = 20        # change w/ -t option
0021 FILES_PER_THREAD = 50   # change w/ -f option
0022 
0023 import thread # If this fails, we can't test this module
0024 import threading
0025 from test.test_support import TestFailed
0026 import StringIO
0027 from traceback import print_exc
0028 import tempfile
0029 
0030 startEvent = threading.Event()
0031 
0032 class TempFileGreedy(threading.Thread):
0033     error_count = 0
0034     ok_count = 0
0035 
0036     def run(self):
0037         self.errors = StringIO.StringIO()
0038         startEvent.wait()
0039         for i in range(FILES_PER_THREAD):
0040             try:
0041                 f = tempfile.TemporaryFile("w+b")
0042                 f.close()
0043             except:
0044                 self.error_count += 1
0045                 print_exc(file=self.errors)
0046             else:
0047                 self.ok_count += 1
0048 
0049 def test_main():
0050     threads = []
0051 
0052     print "Creating"
0053     for i in range(NUM_THREADS):
0054         t = TempFileGreedy()
0055         threads.append(t)
0056         t.start()
0057 
0058     print "Starting"
0059     startEvent.set()
0060 
0061     print "Reaping"
0062     ok = errors = 0
0063     for t in threads:
0064         t.join()
0065         ok += t.ok_count
0066         errors += t.error_count
0067         if t.error_count:
0068             print '%s errors:\n%s' % (t.getName(), t.errors.getvalue())
0069 
0070     msg = "Done: errors %d ok %d" % (errors, ok)
0071     print msg
0072     if errors:
0073         raise TestFailed(msg)
0074 
0075 
0076 if __name__ == "__main__":
0077     import sys, getopt
0078     opts, args = getopt.getopt(sys.argv[1:], "t:f:")
0079     for o, v in opts:
0080         if o == "-f":
0081             FILES_PER_THREAD = int(v)
0082         elif o == "-t":
0083             NUM_THREADS = int(v)
0084     test_main()
0085 

Generated by PyXR 0.9.4
SourceForge.net Logo