rwlaunchpad/test/launchpad

   1 #!/usr/bin/env python3
   2
   3 import argparse
   4 import contextlib
   5 import os
   6 import signal
   7 import subprocess
   8 import sys
   9
  10 import gi
  11 gi.require_version('RwcalYang', '1.0')
  12 gi.require_version('RwCal', '1.0')
  13 gi.require_version('RwLog', '1.0')
  14
  15
  16 TEST_PARSER = "test"
  17
  18
  19 class PyTestRunner:
  20     SYS_CMD = "demos/launchpad.py -m ethsim --skip-prepare-vm -c"
  21     CLOUDSIM_CMD = "cloudsim start"
  22
  23     @property
  24     def rift_install(self):
  25         return os.getenv('RIFT_INSTALL')
  26
  27     @property
  28     def account_script(self):
  29         return os.path.join(
  30                 self.rift_install,
  31                 "usr/rift/systemtest/pytest/mission_control/test_mission_control.py")
  32
  33     @property
  34     def onboard_script(self):
  35         return os.path.join(
  36                 self.rift_install,
  37                 "usr/rift/systemtest/pytest/mission_control/pingpong_vnf/test_onboard_vnf.py")
  38
  39     @property
  40     def records_script(self):
  41         return os.path.join(
  42                 self.rift_install,
  43                 "usr/rift/systemtest/pytest/mission_control/pingpong_vnf/test_records.py")
  44
  45     def run_cmd(self, scripts=None, cal_account="mock"):
  46         scripts = scripts or [self.account_script, self.onboard_script]
  47
  48         cmd = "py.test -v "
  49
  50         # In mock-cal mode we don't need the images.
  51         if cal_account == "mock":
  52             cmd += "--{} --lp-standalone --network-service pingpong_noimg ".format(cal_account)
  53         else:
  54             cmd += "--{} --lp-standalone --network-service pingpong ".format(cal_account)
  55
  56         cmd += " ".join(scripts)
  57         subprocess.call(cmd, shell=True)
  58
  59     @contextlib.contextmanager
  60     def system_start(self, debug_mode=False, cal_account="mock"):
  61
  62
  63         os.environ['LD_PRELOAD'] = os.path.join(
  64                 self.rift_install,
  65                 "usr/lib/rift/preloads/librwxercespreload.so")
  66
  67         sys_cmd = os.path.join(self.rift_install, self.SYS_CMD)
  68         if debug_mode:
  69             sys_cmd += " --mock-cli"
  70
  71         process = subprocess.Popen(
  72             sys_cmd,
  73             shell=True,
  74             preexec_fn=os.setsid)
  75
  76         cloudsim_process = None
  77         if cal_account == "lxc":
  78             # If in LXC start the cloudsim server.
  79             cloudsim_process = subprocess.Popen(
  80                 PyTestRunner.CLOUDSIM_CMD,
  81                 shell=True,
  82                 preexec_fn=os.setsid)
  83
  84         def kill():
  85             os.killpg(process.pid, signal.SIGTERM)
  86             if cloudsim_process:
  87                 os.killpg(cloudsim_process.pid, signal.SIGTERM)
  88                 cloudsim_process.wait()
  89
  90             process.wait()
  91
  92         signal.signal(signal.SIGHUP, kill)
  93         signal.signal(signal.SIGTERM, kill)
  94
  95         yield
  96
  97         kill()
  98
  99
 100 def test_launchpad(args):
 101     pytest = PyTestRunner()
 102
 103     scripts = None
 104     if args.cal == "lxc":
 105         scripts = [pytest.account_script, pytest.onboard_script, pytest.records_script]
 106
 107     with pytest.system_start(cal_account=args.cal):
 108         pytest.run_cmd(scripts=scripts, cal_account=args.cal)
 109
 110
 111 def parse(arguments):
 112     parser = argparse.ArgumentParser(description=__doc__,
 113                                     formatter_class=argparse.RawDescriptionHelpFormatter)
 114     parser.add_argument(
 115             '--log-level', '-l',
 116             default="WARNING",
 117             type=str,
 118             choices=["INFO", "DEBUG", "WARNING", "ERROR"],
 119             help="Set log level, defaults to warning and above.")
 120
 121     subparsers = parser.add_subparsers()
 122
 123     start_parser = subparsers.add_parser(TEST_PARSER, help="Test the LP")
 124     start_parser.add_argument(
 125             '--cal', "-c",
 126             help="Run the server in the foreground. The logs are sent to console.",
 127             default="mock",
 128             choices=["lxc", "mock"])
 129     start_parser.set_defaults(which=TEST_PARSER)
 130
 131     args = parser.parse_args(arguments)
 132
 133     return args
 134
 135
 136 def main(args):
 137
 138     args = parse(args)
 139
 140     if args.which == TEST_PARSER:
 141         test_launchpad(args)
 142
 143
 144 if __name__ == "__main__":
 145     main(sys.argv[1:])