• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Run the _testcapi module tests (tests for the Python/C API):  by defn,
2# these are all functions _testcapi exports whose name begins with 'test_'.
3
4from __future__ import with_statement
5import sys
6import time
7import random
8import unittest
9from test import test_support as support
10try:
11    import thread
12    import threading
13except ImportError:
14    thread = None
15    threading = None
16# Skip this test if the _testcapi module isn't available.
17_testcapi = support.import_module('_testcapi')
18
19class CAPITest(unittest.TestCase):
20
21    def test_buildvalue_N(self):
22        _testcapi.test_buildvalue_N()
23
24
25@unittest.skipUnless(threading, 'Threading required for this test.')
26class TestPendingCalls(unittest.TestCase):
27
28    def pendingcalls_submit(self, l, n):
29        def callback():
30            #this function can be interrupted by thread switching so let's
31            #use an atomic operation
32            l.append(None)
33
34        for i in range(n):
35            time.sleep(random.random()*0.02) #0.01 secs on average
36            #try submitting callback until successful.
37            #rely on regular interrupt to flush queue if we are
38            #unsuccessful.
39            while True:
40                if _testcapi._pending_threadfunc(callback):
41                    break;
42
43    def pendingcalls_wait(self, l, n, context = None):
44        #now, stick around until l[0] has grown to 10
45        count = 0;
46        while len(l) != n:
47            #this busy loop is where we expect to be interrupted to
48            #run our callbacks.  Note that callbacks are only run on the
49            #main thread
50            if False and support.verbose:
51                print "(%i)"%(len(l),),
52            for i in xrange(1000):
53                a = i*i
54            if context and not context.event.is_set():
55                continue
56            count += 1
57            self.assertTrue(count < 10000,
58                "timeout waiting for %i callbacks, got %i"%(n, len(l)))
59        if False and support.verbose:
60            print "(%i)"%(len(l),)
61
62    def test_pendingcalls_threaded(self):
63        #do every callback on a separate thread
64        n = 32 #total callbacks
65        threads = []
66        class foo(object):pass
67        context = foo()
68        context.l = []
69        context.n = 2 #submits per thread
70        context.nThreads = n // context.n
71        context.nFinished = 0
72        context.lock = threading.Lock()
73        context.event = threading.Event()
74
75        threads = [threading.Thread(target=self.pendingcalls_thread,
76                                    args=(context,))
77                   for i in range(context.nThreads)]
78        with support.start_threads(threads):
79            self.pendingcalls_wait(context.l, n, context)
80
81    def pendingcalls_thread(self, context):
82        try:
83            self.pendingcalls_submit(context.l, context.n)
84        finally:
85            with context.lock:
86                context.nFinished += 1
87                nFinished = context.nFinished
88                if False and support.verbose:
89                    print "finished threads: ", nFinished
90            if nFinished == context.nThreads:
91                context.event.set()
92
93    def test_pendingcalls_non_threaded(self):
94        #again, just using the main thread, likely they will all be dispatched at
95        #once.  It is ok to ask for too many, because we loop until we find a slot.
96        #the loop can be interrupted to dispatch.
97        #there are only 32 dispatch slots, so we go for twice that!
98        l = []
99        n = 64
100        self.pendingcalls_submit(l, n)
101        self.pendingcalls_wait(l, n)
102
103
104@unittest.skipUnless(threading and thread, 'Threading required for this test.')
105class TestThreadState(unittest.TestCase):
106
107    @support.reap_threads
108    def test_thread_state(self):
109        # some extra thread-state tests driven via _testcapi
110        def target():
111            idents = []
112
113            def callback():
114                idents.append(thread.get_ident())
115
116            _testcapi._test_thread_state(callback)
117            a = b = callback
118            time.sleep(1)
119            # Check our main thread is in the list exactly 3 times.
120            self.assertEqual(idents.count(thread.get_ident()), 3,
121                             "Couldn't find main thread correctly in the list")
122
123        target()
124        t = threading.Thread(target=target)
125        t.start()
126        t.join()
127
128
129def test_main():
130    for name in dir(_testcapi):
131        if name.startswith('test_'):
132            test = getattr(_testcapi, name)
133            if support.verbose:
134                print "internal", name
135            try:
136                test()
137            except _testcapi.error:
138                raise support.TestFailed, sys.exc_info()[1]
139
140    support.run_unittest(CAPITest, TestPendingCalls, TestThreadState)
141
142if __name__ == "__main__":
143    test_main()
144