[nitsi.git] / src / nitsi / test.py

#!/usr/bin/python3

import configparser
import libvirt
import logging
import os
import time

from . import recipe
from . import virtual_environ
from . import settings
from . import cmd

logger = logging.getLogger("nitsi.test")


class TestException(Exception):
    def __init__(self, message):
        self.message = message

class Test():
    def __init__(self, log_path, dir=None, recipe_file=None, settings_file=None, cmd_settings=None):
        # init settings var
        self.settings = {}

        # Set default values for the settings dict
        self.settings["name"] = ""
        self.settings["description"] = ""
        self.settings["copy_from"] = None
        self.settings["copy_to"] = None
        self.settings["virtual_environ_path"] = None
        self.settings["interactive_error_handling"] = True

        self.cmd_settings = cmd_settings
        self.log_path = log_path

        # Init all vars with None
        self.settings_file = None
        self.recipe_file = None
        self.path = None

        # We need at least a path to a recipe file or a dir to a test
        if not dir and not recipe:
            raise TestException("Did not get a path to a test or to a recipe file")

        # We cannot decide which to use when we get both
        if (dir and recipe_file) or (dir and settings_file):
            raise TestException("Get dir and path to recipe or settings file")

        if dir:
            try:
                if not os.path.isabs(dir):
                    self.path = os.path.abspath(dir)
            except BaseException as e:
                logger.error("Could not get absolute path")
                raise e

            logger.debug("Path of this test is: {}".format(self.path))

            self.recipe_file = "{}/recipe".format(self.path)
            self.settings_file = "{}/settings".format(self.path)

        if recipe_file:
            if not os.path.isabs(recipe_file):
                self.recipe_file = os.path.abspath(recipe_file)
            else:
                self.recipe_file = recipe_file

        if settings_file:
            if not os.path.isabs(settings_file):
                self.settings_file = os.path.abspath(settings_file)
            else:
                self.settings_file = settings_file

        # We can also go on without a settings file
        if self.settings_file:
            if not os.path.isfile(self.settings_file):
                logger.error("No such file: {}".format(self.settings_file))
                raise TestException("No settings file found")

        # os.path.isfile fails if self.recipe_file is None so we cannot use an and statement
        if self.recipe_file:
            if not os.path.isfile(self.recipe_file):
                logger.error("No such file: {}".format(self.recipe_file))
                raise TestException("No recipe file found")
        else:
            logger.error("No such file: {}".format(self.recipe_file))
            raise TestException("No recipe file found")


        # Init logging
        if dir:
             self.log = logger.getChild(os.path.basename(self.path))

        if recipe:
            self.log = logger.getChild(os.path.basename(self.recipe_file))

    def read_settings(self):
        if self.settings_file:
            self.log.debug("Going to read all settings from the ini file")
            try:
                self.config = configparser.ConfigParser()
                self.config.read(self.settings_file)
            except BaseException as e:
                self.log.error("Failed to parse the config")
                raise e

            self.settings["name"] = self.config.get("GENERAL","name", fallback="")
            self.settings["description"] = self.config.get("GENERAL", "description",  fallback="")
            self.settings["copy_to"] = self.config.get("GENERAL", "copy_to", fallback=None)
            self.settings["copy_from"] = self.config.get("GENERAL", "copy_from", fallback=None)
            self.settings["virtual_environ_path"] = self.config.get("VIRTUAL_ENVIRONMENT", "path", fallback=None)

            # We need to parse some settings here because they are loaded from a settings file
            if not os.path.isabs(self.settings["virtual_environ_path"]):
                self.settings["virtual_environ_path"] = os.path.normpath(os.path.dirname(
                    self.settings_file) + "/" + self.settings["virtual_environ_path"])

            # Parse copy_from setting
            if self.settings["copy_from"]:
                self.settings["copy_from"] = settings.settings_parse_copy_from(self.settings["copy_from"],
                path=os.path.dirname(self.settings_file))

        # Update all settings from the cmd
        self.settings.update(self.cmd_settings)

        if not os.path.isabs(self.settings["virtual_environ_path"]):
            self.settings["virtual_environ_path"] = os.path.abspath(self.settings["virtual_environ_path"])


        # Check if we get at least a valid a valid path to virtual environ
        if not self.settings["virtual_environ_path"]:
            self.log.error("No path for virtual environment found.")
            raise TestException("No path for virtual environment found.")

        # Print all settings for debugging purpose
        self.log.debug("Settings are:")
        for key in self.settings:
            self.log.debug("{}: {}".format(key, self.settings[key]))


    def virtual_environ_setup_stage_1(self):
        self.virtual_environ = virtual_environ.VirtualEnviron(self.settings["virtual_environ_path"])

        self.virtual_networks = self.virtual_environ.get_networks()

        self.virtual_machines = self.virtual_environ.get_machines()

    def virtual_environ_setup_stage_2(self):
        # built up which machines which are used in our recipe
        used_machines = []

        for line in self.recipe.recipe:
            if not line[0] in used_machines:
                used_machines.append(line[0])

        self.log.debug("Machines used in this recipe {}".format(used_machines))

        self.used_machine_names = used_machines

        for machine in self.used_machine_names:
            if not machine in self.virtual_environ.machine_names:
                raise TestException("{} is listed as machine in the recipe, but the virtual environmet does not have such a machine".format(machine))


    def virtual_environ_start(self):
        for name in self.virtual_environ.network_names:
            self.virtual_networks[name].define()
            self.virtual_networks[name].start()

        for name in self.used_machine_names:
            self.virtual_machines[name].define()
            self.virtual_machines[name].create_snapshot()
            # We can only copy files when we know which and to which dir
            if self.settings["copy_from"] and self.settings["copy_to"]:
                self.virtual_machines[name].copy_in(self.settings["copy_from"], self.settings["copy_to"])
            self.virtual_machines[name].start()

        # Time to which all serial output log entries are relativ
        log_start_time = time.time()

        # Number of chars of the longest machine name
        longest_machine_name = self.virtual_environ.longest_machine_name

        self.log.info("Try to intialize the serial connection, connect and login on all machines")
        for name in self.used_machine_names:
            self.log.info("Try to initialize the serial connection connect and login on {}".format(name))
            self.virtual_machines[name].serial_init(log_file="{}/test.log".format(self.log_path),
                                                log_start_time=log_start_time,
                                                longest_machine_name=longest_machine_name)
            self.virtual_machines[name].serial_connect()

    def load_recipe(self):
        self.log.info("Going to load the recipe")
        try:
            self.recipe = recipe.Recipe(self.recipe_file,
                fallback_machines=self.virtual_environ.machine_names)

            for line in self.recipe.recipe:
                self.log.debug(line)

            self.log.debug("This was the recipe")
        except BaseException as e:
            self.log.error("Failed to load recipe")
            raise e

    # This functions tries to handle an rror of the test (eg. when 'echo "Hello World"' failed)
    # in an interactive way
    # returns False when the test should exit right now, and True when the test should go on
    def interactive_error_handling(self):
        if not self.settings["interactive_error_handling"]:
            return False

        _cmd = cmd.CMD(intro="You are droppped into an interative debugging shell because of the previous errors",
            help={"exit": "Exit the test rigth now",
                "continue": "Continues the test without any error handling, so do not expect that the test succeeds.",
                "debug": "Disconnects from the serial console and prints the devices to manually connect to the virtual machines." \
                    "This is useful when you can fix th error with some manual commands. Please disconnect from the serial consoles and " \
                    "choose 'exit or 'continue'  when you are done"})

        command = _cmd.get_input(valid_commands=["continue", "exit", "debug"])

        if command == "continue":
            # The test should go on but we do not any debugging, so we return True
            return True
        elif command == "exit":
            # The test should exit right now (normal behaviour)
            return False

        # If we get here we are in debugging mode
        # Disconnect from the serial console:

        for name in self.used_machine_names:
            _cmd.print_to_cmd("Disconnect from the serial console of {}".format(name))
            self.virtual_machines[name].serial_disconnect()

        # Print the serial device for each machine
        for name in self.used_machine_names:
            device  = self.virtual_machines[name].get_serial_device()
            _cmd.print_to_cmd("Serial device of {} is {}".format(name, device))

        _cmd.print_to_cmd("You can now connect to all serial devices, and send custom commands to the virtual machines." \
            "Please type 'continue' or 'exit' when you disconnected from als serial devices and want to go on.")

        command = _cmd.get_input(valid_commands=["continue", "exit"])

        if command == "exit":
            return False

        # We should continue whit the test
        # Reconnect to the serial devices

        for name in self.used_machine_names:
            self.log.info("Try to reconnect to {}".format(name))
            self.virtual_machines[name].serial_connect()

        return True

    def run_recipe(self):
        for line in self.recipe.recipe:
            return_value = self.virtual_machines[line[0]].cmd(line[2])
            self.log.debug("Return value is: {}".format(return_value))
            if return_value != "0" and line[1] == "":
                err_msg = "Failed to execute command '{}' on {}, return code: {}".format(line[2],line[0], return_value)
                # Try to handle this error in an interactive way, if we cannot go on
                # raise an exception and exit
                if not self.interactive_error_handling():
                    raise TestException(err_msg)

            elif return_value == "0" and line[1] == "!":
                err_msg = "Succeded to execute command '{}' on {}, return code: {}".format(line[2],line[0],return_value)
                self.log.error(err_msg)
                # Try to handle this error in an interactive way, if we cannot go on
                # raise an exception and exit
                if not self.interactive_error_handling():
                    raise TestException(err_msg)
            else:
                self.log.debug("Command '{}' on {} returned with: {}".format(line[2],line[0],return_value))

    def virtual_environ_stop(self):
        for name in self.used_machine_names:
            # We just catch exception here to avoid
            # that we stop the cleanup process if only  one command fails
            try:
                self.virtual_machines[name].shutdown()
                self.virtual_machines[name].revert_snapshot()
                self.virtual_machines[name].undefine()
            except BaseException as e:
                self.log.exception(e)

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