# Copyright 2013-2020 Intranet AG and contributors
#
# avocado-i2n is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# avocado-i2n is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with avocado-i2n. If not, see <http://www.gnu.org/licenses/>.
"""
Parser for command line Cartesian parameters.
SUMMARY
------------------------------------------------------
Copyright: Intra2net AG
INTERFACE
------------------------------------------------------
"""
import sys
import os
import re
from typing import Any, Callable
import logging
from avocado.core.settings import settings
from virttest import env_process
from virttest.utils_env import Env
from avocado_vt.test import VirtTest
from virttest.utils_params import Params
from . import params_parser as param
from .cartgraph import graph
from .states import setup as ss
log = logging.getLogger("avocado.job." + __name__)
[docs]
def params_from_cmd(config: Params) -> None:
"""
Produce Cartesian parameters from a command line.
Take care of paths/utilities for all host controls.
:param config: command line arguments
:raises: :py:class:`ValueError` if a command line selected vm is not available
from the configuration and thus supported or internal tests are
restricted from the command line
.. todo:: Any dynamically created config keys here are usually entire data
structures like dictionaries and lists and only used internally during
the run which makes them unfit for displaying to the user and putting
in a namespace scope like the officially registered plugin settings.
Let's wait to see if the multi-suite support in avocado would establish
some standards for doing this first. Until then, the user won't directly
interact with these keys anyway.
"""
suite_path = settings.as_dict().get("i2n.common.suite_path", ".")
sys.path.insert(1, os.path.join(suite_path, "utils"))
# validate typed vm names and possible vm specific restrictions
available_vms = param.all_objects("vms")
available_restrictions = param.all_restrictions()
# defaults usage vs command line overriding
use_tests_default = True
with_nontrivial_restrictions = False
use_vms_default = {vm_name: True for vm_name in available_vms}
with_selected_vms = list(available_vms)
# the run string includes only pure parameters
param_dict = {}
# the tests string includes the test restrictions while the vm strings include the ones for the vm variants
tests_str, nets_str, vm_strs = "", "", {vm: "" for vm in available_vms}
# main tokenizing loop
for cmd_param in config["params"]:
re_param = re.match(r"(\w+)=(.*)", cmd_param)
if re_param is None:
raise ValueError(
f"Found malformed parameter on the command line '{cmd_param}' - "
f"must be of the form <key>=<val>"
)
key, value = re_param.group(1, 2)
if key == "only" or key == "no":
# detect if this is the primary restriction to escape defaults
for variant in re.split(r",|\.|\.\.", value):
if variant in available_restrictions:
use_tests_default = False
# else this is an auxiliary restriction
else:
with_nontrivial_restrictions = True
# main test restriction part
tests_str += "%s %s\n" % (key, value)
elif key.startswith("only_") or key.startswith("no_"):
if re.match("(only|no)_nets", key):
nets_str = (
"%s %s\n" % (key.replace("_nets", ""), value) if value else ""
)
# TODO: unify nets_str with vm_strs treatment across the Cartesian graph interface
param_dict["nets"] = " ".join(
param.all_suffixes_by_restriction(nets_str)
)
else:
for vm_name in available_vms:
if re.match(f"(only|no)_{vm_name}", key):
# escape defaults for this vm and use the command line
use_vms_default[vm_name] = False
# main vm restriction part
vm_str = (
"%s %s\n" % (key.replace(f"_{vm_name}", ""), value)
if value
else ""
)
vm_strs[vm_name] += vm_str
break
else:
raise ValueError(
f"Invalid object restriction {key} (no such object)"
)
# NOTE: comma in a parameter sense implies the same as space in config file
elif key == "vms":
# NOTE: no restrictions of the required vms are allowed during tests since
# these are specified by each test (allowed only for manual setup steps)
with_selected_vms[:] = value.split(",")
for vm_name in with_selected_vms:
if vm_name not in available_vms:
raise ValueError(
"The vm '%s' is not among the supported vms: "
"%s" % (vm_name, ", ".join(available_vms))
)
elif key == "nets":
if nets_str != "":
raise ValueError(
f"Cannot specify explicit net suffixes {value} together with "
f"a nets restriction, currently also specified '{nets_str.rstrip()}'"
)
value = value.replace(",", " ")
param_dict[key] = value
else:
# NOTE: comma on the command line is space in a config file
value = value.replace(",", " ")
param_dict[key] = value
config["param_dict"] = param_dict
log.debug("Parsed param dict '%s'", param_dict)
# get minimal configurations and parse defaults if no command line arguments
config["vms_params"], config["vm_strs"] = full_vm_params_and_strs(
param_dict, vm_strs, use_vms_default
)
config["vms_params"]["vms"] = " ".join(with_selected_vms)
config["available_vms"] = vm_strs.copy()
for vm_name in available_vms:
# the keys of vm strings must be equivalent to the selected vms
if vm_name not in with_selected_vms:
del config["vm_strs"][vm_name]
config["tests_params"], config["tests_str"] = full_tests_params_and_str(
param_dict, tests_str, use_tests_default
)
config["available_restrictions"] = available_restrictions
# control against invoking only runnable tests and empty Cartesian products
control_config = param.Reparsable()
control_config.parse_next_batch(
base_file="sets.cfg",
ovrwrt_file=param.tests_ovrwrt_file(),
ovrwrt_str=config["tests_str"],
ovrwrt_dict=config["param_dict"],
)
control_parser = control_config.get_parser()
if with_nontrivial_restrictions:
log.info(
"%s tests with nontrivial restriction %s",
len(list(control_parser.get_dicts())),
config["tests_str"],
)
# prefix for all tests of the current run making it possible to perform multiple runs in one command
config["prefix"] = ""
# log into files for each major level the way it was done for autotest
config["job.run.store_logging_stream"] = ["avocado.core:DEBUG"]
# dump parsed and traversed graph at each test loading and running step
graph.set_graph_logging_level(
level=config["tests_params"].get_numeric("cartgraph_verbose_level", 20)
)
# set default off and on state backends
from .states import lvm, qcow2, lxc, btrfs, ramfile, pool, vmnet
ss.BACKENDS = {
"qcow2": qcow2.QCOW2Backend,
"qcow2ext": qcow2.QCOW2ExtBackend,
"lvm": lvm.LVMBackend,
"lxc": lxc.LXCBackend,
"btrfs": btrfs.BtrfsBackend,
"qcow2vt": qcow2.QCOW2VTBackend,
"ramfile": ramfile.RamfileBackend,
"vmnet": vmnet.VMNetBackend,
}
ramfile.RamfileBackend.image_state_backend = qcow2.QCOW2ExtBackend
# attach environment processing hooks
env_process_hooks()
[docs]
def full_vm_params_and_strs(
param_dict: dict[str, str] | None,
vm_strs: dict[str, str],
use_vms_default: dict[str, bool],
) -> tuple[Params, dict[str, str]]:
"""
Add default vm parameters and strings for missing command line such.
:param param_dict: runtime parameters used for extra customization
:param vm_strs: command line vm-specific names and variant restrictions
:param use_vms_default: whether to use default variant restriction for a
particular vm
:returns: complete vm parameters and strings
:raises: :py:class:`ValueError` if no command line or default variant
restriction could be found for some vm
"""
vms_config = param.Reparsable()
vms_config.parse_next_batch(
base_file="guest-base.cfg",
ovrwrt_file=param.vms_ovrwrt_file(),
ovrwrt_dict=param_dict,
)
vms_params = vms_config.get_params()
for vm_name in param.all_objects("vms"):
if use_vms_default[vm_name]:
default = vms_params.get("default_only_%s" % vm_name)
vm_strs[vm_name] += "only %s\n" % default if default else ""
log.debug("Parsed vm strings '%s'", vm_strs)
return vms_params, vm_strs
[docs]
def full_tests_params_and_str(
param_dict: dict[str, str] | None, tests_str: str, use_tests_default: bool
) -> tuple[Params, str]:
"""
Add default tests parameters and string for missing command line such.
:param param_dict: runtime parameters used for extra customization
:param tests_str: command line variant restrictions
:param use_tests_default: whether to use default primary restriction
:returns: complete tests parameters and string
:raises: :py:class:`ValueError` if the default primary restriction could is
not valid (among the available ones)
"""
tests_config = param.Reparsable()
tests_config.parse_next_batch(
base_file="groups-base.cfg",
ovrwrt_file=param.tests_ovrwrt_file(),
ovrwrt_dict=param_dict,
)
tests_params = tests_config.get_params()
if use_tests_default:
default = tests_params.get("default_only", "all")
available_restrictions = param.all_restrictions()
if default not in available_restrictions:
raise ValueError(
"Invalid primary restriction 'only=%s'! It has to be one "
"of %s" % (default, ", ".join(available_restrictions))
)
tests_str += "only %s\n" % default
log.debug("Parsed tests string '%s'", tests_str)
return tests_params, tests_str
[docs]
def env_process_hooks() -> None:
"""
Add env processing hooks to handle needed customization steps.
These steps include on/off state get/set operations, vmnet networking,
and instance attachment to environment.
"""
def on_state(fn: Callable[[Any], Any]) -> Any:
def wrapper(test: VirtTest, params: Params, env: Env) -> Any:
params["skip_types"] = "nets/vms/images nets"
fn(params, env)
del params["skip_types"]
return wrapper
def off_state(fn: Callable[[Any], Any]) -> Any:
def wrapper(test: VirtTest, params: Params, env: Env) -> Any:
params["skip_types"] = "nets/vms"
fn(params, env)
del params["skip_types"]
return wrapper
env_process.preprocess_vm_off_hook = off_state(ss.get_states)
env_process.preprocess_vm_on_hook = on_state(ss.get_states)
env_process.postprocess_vm_on_hook = on_state(ss.set_states)
env_process.postprocess_vm_off_hook = off_state(ss.set_states)