diff options
Diffstat (limited to 'Lib/test/test_capi.py')
| -rw-r--r-- | Lib/test/test_capi.py | 32 |
1 files changed, 14 insertions, 18 deletions
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 sys 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 @@ except ImportError: 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 @@ class TestPendingCalls(unittest.TestCase): #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 @@ class TestPendingCalls(unittest.TestCase): 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 @@ class TestPendingCalls(unittest.TestCase): 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 @@ class TestPendingCalls(unittest.TestCase): 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 @@ class TestPendingCalls(unittest.TestCase): @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 @@ def test_main(): 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() |