testing/legion/examples/http_example/http_test.py

   1 # Copyright 2015 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import argparse
   6 import os
   7 import sys
   8
   9 TESTING_DIR = os.path.join(
  10     os.path.dirname(os.path.abspath(__file__)), '..', '..', '..')
  11 sys.path.append(TESTING_DIR)
  12
  13 from legion import legion_test_case
  14
  15
  16 class HttpTest(legion_test_case.TestCase):
  17   """Example HTTP test case."""
  18
  19   @classmethod
  20   def GetArgs(cls):
  21     """Get command line args."""
  22     parser = argparse.ArgumentParser()
  23     parser.add_argument('--http-server')
  24     parser.add_argument('--http-client')
  25     parser.add_argument('--os', default='Ubuntu-14.04')
  26     args, _ = parser.parse_known_args()
  27     return args
  28
  29   @classmethod
  30   def CreateTask(cls, name, task_hash, os_type):
  31     """Create a new task."""
  32     #pylint: disable=unexpected-keyword-arg,no-value-for-parameter
  33     task = super(HttpTest, cls).CreateTask(
  34         name=name,
  35         isolated_hash=task_hash,
  36         dimensions={'os': os_type})
  37     task.Create()
  38     return task
  39
  40   @classmethod
  41   def setUpClass(cls):
  42     """Creates the task machines and waits until they connect."""
  43     args = cls.GetArgs()
  44     cls.http_server = cls.CreateTask(
  45         'http_server', args.http_server, args.os)
  46     cls.http_client = cls.CreateTask(
  47         'http_client', args.http_client, args.os)
  48     cls.http_server.WaitForConnection()
  49     cls.http_client.WaitForConnection()
  50
  51   def testHttpWorks(self):
  52     server_port = '8080'
  53     server_proc = None
  54     client_proc = None
  55     try:
  56       server_ip = self.http_server.rpc.GetIpAddress()
  57       server_proc = self.http_server.Process(
  58           ['python', 'http_server.py', '--port', server_port])
  59       client_proc = self.http_client.Process(
  60           ['python', 'http_client.py', '--server', server_ip,
  61            '--port', server_port])
  62       client_proc.Wait()
  63       self.assertEqual(client_proc.GetReturncode(), 0)
  64     finally:
  65       if server_proc:
  66         server_proc.Kill()
  67         server_proc.Delete()
  68       if client_proc:
  69         client_proc.Kill()
  70         client_proc.Delete()
  71
  72
  73 if __name__ == '__main__':
  74   legion_test_case.main()