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()