testing/legion/examples/hello_world/controller_test.py

   1 #!/usr/bin/env python
   2 # Copyright 2015 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """A simple host test module.
   7
   8 This module runs on the host machine and is responsible for creating 2
   9 task machines, waiting for them, and running RPC calls on them.
  10 """
  11
  12 # Map the legion directory so we can import the host controller.
  13 import sys
  14 sys.path.append('../../')
  15
  16 import argparse
  17 import logging
  18 import time
  19
  20 import test_controller
  21
  22
  23 class ExampleTestController(test_controller.TestController):
  24   """A simple example controller for a test."""
  25
  26   def __init__(self):
  27     super(ExampleTestController, self).__init__()
  28     self.task1 = None
  29     self.task2 = None
  30
  31   def CreateTask(self, isolated_hash):
  32     """Create a task object and set the proper values."""
  33     task = self.CreateNewTask(
  34         isolated_hash=isolated_hash,
  35         dimensions={'os': 'Ubuntu-14.04'},
  36         idle_timeout_secs=90, connection_timeout_secs=90,
  37         verbosity=logging.INFO,
  38         run_id=1)
  39     task.Create()
  40     return task
  41
  42   def SetUp(self):
  43     """Create the task machines and wait until they connect.
  44
  45     In this call the actual creation of the task machines is done in parallel
  46     by the system. The WaitForConnect calls are performed in series but will
  47     return as soon as the tasks connect.
  48     """
  49     parser = argparse.ArgumentParser()
  50     parser.add_argument('--task-hash')
  51     args, _ = parser.parse_known_args()
  52
  53     self.task1 = self.CreateTask(args.task_hash)
  54     self.task2 = self.CreateTask(args.task_hash)
  55     self.task1.WaitForConnection()
  56     self.task2.WaitForConnection()
  57
  58   def RunTest(self):
  59     """Main method to run the test code."""
  60     self.CallEcho(self.task1)
  61     self.CallEcho(self.task2)
  62     self.CallTaskTest(self.task1)
  63     self.CallTaskTest(self.task2)
  64
  65   def CallEcho(self, task):
  66     """Call rpc.Echo on a task."""
  67     logging.info('Calling Echo on %s', task.name)
  68     logging.info(task.rpc.Echo(task.name))
  69
  70   def CallTaskTest(self, task):
  71     """Call task_test.py name on a task."""
  72     logging.info('Calling Subprocess to run "./task_test.py %s"', task.name)
  73     proc = task.Process(['./task_test.py', task.name])
  74     proc.Wait()
  75     retcode = proc.GetReturncode()
  76     stdout = proc.ReadStdout()
  77     stderr = proc.ReadStderr()
  78     logging.info('retcode: %s, stdout: %s, stderr: %s', retcode, stdout, stderr)
  79
  80
  81 if __name__ == '__main__':
  82   ExampleTestController().RunController()