src/nitsi/test.py

   1 #!/usr/bin/python3
   2
   3 import configparser
   4 import libvirt
   5 import logging
   6 import os
   7 import time
   8
   9 from . import recipe
  10 from . import virtual_environ
  11 from . import settings
  12 from . import cmd
  13
  14 logger = logging.getLogger("nitsi.test")
  15
  16
  17 class TestException(Exception):
  18     def __init__(self, message):
  19         self.message = message
  20
  21 class Test():
  22     def __init__(self, log_path, dir=None, recipe_file=None, settings_file=None, settings=None, default_settings_file=None):
  23         # init settings var
  24         self.settings = settings
  25
  26         self.log_path = log_path
  27
  28         # Init all vars with None
  29         self.settings_file = None
  30         self.recipe_file = None
  31         self.path = None
  32
  33         self.default_settings_file = default_settings_file
  34
  35         # We need at least a path to a recipe file or a dir to a test
  36         if not dir and not recipe:
  37             raise TestException("Did not get a path to a test or to a recipe file")
  38
  39         # We cannot decide which to use when we get both
  40         if (dir and recipe_file) or (dir and settings_file):
  41             raise TestException("Get dir and path to recipe or settings file")
  42
  43         if dir:
  44             try:
  45                 if not os.path.isabs(dir):
  46                     self.path = os.path.abspath(dir)
  47             except BaseException as e:
  48                 logger.error("Could not get absolute path")
  49                 raise e
  50
  51             logger.debug("Path of this test is: {}".format(self.path))
  52
  53             self.recipe_file = "{}/recipe".format(self.path)
  54             self.settings_file = "{}/settings".format(self.path)
  55
  56
  57         # We can also go on without a settings file
  58         self.settings_file = self.check_file(self.settings_file, log_level=logging.WARNING)
  59
  60         self.recipe_file = self.check_file(self.recipe_file)
  61         if not self.recipe_file:
  62             raise TestException("No recipe file found")
  63
  64         self.default_settings_file = self.check_file(self.default_settings_file)
  65
  66         # Init logging
  67         if dir:
  68             self.log = logger.getChild(os.path.basename(self.path))
  69         # We get a recipe when we get here
  70         else:
  71             self.log = logger.getChild(os.path.basename(self.recipe_file))
  72
  73         # Parse config and settings:
  74         if self.settings_file:
  75             self.settings.set_config_values_from_file(self.settings_file, type="settings-file")
  76
  77         if self.default_settings_file:
  78             self.settings.set_config_values_from_file(self.default_settings_file, type="default-settings-file")
  79
  80         # Check settings
  81         self.settings.check_config_values()
  82
  83     # Checks the file:
  84     # is the path valid ?
  85     # returns an absolut path, when the file is valid, None when not
  86     def check_file(self, file, log_level=logging.ERROR):
  87         if file:
  88             logger.debug("File to check is: {}".format(file))
  89             if not os.path.isfile(file):
  90                 err_msg = "No such file: {}".format(file)
  91                 logger.log(log_level,err_msg)
  92                 return None
  93             if not os.path.isabs(file):
  94                file = os.path.abspath(file)
  95
  96             return file
  97         return None
  98
  99     def virtual_environ_setup_stage_1(self):
 100         self.virtual_environ = virtual_environ.VirtualEnviron(self.settings.get_config_value("virtual_environ_path"))
 101
 102         self.virtual_networks = self.virtual_environ.get_networks()
 103
 104         self.virtual_machines = self.virtual_environ.get_machines()
 105
 106     def virtual_environ_setup_stage_2(self):
 107         # built up which machines which are used in our recipe
 108         used_machines = []
 109
 110         for line in self.recipe.recipe:
 111             if not line[0] in used_machines:
 112                 used_machines.append(line[0])
 113
 114         self.log.debug("Machines used in this recipe {}".format(used_machines))
 115
 116         self.used_machine_names = used_machines
 117
 118         for machine in self.used_machine_names:
 119             if not machine in self.virtual_environ.machine_names:
 120                 raise TestException("{} is listed as machine in the recipe, but the virtual environmet does not have such a machine".format(machine))
 121
 122
 123     def virtual_environ_start(self):
 124         for name in self.virtual_environ.network_names:
 125             self.virtual_networks[name].define()
 126             self.virtual_networks[name].start()
 127
 128         for name in self.used_machine_names:
 129             self.virtual_machines[name].define()
 130             self.virtual_machines[name].create_snapshot()
 131             # We can only copy files when we know which and to which dir
 132             if self.settings.get_config_value("copy_from") and self.settings.get_config_value("copy_to"):
 133                 self.virtual_machines[name].copy_in(self.settings.get_config_value("copy_from"), self.settings.get_config_value("copy_to"))
 134             self.virtual_machines[name].start()
 135
 136         # Time to which all serial output log entries are relativ
 137         log_start_time = time.time()
 138
 139         # Number of chars of the longest machine name
 140         longest_machine_name = self.virtual_environ.longest_machine_name
 141
 142         self.log.info("Try to intialize the serial connection, connect and login on all machines")
 143         for name in self.used_machine_names:
 144             self.log.info("Try to initialize the serial connection connect and login on {}".format(name))
 145             self.virtual_machines[name].serial_init(log_file="{}/test.log".format(self.log_path),
 146                                                 log_start_time=log_start_time,
 147                                                 longest_machine_name=longest_machine_name)
 148             self.virtual_machines[name].serial_connect()
 149
 150     def load_recipe(self):
 151         self.log.info("Going to load the recipe")
 152         try:
 153             self.recipe = recipe.Recipe(self.recipe_file,
 154                 fallback_machines=self.virtual_environ.machine_names)
 155
 156             for line in self.recipe.recipe:
 157                 self.log.debug(line)
 158
 159             self.log.debug("This was the recipe")
 160         except BaseException as e:
 161             self.log.error("Failed to load recipe")
 162             raise e
 163
 164     # This functions tries to handle an error of the test (eg. when 'echo "Hello World"' failed)
 165     # in an interactive way
 166     # returns False when the test should exit right now, and True when the test should go on
 167     def interactive_error_handling(self):
 168         if not self.settings.get_config_value("interactive_error_handling"):
 169             return False
 170
 171         _cmd = cmd.CMD(intro="You are droppped into an interative debugging shell because of the previous errors",
 172             help={"exit": "Exit the test rigth now",
 173                 "continue": "Continues the test without any error handling, so do not expect that the test succeeds.",
 174                 "debug": "Disconnects from the serial console and prints the devices to manually connect to the virtual machines." \
 175                     "This is useful when you can fix th error with some manual commands. Please disconnect from the serial consoles and " \
 176                     "choose 'exit or 'continue'  when you are done"})
 177
 178         command = _cmd.get_input(valid_commands=["continue", "exit", "debug"])
 179
 180         if command == "continue":
 181             # The test should go on but we do not any debugging, so we return True
 182             return True
 183         elif command == "exit":
 184             # The test should exit right now (normal behaviour)
 185             return False
 186
 187         # If we get here we are in debugging mode
 188         # Disconnect from the serial console:
 189
 190         for name in self.used_machine_names:
 191             _cmd.print_to_cmd("Disconnect from the serial console of {}".format(name))
 192             self.virtual_machines[name].serial_disconnect()
 193
 194         # Print the serial device for each machine
 195         for name in self.used_machine_names:
 196             device  = self.virtual_machines[name].get_serial_device()
 197             _cmd.print_to_cmd("Serial device of {} is {}".format(name, device))
 198
 199         _cmd.print_to_cmd("You can now connect to all serial devices, and send custom commands to the virtual machines." \
 200             "Please type 'continue' or 'exit' when you disconnected from als serial devices and want to go on.")
 201
 202         command = _cmd.get_input(valid_commands=["continue", "exit"])
 203
 204         if command == "exit":
 205             return False
 206
 207         # We should continue whit the test
 208         # Reconnect to the serial devices
 209
 210         for name in self.used_machine_names:
 211             self.log.info("Try to reconnect to {}".format(name))
 212             self.virtual_machines[name].serial_connect()
 213
 214         return True
 215
 216     def run_recipe(self):
 217         for line in self.recipe.recipe:
 218             return_value = self.virtual_machines[line[0]].cmd(line[2])
 219             self.log.debug("Return value is: {}".format(return_value))
 220             if return_value != "0" and line[1] == "":
 221                 err_msg = "Failed to execute command '{}' on {}, return code: {}".format(line[2],line[0], return_value)
 222                 # Try to handle this error in an interactive way, if we cannot go on
 223                 # raise an exception and exit
 224                 if not self.interactive_error_handling():
 225                     raise TestException(err_msg)
 226
 227             elif return_value == "0" and line[1] == "!":
 228                 err_msg = "Succeded to execute command '{}' on {}, return code: {}".format(line[2],line[0],return_value)
 229                 self.log.error(err_msg)
 230                 # Try to handle this error in an interactive way, if we cannot go on
 231                 # raise an exception and exit
 232                 if not self.interactive_error_handling():
 233                     raise TestException(err_msg)
 234             else:
 235                 self.log.debug("Command '{}' on {} returned with: {}".format(line[2],line[0],return_value))
 236
 237     def virtual_environ_stop(self):
 238         for name in self.used_machine_names:
 239             # We just catch exception here to avoid
 240             # that we stop the cleanup process if only  one command fails
 241             try:
 242                 self.virtual_machines[name].shutdown()
 243                 self.virtual_machines[name].revert_snapshot()
 244                 self.virtual_machines[name].undefine()
 245             except BaseException as e:
 246                 self.log.exception(e)
 247
 248         for name in self.virtual_environ.network_names:
 249             try:
 250                 self.virtual_networks[name].undefine()
 251             except BaseException as e:
 252                  self.log.exception(e)
 253