GeezerVison is showing github.com — pythonnet/src/tests/test_thread.py at python3 · fkarb/pythonnet

Expand file tree

test_thread.py

Copy path

More file actions

Latest commit

History

86 lines (63 loc) · 2.75 KB

test_thread.py

Copy path

File metadata and controls

86 lines (63 loc) · 2.75 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

# ===========================================================================

# 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

from Python.Test import ThreadTest

import six

if six.PY3:

import _thread as thread

else:

import thread

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, 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()