mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-10-09 08:31:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

Issue #23799: Added test.support.start_threads() for running and cleaning up

Browse source 
multiple threads.
This commit is contained in:
Serhiy Storchaka 2015-04-01 13:01:14 +03:00
parent 8218bd4caf
commit 263dcd20a3
9 changed files with 77 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

30
Lib/test/test_io.py
View file

@ -1070,11 +1070,8 @@ class BufferedReaderTest(unittest.TestCase, CommonBufferedTests):
errors.append(e)
raise
threads = [threading.Thread(target=f) for x in range(20)]
for t in threads:
t.start()
time.sleep(0.02) # yield
for t in threads:
t.join()
with support.start_threads(threads):
time.sleep(0.02) # yield
self.assertFalse(errors,
"the following exceptions were caught: %r" % errors)
s = b''.join(results)
@ -1393,11 +1390,8 @@ class BufferedWriterTest(unittest.TestCase, CommonBufferedTests):
errors.append(e)
raise
threads = [threading.Thread(target=f) for x in range(20)]
for t in threads:
t.start()
time.sleep(0.02) # yield
for t in threads:
t.join()
with support.start_threads(threads):
time.sleep(0.02) # yield
self.assertFalse(errors,
"the following exceptions were caught: %r" % errors)
bufio.close()
@ -2691,14 +2685,10 @@ class TextIOWrapperTest(unittest.TestCase):
text = "Thread%03d\n" % n
event.wait()
f.write(text)
threads = [threading.Thread(target=lambda n=x: run(n))
threads = [threading.Thread(target=run, args=(x,))
for x in range(20)]
for t in threads:
t.start()
time.sleep(0.02)
event.set()
for t in threads:
t.join()
with support.start_threads(threads, event.set):
time.sleep(0.02)
with self.open(support.TESTFN) as f:
content = f.read()
for n in range(20):
@ -3402,11 +3392,11 @@ class SignalsTest(unittest.TestCase):
# handlers, which in this case will invoke alarm_interrupt().
signal.alarm(1)
try:
self.assertRaises(ZeroDivisionError,
wio.write, item * (support.PIPE_MAX_SIZE // len(item) + 1))
with self.assertRaises(ZeroDivisionError):
wio.write(item * (support.PIPE_MAX_SIZE // len(item) + 1))
finally:
signal.alarm(0)
t.join()
t.join()
# We got one byte, get another one and check that it isn't a
# repeat of the first one.
read_results.append(os.read(r, 1))