1 files changed, 183 insertions, 0 deletions
diff --git a/Tools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py b/Tools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py
new file mode 100644
index 0000000..6f04fd3
--- /dev/null
+++ b/Tools/Scripts/webkitpy/layout_tests/layout_package/message_broker_unittest.py
@@ -0,0 +1,183 @@
+# Copyright (C) 2010 Google Inc. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are
+# met:
+#
+#     * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+#     * Redistributions in binary form must reproduce the above
+# copyright notice, this list of conditions and the following disclaimer
+# in the documentation and/or other materials provided with the
+# distribution.
+#     * Neither the name of Google Inc. nor the names of its
+# contributors may be used to endorse or promote products derived from
+# this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import logging
+import Queue
+import sys
+import thread
+import threading
+import time
+import unittest
+
+from webkitpy.common import array_stream
+from webkitpy.common.system import outputcapture
+from webkitpy.tool import mocktool
+
+from webkitpy.layout_tests import run_webkit_tests
+
+import message_broker
+
+
+class TestThread(threading.Thread):
+    def __init__(self, started_queue, stopping_queue):
+        threading.Thread.__init__(self)
+        self._thread_id = None
+        self._started_queue = started_queue
+        self._stopping_queue = stopping_queue
+        self._timeout = False
+        self._timeout_queue = Queue.Queue()
+        self._exception_info = None
+
+    def id(self):
+        return self._thread_id
+
+    def getName(self):
+        return "worker-0"
+
+    def run(self):
+        self._covered_run()
+
+    def _covered_run(self):
+        # FIXME: this is a separate routine to work around a bug
+        # in coverage: see http://bitbucket.org/ned/coveragepy/issue/85.
+        self._thread_id = thread.get_ident()
+        try:
+            self._started_queue.put('')
+            msg = self._stopping_queue.get()
+            if msg == 'KeyboardInterrupt':
+                raise KeyboardInterrupt
+            elif msg == 'Exception':
+                raise ValueError()
+            elif msg == 'Timeout':
+                self._timeout = True
+                self._timeout_queue.get()
+        except:
+            self._exception_info = sys.exc_info()
+
+    def exception_info(self):
+        return self._exception_info
+
+    def next_timeout(self):
+        if self._timeout:
+            self._timeout_queue.put('done')
+            return time.time() - 10
+        return time.time()
+
+    def clear_next_timeout(self):
+        self._next_timeout = None
+
+class TestHandler(logging.Handler):
+    def __init__(self, astream):
+        logging.Handler.__init__(self)
+        self._stream = astream
+
+    def emit(self, record):
+        self._stream.write(self.format(record))
+
+
+class MultiThreadedBrokerTest(unittest.TestCase):
+    class MockTestRunner(object):
+        def __init__(self):
+            pass
+
+        def __del__(self):
+            pass
+
+        def update(self):
+            pass
+
+    def run_one_thread(self, msg):
+        runner = self.MockTestRunner()
+        port = None
+        options = mocktool.MockOptions(child_processes='1')
+        starting_queue = Queue.Queue()
+        stopping_queue = Queue.Queue()
+        broker = message_broker.MultiThreadedBroker(port, options)
+        broker._test_runner = runner
+        child_thread = TestThread(starting_queue, stopping_queue)
+        broker._workers['worker-0'] = message_broker._WorkerState('worker-0')
+        broker._workers['worker-0'].thread = child_thread
+        child_thread.start()
+        started_msg = starting_queue.get()
+        stopping_queue.put(msg)
+        return broker.run_message_loop()
+
+    def test_basic(self):
+        interrupted = self.run_one_thread('')
+        self.assertFalse(interrupted)
+
+    def test_interrupt(self):
+        self.assertRaises(KeyboardInterrupt, self.run_one_thread, 'KeyboardInterrupt')
+
+    def test_timeout(self):
+        oc = outputcapture.OutputCapture()
+        oc.capture_output()
+        interrupted = self.run_one_thread('Timeout')
+        self.assertFalse(interrupted)
+        oc.restore_output()
+
+    def test_exception(self):
+        self.assertRaises(ValueError, self.run_one_thread, 'Exception')
+
+
+class Test(unittest.TestCase):
+    def test_find_thread_stack_found(self):
+        id, stack = sys._current_frames().items()[0]
+        found_stack = message_broker._find_thread_stack(id)
+        self.assertNotEqual(found_stack, None)
+
+    def test_find_thread_stack_not_found(self):
+        found_stack = message_broker._find_thread_stack(0)
+        self.assertEqual(found_stack, None)
+
+    def test_log_wedged_worker(self):
+        oc = outputcapture.OutputCapture()
+        oc.capture_output()
+        logger = message_broker._log
+        astream = array_stream.ArrayStream()
+        handler = TestHandler(astream)
+        logger.addHandler(handler)
+
+        starting_queue = Queue.Queue()
+        stopping_queue = Queue.Queue()
+        child_thread = TestThread(starting_queue, stopping_queue)
+        child_thread.start()
+        msg = starting_queue.get()
+
+        message_broker.log_wedged_worker(child_thread.getName(),
+                                         child_thread.id())
+        stopping_queue.put('')
+        child_thread.join(timeout=1.0)
+
+        self.assertFalse(astream.empty())
+        self.assertFalse(child_thread.isAlive())
+        oc.restore_output()
+
+
+if __name__ == '__main__':
+    unittest.main()