← Back to branch summary

~ubuntu-branches/ubuntu/maverick/python3.1/maverick

« back to all changes in this revision

Viewing changes to Lib/test/test_capi.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2009-03-23 00:01:27 UTC
  • Revision ID: james.westby@ubuntu.com-20090323000127-5fstfxju4ufrhthq
Tags: upstream-3.1~a1+20090322
Import upstream version 3.1~a1+20090322

Show diffs side-by-side

added added

removed removed

Lines of Context:
Lib/test/test_capi.py
 
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
 
 
4
from __future__ import with_statement
 
5
import sys
 
6
import time
 
7
import random
 
8
import unittest
 
9
import threading
 
10
from test import support
 
11
import _testcapi
 
12
 
 
13
 
 
14
def testfunction(self):
 
15
    """some doc"""
 
16
    return self
 
17
 
 
18
class InstanceMethod:
 
19
    id = _testcapi.instancemethod(id)
 
20
    testfunction = _testcapi.instancemethod(testfunction)
 
21
 
 
22
class CAPITest(unittest.TestCase):
 
23
 
 
24
    def test_instancemethod(self):
 
25
        inst = InstanceMethod()
 
26
        self.assertEqual(id(inst), inst.id())
 
27
        self.assert_(inst.testfunction() is inst)
 
28
        self.assertEqual(inst.testfunction.__doc__, testfunction.__doc__)
 
29
        self.assertEqual(InstanceMethod.testfunction.__doc__, testfunction.__doc__)
 
30
 
 
31
        InstanceMethod.testfunction.attribute = "test"
 
32
        self.assertEqual(testfunction.attribute, "test")
 
33
        self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute", "test")
 
34
 
 
35
 
 
36
class TestPendingCalls(unittest.TestCase):
 
37
 
 
38
    def pendingcalls_submit(self, l, n):
 
39
        def callback():
 
40
            #this function can be interrupted by thread switching so let's
 
41
            #use an atomic operation
 
42
            l.append(None)
 
43
 
 
44
        for i in range(n):
 
45
            time.sleep(random.random()*0.02) #0.01 secs on average
 
46
            #try submitting callback until successful.
 
47
            #rely on regular interrupt to flush queue if we are
 
48
            #unsuccessful.
 
49
            while True:
 
50
                if _testcapi._pending_threadfunc(callback):
 
51
                    break;
 
52
 
 
53
    def pendingcalls_wait(self, l, n, context = None):
 
54
        #now, stick around until l[0] has grown to 10
 
55
        count = 0;
 
56
        while len(l) != n:
 
57
            #this busy loop is where we expect to be interrupted to
 
58
            #run our callbacks.  Note that callbacks are only run on the
 
59
            #main thread
 
60
            if False and support.verbose:
 
61
                print("(%i)"%(len(l),),)
 
62
            for i in range(1000):
 
63
                a = i*i
 
64
            if context and not context.event.is_set():
 
65
                continue
 
66
            count += 1
 
67
            self.failUnless(count < 10000,
 
68
                "timeout waiting for %i callbacks, got %i"%(n, len(l)))
 
69
        if False and support.verbose:
 
70
            print("(%i)"%(len(l),))
 
71
 
 
72
    def test_pendingcalls_threaded(self):
 
73
 
 
74
        #do every callback on a separate thread
 
75
        n = 32 #total callbacks
 
76
        threads = []
 
77
        class foo(object):pass
 
78
        context = foo()
 
79
        context.l = []
 
80
        context.n = 2 #submits per thread
 
81
        context.nThreads = n // context.n
 
82
        context.nFinished = 0
 
83
        context.lock = threading.Lock()
 
84
        context.event = threading.Event()
 
85
 
 
86
        for i in range(context.nThreads):
 
87
            t = threading.Thread(target=self.pendingcalls_thread, args = (context,))
 
88
            t.start()
 
89
            threads.append(t)
 
90
 
 
91
        self.pendingcalls_wait(context.l, n, context)
 
92
 
 
93
        for t in threads:
 
94
            t.join()
 
95
 
 
96
    def pendingcalls_thread(self, context):
 
97
        try:
 
98
            self.pendingcalls_submit(context.l, context.n)
 
99
        finally:
 
100
            with context.lock:
 
101
                context.nFinished += 1
 
102
                nFinished = context.nFinished
 
103
                if False and support.verbose:
 
104
                    print("finished threads: ", nFinished)
 
105
            if nFinished == context.nThreads:
 
106
                context.event.set()
 
107
 
 
108
    def test_pendingcalls_non_threaded(self):
 
109
        #again, just using the main thread, likely they will all be dispathced at
 
110
        #once.  It is ok to ask for too many, because we loop until we find a slot.
 
111
        #the loop can be interrupted to dispatch.
 
112
        #there are only 32 dispatch slots, so we go for twice that!
 
113
        l = []
 
114
        n = 64
 
115
        self.pendingcalls_submit(l, n)
 
116
        self.pendingcalls_wait(l, n)
 
117
 
 
118
 
 
119
def test_main():
 
120
    support.run_unittest(CAPITest)
 
121
 
 
122
    for name in dir(_testcapi):
 
123
        if name.startswith('test_'):
 
124
            test = getattr(_testcapi, name)
 
125
            if support.verbose:
 
126
                print("internal", name)
 
127
            test()
 
128
 
 
129
    # some extra thread-state tests driven via _testcapi
 
130
    def TestThreadState():
 
131
        if support.verbose:
 
132
            print("auto-thread-state")
 
133
 
 
134
        idents = []
 
135
 
 
136
        def callback():
 
137
            idents.append(_thread.get_ident())
 
138
 
 
139
        _testcapi._test_thread_state(callback)
 
140
        a = b = callback
 
141
        time.sleep(1)
 
142
        # Check our main thread is in the list exactly 3 times.
 
143
        if idents.count(_thread.get_ident()) != 3:
 
144
            raise support.TestFailed(
 
145
                        "Couldn't find main thread correctly in the list")
 
146
 
 
147
    try:
 
148
        _testcapi._test_thread_state
 
149
        have_thread_state = True
 
150
    except AttributeError:
 
151
        have_thread_state = False
 
152
 
 
153
    if have_thread_state:
 
154
        import _thread
 
155
        import time
 
156
        TestThreadState()
 
157
        import threading
 
158
        t = threading.Thread(target=TestThreadState)
 
159
        t.start()
 
160
        t.join()
 
161
 
 
162
    support.run_unittest(TestPendingCalls)
 
163
 
 
164
 
 
165
if __name__ == "__main__":
 
166
    test_main()