Adding a cross-task eventing server.

testing/legion/legion_test_case.py

 # Copyright 2015 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Adds unittest-esque functionality to Legion."""
 
 import argparse
 import logging
 import sys
 import unittest
 
 # pylint: disable=relative-import
 # Import common_lib first so we can setup the environment
 from lib import common_lib
 common_lib.SetupEnvironment()
 
+from legion.lib import event_server
 from legion.lib import task_controller
 from legion.lib import task_registration_server
 
 BANNER_WIDTH = 80
 
 
 class TestCase(unittest.TestCase):
   """Test case class with added Legion support."""
 
   _registration_server = None
@@ skipping 64 matching lines @@
     """Convenience method to create a new task."""
     task = task_controller.TaskController(
         reg_server_port=cls._registration_server.port, *args, **kwargs)
     cls._registration_server.RegisterTaskCallback(
         task.otp, task.OnConnect)
     return task
 
   @classmethod
   def _SetUpFramework(cls):
     """Perform the framework-specific setup operations."""
+    # Setup the registration server
     cls._registration_server = (
         task_registration_server.TaskRegistrationServer())
+    common_lib.OnShutdown += cls._registration_server.Shutdown
     cls._registration_server.Start()
 
+    # Setup the event server
+    cls.event_server = event_server.ThreadedServer()
+    common_lib.OnShutdown += cls.event_server.shutdown
+    cls.event_server.start()
+
   @classmethod
   def _TearDownFramework(cls):
     """Perform the framework-specific teardown operations."""
     common_lib.Shutdown()
 
   @classmethod
   def _HandleSetUpClass(cls):
     """Performs common class-level setup operations.
 
     This method performs test-wide setup such as starting the registration
@@ skipping 19 matching lines @@
     cls._LogInfoBanner('tearDownClass', 'Performs class level tear down.')
     try:
       if not setup_failed:
         cls._OriginalTearDownClassMethod()
     finally:
       cls._TearDownFramework()
 
 
 def main():
   unittest.main(verbosity=0, argv=sys.argv[:1])