# ===========================================================================
# 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()