test_MultiPythonThread.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import threading
import time
import unittest

import STPyV8


class TestMultiPythonThread(unittest.TestCase):
    def testMultiPythonThread(self):
        class Global:
            count = 0
            started = threading.Event()
            finished = threading.Semaphore(0)

            def sleep(self, ms):
                time.sleep(ms / 1000.0)

                self.count += 1

        g = Global()

        def run():
            with STPyV8.JSIsolate():
                with STPyV8.JSContext(g) as ctxt:
                    ctxt.eval("""
                        started.wait();

                        for (i=0; i<10; i++)
                        {
                            sleep(100);
                        }

                        finished.release();
                    """)

        threading.Thread(target = run).start()

        now = time.time()

        self.assertEqual(0, g.count)

        g.started.set()
        g.finished.acquire()

        self.assertEqual(10, g.count)

        self.assertTrue((time.time() - now) >= 1)