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