-
Notifications
You must be signed in to change notification settings - Fork 774
Expand file tree
/
Copy pathtest_thread.py
More file actions
58 lines (42 loc) · 1.68 KB
/
test_thread.py
File metadata and controls
58 lines (42 loc) · 1.68 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
# -*- coding: utf-8 -*-
"""Test CLR bridge threading and GIL handling."""
import threading
import time
import _thread as thread
from .utils import dprint
def test_simple_callback_to_python():
"""Test a call to managed code that then calls back into Python."""
from Python.Test import ThreadTest
dprint("thread %s SimpleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString("spam")
assert result == "spam"
dprint("thread %s SimpleCallBack ret" % thread.get_ident())
def test_double_callback_to_python():
"""Test a call to managed code that then calls back into Python
that then calls managed code that then calls Python again."""
from Python.Test import ThreadTest
dprint("thread %s DoubleCallBack" % thread.get_ident())
result = ThreadTest.CallEchoString2("spam")
assert result == "spam"
dprint("thread %s DoubleCallBack ret" % thread.get_ident())
def test_python_thread_calls_to_clr():
"""Test calls by Python-spawned threads into managed code."""
# This test is very likely to hang if something is wrong ;)
import System
done = []
def run_thread():
for i in range(10):
time.sleep(0.1)
dprint("thread %s %d" % (thread.get_ident(), i))
mstr = System.String("thread %s %d" % (thread.get_ident(), i))
dprint(mstr.ToString())
done.append(None)
dprint("thread %s %d done" % (thread.get_ident(), i))
def start_threads(count):
for _ 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)