Issue #23799: Added test.test_support.start_threads() for running and
cleaning up multiple threads.
diff --git a/Lib/test/test_capi.py b/Lib/test/test_capi.py
index a2cb5c7..2029265 100644
--- a/Lib/test/test_capi.py
+++ b/Lib/test/test_capi.py
@@ -6,7 +6,7 @@
import time
import random
import unittest
-from test import test_support
+from test import test_support as support
try:
import thread
import threading
@@ -14,7 +14,7 @@
thread = None
threading = None
# Skip this test if the _testcapi module isn't available.
-_testcapi = test_support.import_module('_testcapi')
+_testcapi = support.import_module('_testcapi')
@unittest.skipUnless(threading, 'Threading required for this test.')
@@ -42,7 +42,7 @@
#this busy loop is where we expect to be interrupted to
#run our callbacks. Note that callbacks are only run on the
#main thread
- if False and test_support.verbose:
+ if False and support.verbose:
print "(%i)"%(len(l),),
for i in xrange(1000):
a = i*i
@@ -51,7 +51,7 @@
count += 1
self.assertTrue(count < 10000,
"timeout waiting for %i callbacks, got %i"%(n, len(l)))
- if False and test_support.verbose:
+ if False and support.verbose:
print "(%i)"%(len(l),)
def test_pendingcalls_threaded(self):
@@ -67,15 +67,11 @@
context.lock = threading.Lock()
context.event = threading.Event()
- for i in range(context.nThreads):
- t = threading.Thread(target=self.pendingcalls_thread, args = (context,))
- t.start()
- threads.append(t)
-
- self.pendingcalls_wait(context.l, n, context)
-
- for t in threads:
- t.join()
+ threads = [threading.Thread(target=self.pendingcalls_thread,
+ args=(context,))
+ for i in range(context.nThreads)]
+ with support.start_threads(threads):
+ self.pendingcalls_wait(context.l, n, context)
def pendingcalls_thread(self, context):
try:
@@ -84,7 +80,7 @@
with context.lock:
context.nFinished += 1
nFinished = context.nFinished
- if False and test_support.verbose:
+ if False and support.verbose:
print "finished threads: ", nFinished
if nFinished == context.nThreads:
context.event.set()
@@ -103,7 +99,7 @@
@unittest.skipUnless(threading and thread, 'Threading required for this test.')
class TestThreadState(unittest.TestCase):
- @test_support.reap_threads
+ @support.reap_threads
def test_thread_state(self):
# some extra thread-state tests driven via _testcapi
def target():
@@ -129,14 +125,14 @@
for name in dir(_testcapi):
if name.startswith('test_'):
test = getattr(_testcapi, name)
- if test_support.verbose:
+ if support.verbose:
print "internal", name
try:
test()
except _testcapi.error:
- raise test_support.TestFailed, sys.exc_info()[1]
+ raise support.TestFailed, sys.exc_info()[1]
- test_support.run_unittest(TestPendingCalls, TestThreadState)
+ support.run_unittest(TestPendingCalls, TestThreadState)
if __name__ == "__main__":
test_main()