Repository navigation
Expand file tree
/
Copy pathtest_thread.py
More file actions
80 lines (58 loc) · 2.69 KB
/
Copy pathtest_thread.py
File metadata and controls
80 lines (58 loc) · 2.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# ===========================================================================
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
# ===========================================================================
import sys, os, string, unittest, types, thread
from Python.Test import ThreadTest
def dprint(msg):
# Debugging helper to trace thread-related tests.
if 0: print msg
class ThreadTests(unittest.TestCase):
"""Test CLR bridge threading and GIL handling."""
def testSimpleCallbackToPython(self):
"""Test a call to managed code that then calls back into Python."""
dprint("thread %s SimpleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString("spam")
self.assertTrue(result == "spam")
dprint("thread %s SimpleCallBack ret" % thread.get_ident())
def testDoubleCallbackToPython(self):
"""Test a call to managed code that then calls back into Python
that then calls managed code that then calls Python again."""
dprint("thread %s DoubleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString2("spam")
self.assertTrue(result == "spam")
dprint("thread %s DoubleCallBack ret" % thread.get_ident())
def testPythonThreadCallsToCLR(self):
"""Test calls by Python-spawned threads into managed code."""
# This test is very likely to hang if something is wrong ;)
import threading, thread, time
from System import String
done = []
def run_thread():
for i in range(10):
time.sleep(0.1)
dprint("thread %s %d" % (thread.get_ident(), i))
mstr = String("thread %s %d" % (thread.get_ident(), i))
pstr = mstr.ToString()
done.append(None)
dprint("thread %s %d done" % (thread.get_ident(), i))
def start_threads(count):
for i in range(count):
thread = threading.Thread(target=run_thread)
thread.start()
start_threads(5)
while len(done) < 50:
dprint(len(done))
time.sleep(0.1)
return
def test_suite():
return unittest.makeSuite(ThreadTests)
def main():
for i in range(50):
unittest.TextTestRunner().run(test_suite())
if __name__ == '__main__':
main()