Adding a cross-task eventing server.

testing/legion/examples/events/controller.py

+#!/usr/bin/env python
+# Copyright 2016 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.
+
+"""A simple event server test case.
+
+This test will go through all of the possible event states and ensures their
+return codes are what are expected.
+"""
+
+import argparse
+import logging
+import os
+import sys
+import time
+
+# Map the testing directory so we can import legion.legion_test_case.
+TESTING_DIR = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)),
+    '..', '..', '..', '..', 'testing')
+sys.path.append(TESTING_DIR)
+
+from legion import legion_test_case
+from legion.lib import common_lib
+
+
+class EventTestController(legion_test_case.TestCase):
+  """A simple example controller for a test."""
+
+  @classmethod
+  def CreateTestTask(cls):
+    """Create a new task."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--task-hash')
+    parser.add_argument('--os', default='Ubuntu-14.04')
+    args, _ = parser.parse_known_args()

[M-A Ruel, 2016/01/18 21:57:36]

why not parse_args()?

[Mike Meade, 2016/01/19 21:53:23]

Because there can be other flags (i.e. flags for the framework) that can also be
present. If there are more args on the command line that there are flags defined
it throws a flag not recognized error. Using parse_known_args lets different
components define and parse their own flags.

+
+    task = cls.CreateTask(
+        isolated_hash=args.task_hash,
+        dimensions={'os': args.os},
+        idle_timeout_secs=90,
+        connection_timeout_secs=90,
+        verbosity=logging.DEBUG)
+    task.Create()
+    return task
+
+  @classmethod
+  def setUpClass(cls):
+    """Creates the task machines and waits until they connect."""
+    cls.task = cls.CreateTestTask()
+    cls.task.WaitForConnection()
+
+  def testEventTest(self):
+    cmd = [
+        'python',
+        'task.py',
+        '--address', str(common_lib.MY_IP),
+        '--port', str(self.event_server.port)
+        ]
+    process = self.task.Process(cmd)
+    process.Wait()
+    retcode = process.GetReturncode()
+    if retcode != 0:
+      logging.info('STDOUT:\n%s', process.ReadStdout())
+      logging.info('STDERR:\n%s', process.ReadStderr())
+    self.assertEqual(retcode, 0)
+    # Add a success logging statement to make the logs a little easier to read.
+    logging.info('Success')
+
+
+if __name__ == '__main__':
+  legion_test_case.main()