|
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. Run the test alone to boost either |
|
14 via cmdline switches: |
|
15 |
|
16 -f FILES_PER_THREAD (int) |
|
17 -t NUM_THREADS (int) |
|
18 """ |
|
19 |
|
20 NUM_THREADS = 20 # change w/ -t option |
|
21 FILES_PER_THREAD = 50 # change w/ -f option |
|
22 |
|
23 import thread # If this fails, we can't test this module |
|
24 import threading |
|
25 from test.test_support import TestFailed, threading_setup, threading_cleanup |
|
26 import StringIO |
|
27 from traceback import print_exc |
|
28 import tempfile |
|
29 |
|
30 startEvent = threading.Event() |
|
31 |
|
32 class TempFileGreedy(threading.Thread): |
|
33 error_count = 0 |
|
34 ok_count = 0 |
|
35 |
|
36 def run(self): |
|
37 self.errors = StringIO.StringIO() |
|
38 startEvent.wait() |
|
39 for i in range(FILES_PER_THREAD): |
|
40 try: |
|
41 f = tempfile.TemporaryFile("w+b") |
|
42 f.close() |
|
43 except: |
|
44 self.error_count += 1 |
|
45 print_exc(file=self.errors) |
|
46 else: |
|
47 self.ok_count += 1 |
|
48 |
|
49 def test_main(): |
|
50 threads = [] |
|
51 thread_info = threading_setup() |
|
52 |
|
53 print "Creating" |
|
54 for i in range(NUM_THREADS): |
|
55 t = TempFileGreedy() |
|
56 threads.append(t) |
|
57 t.start() |
|
58 |
|
59 print "Starting" |
|
60 startEvent.set() |
|
61 |
|
62 print "Reaping" |
|
63 ok = errors = 0 |
|
64 for t in threads: |
|
65 t.join() |
|
66 ok += t.ok_count |
|
67 errors += t.error_count |
|
68 if t.error_count: |
|
69 print '%s errors:\n%s' % (t.getName(), t.errors.getvalue()) |
|
70 |
|
71 msg = "Done: errors %d ok %d" % (errors, ok) |
|
72 print msg |
|
73 if errors: |
|
74 raise TestFailed(msg) |
|
75 |
|
76 threading_cleanup(*thread_info) |
|
77 |
|
78 if __name__ == "__main__": |
|
79 import sys, getopt |
|
80 opts, args = getopt.getopt(sys.argv[1:], "t:f:") |
|
81 for o, v in opts: |
|
82 if o == "-f": |
|
83 FILES_PER_THREAD = int(v) |
|
84 elif o == "-t": |
|
85 NUM_THREADS = int(v) |
|
86 test_main() |