--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+import importlib
+import inspect
+import json
+import os
+import pkgutil
+import shutil
+from abc import ABC, abstractmethod
+from itertools import chain
+from typing import Set
+
+from tests.helpers import (
+ get_param,
+ get_environment_pair,
+ prop_iterator,
+ get_output_dir,
+ is_base_module,
+)
+from tests.parametrizers import parametrize_heat_templates
+from tests.structures import NeutronPortProcessor, Heat
+from tests.test_environment_file_parameters import get_preload_excluded_parameters
+from tests.utils import nested_dict
+from tests.utils.vm_types import get_vm_type_for_nova_server
+
+
+# This is only used to fake out parametrizers
+class DummyMetafunc:
+ def __init__(self, config):
+ self.inputs = {}
+ self.config = config
+
+ def parametrize(self, name, file_list):
+ self.inputs[name] = file_list
+
+
+def get_heat_templates(config):
+ """
+ Returns the Heat template paths discovered by the pytest parameterizers
+ :param config: pytest config
+ :return: list of heat template paths
+ """
+ meta = DummyMetafunc(config)
+ parametrize_heat_templates(meta)
+ heat_templates = meta.inputs.get("heat_templates", [])
+ if isinstance(heat_templates, list) and len(heat_templates) > 0:
+ heat_templates = heat_templates[0]
+ else:
+ return
+ return heat_templates
+
+
+def get_json_template(template_dir, template_name):
+ template_name = template_name + ".json"
+ with open(os.path.join(template_dir, template_name)) as f:
+ return json.loads(f.read())
+
+
+def remove(sequence, exclude, key=None):
+ """
+ Remove a copy of sequence that items occur in exclude.
+
+ :param sequence: sequence of objects
+ :param exclude: objects to excluded (must support ``in`` check)
+ :param key: optional function to extract key from item in sequence
+ :return: list of items not in the excluded
+ """
+ key_func = key if key else lambda x: x
+ result = (s for s in sequence if key_func(s) not in exclude)
+ return set(result) if isinstance(sequence, Set) else list(result)
+
+
+def get_or_create_template(template_dir, key, value, sequence, template_name):
+ """
+ Search a sequence of dicts where a given key matches value. If
+ found, then it returns that item. If not, then it loads the
+ template identified by template_name, adds it ot the sequence, and
+ returns the template
+ """
+ for item in sequence:
+ if item[key] == value:
+ return item
+ new_template = get_json_template(template_dir, template_name)
+ sequence.append(new_template)
+ return new_template
+
+
+def replace(param):
+ """
+ Optionally used by the preload generator to wrap items in the preload
+ that need to be replaced by end users
+ :param param: p
+ """
+ return "VALUE FOR: {}".format(param) if param else ""
+
+
+class AbstractPreloadGenerator(ABC):
+ """
+ All preload generators must inherit from this class and implement the
+ abstract methods.
+
+ Preload generators are automatically discovered at runtime via a plugin
+ architecture. The system path is scanned looking for modules with the name
+ preload_*, then all non-abstract classes that inherit from AbstractPreloadGenerator
+ are registered as preload plugins
+
+ Attributes:
+ :param vnf: Instance of Vnf that contains the preload data
+ :param base_output_dir: Base directory to house the preloads. All preloads
+ must be written to a subdirectory under this directory
+ """
+
+ def __init__(self, vnf, base_output_dir):
+ self.vnf = vnf
+ self.base_output_dir = base_output_dir
+ os.makedirs(self.output_dir, exist_ok=True)
+
+ @classmethod
+ @abstractmethod
+ def format_name(cls):
+ """
+ String name to identify the format (ex: VN-API, GR-API)
+ """
+ raise NotImplementedError()
+
+ @classmethod
+ @abstractmethod
+ def output_sub_dir(cls):
+ """
+ String sub-directory name that will appear under ``base_output_dir``
+ """
+ raise NotImplementedError()
+
+ @classmethod
+ @abstractmethod
+ def supports_output_passing(cls):
+ """
+ Some preload methods allow automatically mapping output parameters in the
+ base module to the input parameter of other modules. This means these
+ that the incremental modules do not need these base module outputs in their
+ preloads.
+
+ At this time, VNF-API does not support output parameter passing, but
+ GR-API does.
+
+ If this is true, then the generator will call Vnf#filter_output_params
+ after the preload module for the base module has been created
+ """
+ raise NotImplementedError()
+
+ @abstractmethod
+ def generate_module(self, module):
+ """
+ Create the preloads and write them to ``self.output_dir``. This
+ method is responsible for generating the content of the preload and
+ writing the file to disk.
+ """
+ raise NotImplementedError()
+
+ @property
+ def output_dir(self):
+ return os.path.join(self.base_output_dir, self.output_sub_dir())
+
+ def generate(self):
+ # handle the base module first
+ print("\nGenerating {} preloads".format(self.format_name()))
+ self.generate_module(self.vnf.base_module)
+ print("... generated template for {}".format(self.vnf.base_module))
+ if self.supports_output_passing():
+ self.vnf.filter_base_outputs()
+ for mod in self.vnf.incremental_modules:
+ self.generate_module(mod)
+ print("... generated for {}".format(mod))
+
+
+class FilterBaseOutputs(ABC):
+ """
+ Invoked to remove parameters in an object that appear in the base module.
+ Base output parameters can be passed to incremental modules
+ so they do not need to be defined in a preload. This method can be
+ invoked on a module to pre-filter the parameters before a preload is
+ created.
+
+ The method should remove the parameters that exist in the base module from
+ both itself and any sub-objects.
+ """
+
+ @abstractmethod
+ def filter_output_params(self, base_outputs):
+ raise NotImplementedError()
+
+
+class IpParam:
+ def __init__(self, ip_addr_param, port):
+ self.param = ip_addr_param or ""
+ self.port = port
+
+ @property
+ def ip_version(self):
+ return 6 if "_v6_" in self.param else 4
+
+ def __hash__(self):
+ return hash(self.param)
+
+ def __eq__(self, other):
+ return hash(self) == hash(other)
+
+ def __str__(self):
+ return "{}(v{})".format(self.param, self.ip_version)
+
+ def __repr(self):
+ return str(self)
+
+
+class Network(FilterBaseOutputs):
+ def __init__(self, role, name_param):
+ self.network_role = role
+ self.name_param = name_param
+ self.subnet_params = set()
+
+ def filter_output_params(self, base_outputs):
+ self.subnet_params = remove(self.subnet_params, base_outputs)
+
+ def __hash__(self):
+ return hash(self.network_role)
+
+ def __eq__(self, other):
+ return hash(self) == hash(other)
+
+
+class Port(FilterBaseOutputs):
+ def __init__(self, vm, network):
+ self.vm = vm
+ self.network = network
+ self.fixed_ips = []
+ self.floating_ips = []
+ self.uses_dhcp = True
+
+ def add_ips(self, props):
+ props = props.get("properties") or props
+ for fixed_ip in props.get("fixed_ips") or []:
+ if not isinstance(fixed_ip, dict):
+ continue
+ ip_address = get_param(fixed_ip.get("ip_address"))
+ subnet = get_param(fixed_ip.get("subnet") or fixed_ip.get("subnet_id"))
+ if ip_address:
+ self.uses_dhcp = False
+ self.fixed_ips.append(IpParam(ip_address, self))
+ if subnet:
+ self.network.subnet_params.add(subnet)
+ for ip in prop_iterator(props, "allowed_address_pairs", "ip_address"):
+ self.uses_dhcp = False
+ param = get_param(ip) if ip else ""
+ if param:
+ self.floating_ips.append(IpParam(param, self))
+
+ def filter_output_params(self, base_outputs):
+ self.fixed_ips = remove(self.fixed_ips, base_outputs, key=lambda ip: ip.param)
+ self.floating_ips = remove(
+ self.floating_ips, base_outputs, key=lambda ip: ip.param
+ )
+
+
+class VirtualMachineType(FilterBaseOutputs):
+ def __init__(self, vm_type, vnf_module):
+ self.vm_type = vm_type
+ self.names = []
+ self.ports = []
+ self.vm_count = 0
+ self.vnf_module = vnf_module
+
+ def filter_output_params(self, base_outputs):
+ self.names = remove(self.names, base_outputs)
+ for port in self.ports:
+ port.filter_output_params(base_outputs)
+
+ @property
+ def networks(self):
+ return {port.network for port in self.ports}
+
+ @property
+ def floating_ips(self):
+ for port in self.ports:
+ for ip in port.floating_ips:
+ yield ip
+
+ @property
+ def fixed_ips(self):
+ for port in self.ports:
+ for ip in port.fixed_ips:
+ yield ip
+
+ def update_ports(self, network, props):
+ port = self.get_or_create_port(network)
+ port.add_ips(props)
+
+ def get_or_create_port(self, network):
+ for port in self.ports:
+ if port.network == network:
+ return port
+ port = Port(self, network)
+ self.ports.append(port)
+ return port
+
+
+class Vnf:
+ def __init__(self, templates):
+ self.modules = [VnfModule(t, self) for t in templates]
+ self.uses_contrail = self._uses_contrail()
+ self.base_module = next(
+ (mod for mod in self.modules if mod.is_base_module), None
+ )
+ self.incremental_modules = [m for m in self.modules if not m.is_base_module]
+
+ def _uses_contrail(self):
+ for mod in self.modules:
+ resources = mod.heat.get_all_resources()
+ types = (r.get("type", "") for r in resources.values())
+ if any(t.startswith("OS::ContrailV2") for t in types):
+ return True
+ return False
+
+ @property
+ def base_output_params(self):
+ return self.base_module.heat.outputs
+
+ def filter_base_outputs(self):
+ non_base_modules = (m for m in self.modules if not m.is_base_module)
+ for mod in non_base_modules:
+ mod.filter_output_params(self.base_output_params)
+
+
+def yield_by_count(sequence):
+ """
+ Iterates through sequence and yields each item according to its __count__
+ attribute. If an item has a __count__ of it will be returned 3 times
+ before advancing to the next item in the sequence.
+
+ :param sequence: sequence of dicts (must contain __count__)
+ :returns: generator of tuple key, value pairs
+ """
+ for key, value in sequence.items():
+ for i in range(value["__count__"]):
+ yield (key, value)
+
+
+def env_path(heat_path):
+ """
+ Create the path to the env file for the give heat path.
+ :param heat_path: path to heat file
+ :return: path to env file (assumes it is present and named correctly)
+ """
+ base_path = os.path.splitext(heat_path)[0]
+ return "{}.env".format(base_path)
+
+
+class VnfModule(FilterBaseOutputs):
+ def __init__(self, template_file, vnf):
+ self.vnf = vnf
+ self.vnf_name = os.path.splitext(os.path.basename(template_file))[0]
+ self.template_file = template_file
+ self.heat = Heat(filepath=template_file, envpath=env_path(template_file))
+ env_pair = get_environment_pair(self.template_file)
+ env_yaml = env_pair.get("eyml") if env_pair else {}
+ self.parameters = env_yaml.get("parameters") or {}
+ self.networks = []
+ self.virtual_machine_types = self._create_vm_types()
+ self._add_networks()
+ self.outputs_filtered = False
+
+ def filter_output_params(self, base_outputs):
+ for vm in self.virtual_machine_types:
+ vm.filter_output_params(base_outputs)
+ for network in self.networks:
+ network.filter_output_params(base_outputs)
+ self.parameters = {
+ k: v for k, v in self.parameters.items() if k not in base_outputs
+ }
+ self.networks = [
+ network
+ for network in self.networks
+ if network.name_param not in base_outputs or network.subnet_params
+ ]
+ self.outputs_filtered = True
+
+ def _create_vm_types(self):
+ servers = self.heat.get_resource_by_type("OS::Nova::Server", all_resources=True)
+ vm_types = {}
+ for _, props in yield_by_count(servers):
+ vm_type = get_vm_type_for_nova_server(props)
+ vm = vm_types.setdefault(vm_type, VirtualMachineType(vm_type, self))
+ vm.vm_count += 1
+ name = nested_dict.get(props, "properties", "name", default={})
+ vm_name = get_param(name) if name else ""
+ vm.names.append(vm_name)
+ return list(vm_types.values())
+
+ def _add_networks(self):
+ ports = self.heat.get_resource_by_type("OS::Neutron::Port", all_resources=True)
+ for rid, props in yield_by_count(ports):
+ resource_type, port_match = NeutronPortProcessor.get_rid_match_tuple(rid)
+ if resource_type != "external":
+ continue
+ network_role = port_match.group("network_role")
+ vm = self._get_vm_type(port_match.group("vm_type"))
+ network = self._get_network(network_role, props)
+ vm.update_ports(network, props)
+
+ @property
+ def is_base_module(self):
+ return is_base_module(self.template_file)
+
+ @property
+ def availability_zones(self):
+ """Returns a list of all availability zone parameters found in the template"""
+ return sorted(
+ p for p in self.heat.parameters if p.startswith("availability_zone")
+ )
+
+ @property
+ def preload_parameters(self):
+ """
+ Subset of parameters from the env file that can be overridden in
+ tag values. Per VNF Heat Guidelines, specific parameters such as
+ flavor, image, etc. must not be overridden so they are excluded.
+
+ :return: dict of parameters suitable for the preload
+ """
+ excluded = get_preload_excluded_parameters(self.template_file)
+ return {k: v for k, v in self.parameters.items() if k not in excluded}
+
+ def _get_vm_type(self, vm_type):
+ for vm in self.virtual_machine_types:
+ if vm_type.lower() == vm.vm_type.lower():
+ return vm
+ raise RuntimeError("Encountered unknown VM type: {}".format(vm_type))
+
+ def _get_network(self, network_role, props):
+ network_prop = nested_dict.get(props, "properties", "network") or {}
+ name_param = get_param(network_prop) if network_prop else ""
+ for network in self.networks:
+ if network.network_role.lower() == network_role.lower():
+ return network
+ new_network = Network(network_role, name_param)
+ self.networks.append(new_network)
+ return new_network
+
+ def __str__(self):
+ return "VNF Module ({})".format(os.path.basename(self.template_file))
+
+ def __repr__(self):
+ return str(self)
+
+ def __hash__(self):
+ return hash(self.vnf_name)
+
+ def __eq__(self, other):
+ return hash(self) == hash(other)
+
+
+def create_preloads(config, exitstatus):
+ """
+ Create preloads in every format that can be discovered by get_generator_plugins
+ """
+ if config.getoption("self_test"):
+ return
+ print("+===================================================================+")
+ print("| Preload Template Generation |")
+ print("+===================================================================+")
+
+ preload_dir = os.path.join(get_output_dir(config), "preloads")
+ if os.path.exists(preload_dir):
+ shutil.rmtree(preload_dir)
+ heat_templates = get_heat_templates(config)
+ vnf = None
+ for gen_class in get_generator_plugins():
+ vnf = Vnf(heat_templates)
+ generator = gen_class(vnf, preload_dir)
+ generator.generate()
+ if vnf and vnf.uses_contrail:
+ print(
+ "\nWARNING: Preload template generation does not support Contrail\n"
+ "at this time, but Contrail resources were detected. The preload \n"
+ "template may be incomplete."
+ )
+ if exitstatus != 0:
+ print(
+ "\nWARNING: Heat violations detected. Preload templates may be\n"
+ "incomplete."
+ )
+
+
+def is_preload_generator(class_):
+ """
+ Returns True if the class is an implementation of AbstractPreloadGenerator
+ """
+ return (
+ inspect.isclass(class_)
+ and not inspect.isabstract(class_)
+ and issubclass(class_, AbstractPreloadGenerator)
+ )
+
+
+def get_generator_plugins():
+ """
+ Scan the system path for modules that are preload plugins and discover
+ and return the classes that implement AbstractPreloadGenerator in those
+ modules
+ """
+ preload_plugins = (
+ importlib.import_module(name)
+ for finder, name, ispkg in pkgutil.iter_modules()
+ if name.startswith("preload_")
+ )
+ members = chain.from_iterable(
+ inspect.getmembers(mod, is_preload_generator) for mod in preload_plugins
+ )
+ return [m[1] for m in members]
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+from .grapi_generator import GrApiPreloadGenerator
+
+__all__ = ["GrApiPreloadGenerator"]
--- /dev/null
+{
+ "input": {
+ "request-information": {
+ "request-id": "robot12",
+ "order-version": "1",
+ "notification-url": "openecomp.org",
+ "order-number": "1",
+ "request-action": "PreloadVfModuleRequest"
+ },
+ "sdnc-request-header": {
+ "svc-request-id": "robot12",
+ "svc-notification-url": "http://openecomp.org:8080/adapters/rest/SDNCNotify",
+ "svc-action": "reserve"
+ },
+ "preload-vf-module-topology-information": {
+ "vnf-topology-identifier-structure": {
+ "vnf-name": "",
+ "vnf-type": ""
+ },
+ "vnf-resource-assignments": {
+ "availability-zones": {
+ "availability-zone": []
+ },
+ "vnf-networks": {
+ "vnf-network": []
+ }
+ },
+ "vf-module-topology": {
+ "vf-module-assignments": {
+ "vms": {
+ "vm": []
+ }
+ },
+ "vf-module-topology-identifier": {
+ "vf-module-type": "",
+ "vf-module-name": ""
+ },
+ "vf-module-parameters": {
+ "param": []
+ }
+ }
+ }
+ }
+}
--- /dev/null
+{
+ "name": "",
+ "value": ""
+}
--- /dev/null
+{
+ "network-role": "",
+ "network-information-items": {
+ "network-information-item": [
+ {
+ "ip-version": "4",
+ "use-dhcp": "N",
+ "ip-count": 0,
+ "network-ips": {
+ "network-ip": []
+ }
+ },
+ {
+ "ip-version": "6",
+ "use-dhcp": "N",
+ "ip-count": 0,
+ "network-ips": {
+ "network-ip": []
+ }
+ }
+ ]
+ },
+ "mac-addresses": {
+ "mac-address": []
+ },
+ "floating-ips": {
+ "floating-ip-v4": [],
+ "floating-ip-v6": []
+ },
+ "interface-route-prefixes": {
+ "interface-route-prefix": []
+ }
+}
--- /dev/null
+{
+ "vm-type": "",
+ "vm-count": 0,
+ "vm-names": {
+ "vm-name": []
+ },
+ "vm-networks": {
+ "vm-network": []
+ }
+}
--- /dev/null
+{
+ "network-role": "",
+ "network-name": ""
+}
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+import json
+import os
+
+from preload import (
+ AbstractPreloadGenerator,
+ get_or_create_template,
+ get_json_template,
+ replace,
+)
+
+THIS_DIR = os.path.dirname(os.path.abspath(__file__))
+DATA_DIR = os.path.join(THIS_DIR, "grapi_data")
+
+
+def get_or_create_network_template(network, vm_networks):
+ """
+ If the network role already exists in vm_networks, then
+ return that otherwise create a blank template and return that
+ """
+ return get_or_create_template(
+ DATA_DIR, "network-role", network, vm_networks, "vm-network"
+ )
+
+
+def add_fixed_ips(network_template, fixed_ips, uses_dhcp):
+ items = network_template["network-information-items"]["network-information-item"]
+ ipv4s = next(item for item in items if item["ip-version"] == "4")
+ ipv6s = next(item for item in items if item["ip-version"] == "6")
+ if uses_dhcp:
+ ipv4s["use-dhcp"] = "Y"
+ ipv6s["use-dhcp"] = "Y"
+ for ip in fixed_ips:
+ target = ipv4s if ip.ip_version == 4 else ipv6s
+ ips = target["network-ips"]["network-ip"]
+ if ip.param not in ips:
+ ips.append(replace(ip.param))
+ target["ip-count"] += 1
+
+
+def add_floating_ips(network_template, floating_ips):
+ for ip in floating_ips:
+ key = "floating-ip-v4" if ip.ip_version == 4 else "floating-ip-v6"
+ ips = network_template["floating-ips"][key]
+ value = replace(ip.param)
+ if value not in ips:
+ ips.append(value)
+
+
+class GrApiPreloadGenerator(AbstractPreloadGenerator):
+ @classmethod
+ def supports_output_passing(cls):
+ return True
+
+ @classmethod
+ def format_name(cls):
+ return "GR-API"
+
+ @classmethod
+ def output_sub_dir(cls):
+ return "grapi"
+
+ def generate_module(self, vnf_module):
+ template = get_json_template(DATA_DIR, "preload_template")
+ self._populate(template, vnf_module)
+ vnf_name = vnf_module.vnf_name
+ outfile = "{}/{}.json".format(self.output_dir, vnf_name)
+ with open(outfile, "w") as f:
+ json.dump(template, f, indent=4)
+
+ def _populate(self, preload, vnf_module):
+ self._add_vnf_metadata(preload)
+ self._add_vms(preload, vnf_module)
+ self._add_availability_zones(preload, vnf_module)
+ self._add_parameters(preload, vnf_module)
+ self._add_vnf_networks(preload, vnf_module)
+
+ @staticmethod
+ def _add_vms(preload, vnf_module):
+ vms = preload["input"]["preload-vf-module-topology-information"][
+ "vf-module-topology"
+ ]["vf-module-assignments"]["vms"]["vm"]
+ for vm in vnf_module.virtual_machine_types:
+ vm_template = get_json_template(DATA_DIR, "vm")
+ vms.append(vm_template)
+ vm_template["vm-type"] = vm.vm_type
+ vm_template["vm-names"]["vm-name"].extend(map(replace, vm.names))
+ vm_template["vm-count"] = vm.vm_count
+ vm_networks = vm_template["vm-networks"]["vm-network"]
+ for port in vm.ports:
+ role = port.network.network_role
+ network_template = get_or_create_network_template(role, vm_networks)
+ network_template["network-role"] = role
+ add_fixed_ips(network_template, port.fixed_ips, port.uses_dhcp)
+ add_floating_ips(network_template, port.floating_ips)
+
+ @staticmethod
+ def _add_availability_zones(preload, vnf_module):
+ zones = preload["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["availability-zones"]["availability-zone"]
+ zones.extend(map(replace, vnf_module.availability_zones))
+
+ @staticmethod
+ def _add_parameters(preload, vnf_module):
+ params = [
+ {"name": key, "value": value}
+ for key, value in vnf_module.preload_parameters.items()
+ ]
+ preload["input"]["preload-vf-module-topology-information"][
+ "vf-module-topology"
+ ]["vf-module-parameters"]["param"].extend(params)
+
+ @staticmethod
+ def _add_vnf_networks(preload, vnf_module):
+ networks = preload["input"]["preload-vf-module-topology-information"][
+ "vnf-resource-assignments"
+ ]["vnf-networks"]["vnf-network"]
+ for network in vnf_module.networks:
+ network_data = {
+ "network-role": network.network_role,
+ "network-name": replace("network name of {}".format(network.name_param)),
+ }
+ if network.subnet_params:
+ network_data["subnets-data"] = {"subnet-data": []}
+ subnet_data = network_data["subnets-data"]["subnet-data"]
+ for subnet_param in network.subnet_params:
+ subnet_data.append({"subnet-id": replace(subnet_param)})
+ networks.append(network_data)
+
+ @staticmethod
+ def _add_vnf_metadata(preload):
+ topology = preload["input"]["preload-vf-module-topology-information"]
+ vnf_meta = topology["vnf-topology-identifier-structure"]
+ vnf_meta["vnf-name"] = replace("vnf_name")
+ vnf_meta["vnf-type"] = replace("Concatenation of "
+ "<Service Name>/<VF Instance Name> "
+ "MUST MATCH SDC")
+ module_meta = topology["vf-module-topology"]["vf-module-topology-identifier"]
+ module_meta["vf-module-name"] = replace("vf_module_name")
+ module_meta["vf-module-type"] = replace("<vfModuleModelName> from CSAR or SDC")
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+from .vnfapi_generator import VnfApiPreloadGenerator
+
+__all__ = ["VnfApiPreloadGenerator"]
--- /dev/null
+{
+ "input": {
+ "request-information": {
+ "request-id": "robot12",
+ "order-version": "1",
+ "notification-url": "openecomp.org",
+ "order-number": "1",
+ "request-action": "PreloadVNFRequest"
+ },
+ "sdnc-request-header": {
+ "svc-request-id": "robot12",
+ "svc-notification-url": "http://openecomp.org:8080/adapters/rest/SDNCNotify",
+ "svc-action": "reserve"
+ },
+ "vnf-topology-information": {
+ "vnf-topology-identifier": {
+ "vnf-name": "",
+ "vnf-type": "",
+ "generic-vnf-type": "",
+ "generic-vnf-name": ""
+ },
+ "vnf-assignments": {
+ "availability-zones": [],
+ "vnf-networks": [],
+ "vnf-vms": []
+ },
+ "vnf-parameters": []
+ }
+ }
+}
\ No newline at end of file
--- /dev/null
+{
+ "vnf-parameter-name": "",
+ "vnf-parameter-value": ""
+}
--- /dev/null
+{
+ "network-role": "",
+ "network-role-tag": "",
+ "ip-count": 0,
+ "ip-count-ipv6": 0,
+ "floating-ip": "",
+ "floating-ip-v6": "",
+ "network-ips": [],
+ "network-ips-v6": [],
+ "network-macs": [],
+ "interface-route-prefixes": []
+}
--- /dev/null
+{
+ "vm-type": "",
+ "vm-count": 0,
+ "vm-names": {
+ "vm-name": []
+ },
+ "vm-networks": []
+}
--- /dev/null
+{
+ "network-role": "",
+ "network-name": ""
+}
--- /dev/null
+# -*- coding: utf8 -*-
+# ============LICENSE_START====================================================
+# org.onap.vvp/validation-scripts
+# ===================================================================
+# Copyright © 2019 AT&T Intellectual Property. All rights reserved.
+# ===================================================================
+#
+# Unless otherwise specified, all software contained herein is licensed
+# under the Apache License, Version 2.0 (the "License");
+# you may not use this software except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+#
+#
+# Unless otherwise specified, all documentation contained herein is licensed
+# under the Creative Commons License, Attribution 4.0 Intl. (the "License");
+# you may not use this documentation except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://creativecommons.org/licenses/by/4.0/
+#
+# Unless required by applicable law or agreed to in writing, documentation
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# ============LICENSE_END============================================
+#
+#
+
+import json
+import os
+
+from preload import (
+ AbstractPreloadGenerator,
+ get_json_template,
+ get_or_create_template,
+ replace,
+)
+
+THIS_DIR = os.path.dirname(os.path.abspath(__file__))
+DATA_DIR = os.path.join(THIS_DIR, "vnfapi_data")
+
+
+def add_fixed_ips(network_template, port):
+ for ip in port.fixed_ips:
+ if ip.ip_version == 4:
+ network_template["network-ips"].append({"ip-address": replace(ip.param)})
+ network_template["ip-count"] += 1
+ else:
+ network_template["network-ips-v6"].append({"ip-address": replace(ip.param)})
+ network_template["ip-count-ipv6"] += 1
+
+
+def add_floating_ips(network_template, network):
+ # only one floating IP is really supported, in the preload model
+ # so for now we'll just use the last one. We might revisit this
+ # and if multiple floating params exist, then come up with an
+ # approach to pick just one
+ for ip in network.floating_ips:
+ key = "floating-ip" if ip.ip_version == 4 else "floating-ip-v6"
+ network_template[key] = replace(ip.param)
+
+
+def get_or_create_network_template(network_role, vm_networks):
+ """
+ If the network role already exists in vm_networks, then
+ return that otherwise create a blank template and return that
+ """
+ return get_or_create_template(
+ DATA_DIR, "network-role", network_role, vm_networks, "vm-network"
+ )
+
+
+class VnfApiPreloadGenerator(AbstractPreloadGenerator):
+ @classmethod
+ def supports_output_passing(cls):
+ return False
+
+ @classmethod
+ def format_name(cls):
+ return "VNF-API"
+
+ @classmethod
+ def output_sub_dir(cls):
+ return "vnfapi"
+
+ def generate_module(self, vnf_module):
+ preload = get_json_template(DATA_DIR, "preload_template")
+ self._populate(preload, vnf_module)
+ outfile = "{}/{}.json".format(self.output_dir, vnf_module.vnf_name)
+ with open(outfile, "w") as f:
+ json.dump(preload, f, indent=4)
+
+ def _populate(self, preload, vnf_module):
+ self._add_availability_zones(preload, vnf_module)
+ self._add_vnf_networks(preload, vnf_module)
+ self._add_vms(preload, vnf_module)
+ self._add_parameters(preload, vnf_module)
+
+ @staticmethod
+ def _add_availability_zones(preload, vnf_module):
+ zones = preload["input"]["vnf-topology-information"]["vnf-assignments"][
+ "availability-zones"
+ ]
+ for zone in vnf_module.availability_zones:
+ zones.append({"availability-zone": replace(zone)})
+
+ @staticmethod
+ def _add_vnf_networks(preload, vnf_module):
+ networks = preload["input"]["vnf-topology-information"]["vnf-assignments"][
+ "vnf-networks"
+ ]
+ for network in vnf_module.networks:
+ network_data = {
+ "network-role": network.network_role,
+ "network-name": replace(
+ "network name for {}".format(network.name_param)
+ ),
+ }
+ for subnet in network.subnet_params:
+ key = "ipv6-subnet-id" if "_v6_" in subnet else "subnet-id"
+ network_data[key] = subnet
+ networks.append(network_data)
+
+ @staticmethod
+ def _add_vms(preload, vnf_module):
+ vm_list = preload["input"]["vnf-topology-information"]["vnf-assignments"][
+ "vnf-vms"
+ ]
+ for vm in vnf_module.virtual_machine_types:
+ vm_template = get_json_template(DATA_DIR, "vm")
+ vm_template["vm-type"] = vm.vm_type
+ vm_template["vm-count"] = vm.vm_count
+ vm_template["vm-names"]["vm-name"].extend(map(replace, vm.names))
+ vm_list.append(vm_template)
+ vm_networks = vm_template["vm-networks"]
+ for port in vm.ports:
+ role = port.network.network_role
+ network_template = get_or_create_network_template(role, vm_networks)
+ network_template["network-role"] = role
+ network_template["network-role-tag"] = role
+ network_template["use-dhcp"] = "Y" if port.uses_dhcp else "N"
+ add_fixed_ips(network_template, port)
+ add_floating_ips(network_template, port)
+
+ @staticmethod
+ def _add_parameters(preload, vnf_module):
+ params = preload["input"]["vnf-topology-information"]["vnf-parameters"]
+ for key, value in vnf_module.preload_parameters.items():
+ params.append({"vnf-parameter-name": key, "vnf-parameter-value": value})
import os
import re
import time
+
+from preload import create_preloads
+from tests.helpers import get_output_dir
+
try:
from html import escape
except ImportError:
ALL_RESULTS = []
-def get_output_dir(config):
- """
- Retrieve the output directory for the reports and create it if necessary
- :param config: pytest configuration
- :return: output directory as string
- """
- output_dir = config.option.output_dir or DEFAULT_OUTPUT_DIR
- if not os.path.exists(output_dir):
- os.makedirs(output_dir, exist_ok=True)
- return output_dir
-
-
def extract_error_msg(rep):
"""
If a custom error message was provided, then extract it otherwise
)
+def pytest_terminal_summary(terminalreporter, exitstatus):
+ # Ensures all preload information and warnings appear after
+ # test results
+ create_preloads(terminalreporter.config, exitstatus)
+
+
# noinspection PyUnusedLocal
def pytest_collection_modifyitems(session, config, items):
"""
{
"file_links": make_href(failure.files, template_path),
"test_id": failure.test_id,
- "error_message": escape(failure.error_message).replace("\n",
- "<br/><br/>"),
+ "error_message": escape(failure.error_message).replace(
+ "\n", "<br/><br/>"
+ ),
"raw_output": escape(failure.raw_output),
"requirements": docutils.core.publish_parts(
writer_name="html", source=failure.requirement_text(reqs)
from boltons import funcutils
from tests import cached_yaml as yaml
-VERSION = "1.1.0"
+__path__ = [os.path.dirname(os.path.abspath(__file__))]
+DEFAULT_OUTPUT_DIR = "{}/../output".format(__path__[0])
+RE_BASE = re.compile(r"(^base$)|(^base_)|(_base_)|(_base$)")
+
+
+def is_base_module(template_path):
+ basename = os.path.basename(template_path).lower()
+ name, extension = os.path.splitext(basename)
+ is_yaml = extension in {".yml", ".yaml"}
+ return is_yaml and RE_BASE.search(name) and not name.endswith("_volume")
def check_basename_ending(template_type, basename):
return invalid_params
-RE_BASE = re.compile(r"(^base$)|(^base_)|(_base_)|(_base$)")
-
-
def get_base_template_from_yaml_files(yaml_files):
"""Return first filepath to match RE_BASE
"""
else:
return param
return None
+
+
+def get_output_dir(config):
+ """
+ Retrieve the output directory for the reports and create it if necessary
+ :param config: pytest configuration
+ :return: output directory as string
+ """
+ output_dir = config.option.output_dir or DEFAULT_OUTPUT_DIR
+ if not os.path.exists(output_dir):
+ os.makedirs(output_dir, exist_ok=True)
+ return output_dir
import sys
from tests import cached_yaml as yaml
-from tests.helpers import load_yaml
+from tests.helpers import load_yaml, get_param
from .utils import nested_dict
VERSION = "4.2.0"
resource_type=ContrailV2VirtualMachineInterfaceProcessor.resource_type
)
- def get_all_resources(self, base_dir):
+ def get_all_resources(self, base_dir=None, count=1):
"""
- Like ``resources``,
- but this returns all the resources definitions
+ Like ``resources``, but this returns all the resources definitions
defined in the template, resource groups, and nested YAML files.
+
+ A special variable will be added to all resource properties (__count__).
+ This will normally be 1, but if the resource is generated by a
+ ResourceGroup **and** an env file is present, then the count will be
+ the value from the env file (assuming this follows standard VNF Heat
+ Guidelines)
"""
+ base_dir = base_dir or self.dirname
resources = {}
for r_id, r_data in self.resources.items():
+ r_data["__count__"] = count
resources[r_id] = r_data
resource = Resource(r_id, r_data)
if resource.is_nested():
+ nested_count = resource.get_count(self.env)
nested = Heat(os.path.join(base_dir, resource.get_nested_filename()))
- resources.update(nested.get_all_resources(base_dir))
+ nested_resources = nested.get_all_resources(count=nested_count)
+ resources.update(nested_resources)
return resources
@staticmethod
"""
return _HEAT_PROCESSORS
- def get_resource_by_type(self, resource_type):
+ def get_resource_by_type(self, resource_type, all_resources=False):
"""Return dict of resources whose type is `resource_type`.
key is resource_id, value is resource.
"""
+ resources = self.get_all_resources() if all_resources else self.resources
return {
rid: resource
- for rid, resource in self.resources.items()
+ for rid, resource in resources.items()
if self.nested_get(resource, "type") == resource_type
}
else:
return self.properties
+ def get_count(self, env):
+ if self.resource_type == "OS::Heat::ResourceGroup":
+ if not env:
+ return 1
+ env_params = env.parameters
+ count_param = get_param(self.properties["count"])
+ count_value = env_params.get(count_param) if count_param else 1
+ try:
+ return int(count_value)
+ except (ValueError, TypeError):
+ print((
+ "WARNING: Invalid value for count parameter {}. Expected "
+ "an integer, but got {}. Defaulting to 1"
+ ).format(count_param, count_value))
+ return 1
+
@property
def depends_on(self):
"""
""" environment file structure
"""
import os
-from collections import Iterable
-from tests.structures import Heat
-from tests.utils import nested_dict
-from .helpers import (
- validates,
- categories,
+import re
+import pytest
+from tests.helpers import (
+ prop_iterator,
+ get_param,
get_environment_pair,
+ validates,
find_environment_file,
- get_param,
+ categories,
)
-import re
-import pytest
-from tests import cached_yaml as yaml
-
-VERSION = "1.0.0"
-
-# pylint: disable=invalid-name
-
-
-def check_parameter_exists(pattern, parameters):
- if not parameters:
- return False
-
- for param in parameters:
- if pattern.search(param):
- return True
-
- return False
-
-
-def check_param_in_env_file(environment_pair, param, DESIRED, exclude_parameter=None):
-
- # workaround for internal/external parameters
- if exclude_parameter and re.match(exclude_parameter, param):
- return False
-
- if not environment_pair:
- pytest.skip("No heat/env pair could be identified")
-
- env_file = environment_pair.get("eyml")
-
- pattern = re.compile(r"^{}$".format(param))
-
- if "parameters" not in env_file:
- pytest.skip("No parameters specified in the environment file")
+from tests.structures import Heat
+from tests.utils.nested_files import file_is_a_nested_template
+
+
+# Whats persistent mean? It means it goes in env.
+# When adding an additional case, note the ","
+# at the end of a property to make it a tuple.
+ENV_PARAMETER_SPEC = {
+ "PLATFORM PROVIDED": [
+ {"property": ("vnf_id",), "persistent": False, "kwargs": {}},
+ {"property": ("vnf_name",), "persistent": False, "kwargs": {}},
+ {"property": ("vf_module_id",), "persistent": False, "kwargs": {}},
+ {"property": ("vf_module_index",), "persistent": False, "kwargs": {}},
+ {"property": ("vf_module_name",), "persistent": False, "kwargs": {}},
+ {"property": ("workload_context",), "persistent": False, "kwargs": {}},
+ {"property": ("environment_context",), "persistent": False, "kwargs": {}},
+ {"property": (r"^(.+?)_net_fqdn$",), "persistent": False, "kwargs": {}},
+ ],
+ "ALL": [{"property": ("name",), "persistent": False, "kwargs": {}}],
+ "OS::Nova::Server": [
+ {"property": ("image",), "persistent": True, "kwargs": {}},
+ {"property": ("flavor",), "persistent": True, "kwargs": {}},
+ {"property": ("availability_zone",), "persistent": False, "kwargs": {}},
+ ],
+ "OS::Neutron::Port": [
+ {"property": ("network",), "persistent": False, "kwargs": {}},
+ {
+ "property": ("fixed_ips", "ip_address"),
+ "persistent": False,
+ "network_type": "external",
+ "kwargs": {"exclude_parameter": re.compile(r"^(.+?)_int_(.+?)$")},
+ },
+ {
+ "property": ("fixed_ips", "ip_address"),
+ "persistent": True,
+ "network_type": "internal",
+ "kwargs": {"exclude_parameter": re.compile(r"^((?!_int_).)*$")},
+ },
+ {"property": ("fixed_ips", "subnet"), "persistent": False, "kwargs": {}},
+ {
+ "property": ("fixed_ips", "allowed_address_pairs"),
+ "persistent": False,
+ "network_type": "external",
+ "kwargs": {"exclude_parameter": re.compile(r"^(.+?)_int_(.+?)$")},
+ },
+ {
+ "property": ("fixed_ips", "allowed_address_pairs"),
+ "persistent": True,
+ "network_type": "internal",
+ "kwargs": {"exclude_parameter": re.compile(r"^((?!_int_).)*$")},
+ },
+ ],
+ "OS::ContrailV2::InterfaceRouteTable": [
+ {
+ "property": (
+ "interface_route_table_routes",
+ "interface_route_table_routes_route",
+ ),
+ "persistent": False,
+ "kwargs": {},
+ }
+ ],
+ "OS::Heat::ResourceGroup": [
+ {
+ "property": ("count",),
+ "persistent": True,
+ "kwargs": {
+ "exclude_resource": re.compile(
+ r"^(.+?)_subint_(.+?)_port_(.+?)_subinterfaces$"
+ )
+ },
+ }
+ ],
+ "OS::ContrailV2::InstanceIp": [
+ {
+ "property": ("instance_ip_address",),
+ "persistent": False,
+ "network_type": "external",
+ "kwargs": {"exclude_resource": re.compile(r"^.*_int_.*$")},
+ },
+ {
+ "property": ("instance_ip_address",),
+ "persistent": True,
+ "network_type": "internal",
+ "kwargs": {"exclude_resource": re.compile(r"(?!.*_int_.*)")},
+ },
+ {
+ "property": ("subnet_uuid",),
+ "persistent": False,
+ "network_type": "internal",
+ "kwargs": {"exclude_resource": re.compile(r"(?!.*_int_.*)")},
+ },
+ ],
+ "OS::ContrailV2::VirtualMachineInterface": [
+ {
+ "property": (
+ "virtual_machine_interface_allowed_address_pairs",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair_ip",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair_ip_ip_prefix",
+ ),
+ "persistent": False,
+ "network_type": "external",
+ "kwargs": {"exclude_resource": re.compile(r"(?!.*_int_.*)")},
+ }
+ ],
+}
+
+
+def run_test_parameter(yaml_file, resource_type, *prop, **kwargs):
+ template_parameters = []
+ invalid_parameters = []
+ param_spec = {}
+ parameter_spec = ENV_PARAMETER_SPEC.get(
+ resource_type
+ ) # matching spec dict on resource type
+ for spec in parameter_spec:
+ # iterating through spec dict and trying to match on property
+ if spec.get("property") == prop:
+ yep = True
+ for (
+ k,
+ v,
+ ) in (
+ kwargs.items()
+ ): # now matching on additional kwargs passed in from test (i.e. network_type)
+ if not spec.get(k) or spec.get(k) != v:
+ yep = False
+ if yep:
+ param_spec = spec
+ if resource_type == "PLATFORM PROVIDED":
+ if file_is_a_nested_template(yaml_file):
+ pytest.skip(
+ "Not checking nested files for PLATFORM PROVIDED params"
+ )
+ template_parameters.append(
+ {"resource": "", "param": param_spec.get("property")[0]}
+ )
+ else:
+ all_resources = False
+ if resource_type == "ALL":
+ all_resources = True
+ template_parameters = get_template_parameters(
+ yaml_file,
+ resource_type,
+ param_spec,
+ all_resources=all_resources,
+ ) # found the correct spec, proceeding w/ test
+ break
+
+ for parameter in template_parameters:
+ param = parameter.get("param")
+ persistence = param_spec.get("persistent")
+
+ if env_violation(yaml_file, param, spec.get("persistent")):
+ human_text = "must" if persistence else "must not"
+ human_text2 = "was not" if persistence else "was"
+
+ invalid_parameters.append(
+ "{} parameter {} {} be enumerated in an environment file, but "
+ "parameter {} for {} {} found.".format(
+ resource_type, prop, human_text, param, yaml_file, human_text2
+ )
+ )
- return (
- check_parameter_exists(pattern, env_file.get("parameters", {})) is not DESIRED
- )
+ assert not invalid_parameters, "\n".join(invalid_parameters)
-"""
-This function supports this structure, deviations
-may or may not work without enhancement
-
-resource_id:
- type: <resource_type>
- properties:
- prop0: { get_param: parameter_0 }
- prop1: # this is a list of dicts
- - nested_prop_0: { get_param: parameter_1 }
- - nested_prop_1: { get_param: [parameter_2, {index}] }
- prop2: # this is a dict of dicts
- nested_prop_0: { get_param: parameter_1 }
- prop3: { get_param: [parameter_3, 0]}
-"""
+def get_preload_excluded_parameters(yaml_file):
+ """
+ Returns set of all parameters that should not be included in the preload's
+ vnf parameters/tag-values section.
+ """
+ results = []
+ for resource_type, specs in ENV_PARAMETER_SPEC.items():
+ # apply to all resources if not in the format of an OpenStack resource
+ all_resources = "::" not in resource_type
+ for spec in specs:
+ results.extend(get_template_parameters(yaml_file, resource_type,
+ spec, all_resources))
+ return {item["param"] for item in results}
-def check_resource_parameter(
- environment_pair,
- prop,
- DESIRED,
- resource_type,
- resource_type_inverse=False,
- nested_prop="",
- exclude_resource="",
- exclude_parameter="",
-):
- if not environment_pair:
- pytest.skip("No heat/env pair could be identified")
+def get_template_parameters(yaml_file, resource_type, spec, all_resources=False):
+ parameters = []
- env_file = environment_pair.get("eyml")
- template_file = environment_pair.get("yyml")
+ heat = Heat(yaml_file)
+ if all_resources:
+ resources = heat.resources
+ else:
+ resources = heat.get_resource_by_type(resource_type)
- if "parameters" not in env_file:
- pytest.skip("No parameters specified in the environment file")
+ for rid, resource_props in resources.items():
+ for param in prop_iterator(resource_props, *spec.get("property")):
+ if param and get_param(param) and param_helper(spec, get_param(param), rid):
+ # this is first getting the param
+ # then checking if its actually using get_param
+ # then checking a custom helper function (mostly for internal vs external networks)
+ parameters.append({"resource": rid, "param": get_param(param)})
- invalid_parameters = []
- if template_file:
- for resource, resource_prop in template_file.get("resources", {}).items():
-
- # workaround for subinterface resource groups
- if exclude_resource and re.match(exclude_resource, resource):
- continue
-
- if (
- resource_prop.get("type") == resource_type and not resource_type_inverse
- ) or (resource_prop.get("type") != resource_type and resource_type_inverse):
-
- pattern = False
-
- if not resource_prop.get("properties"):
- continue
-
- resource_parameter = resource_prop.get("properties").get(prop)
-
- if not resource_parameter:
- continue
- if isinstance(resource_parameter, list) and nested_prop:
- for param in resource_parameter:
- nested_param = param.get(nested_prop)
- if not nested_param:
- continue
-
- if isinstance(nested_param, dict):
- pattern = nested_param.get("get_param")
- else:
- pattern = ""
-
- if not pattern:
- continue
-
- if isinstance(pattern, list):
- pattern = pattern[0]
-
- if check_param_in_env_file(
- environment_pair,
- pattern,
- DESIRED,
- exclude_parameter=exclude_parameter,
- ):
- invalid_parameters.append(pattern)
-
- elif isinstance(resource_parameter, dict):
- if nested_prop and nested_prop in resource_parameter:
- resource_parameter = resource_parameter.get(nested_prop)
-
- pattern = resource_parameter.get("get_param")
- if not pattern:
- continue
-
- if isinstance(pattern, list):
- pattern = pattern[0]
-
- if check_param_in_env_file(
- environment_pair,
- pattern,
- DESIRED,
- exclude_parameter=exclude_parameter,
- ):
- invalid_parameters.append(pattern)
- else:
- continue
+ return parameters
- return set(invalid_parameters)
+def env_violation(yaml_file, parameter, persistent):
+ # Returns True IF there's a violation, False if everything looks good.
-def run_check_resource_parameter(
- yaml_file, prop, DESIRED, resource_type, check_resource=True, **kwargs
-):
filepath, filename = os.path.split(yaml_file)
environment_pair = get_environment_pair(yaml_file)
-
- if not environment_pair:
- # this is a nested file
-
- if not check_resource:
- # dont check env for nested files
- # This will be tested separately for parent template
- pytest.skip("This test doesn't apply to nested files")
-
- environment_pair = find_environment_file(yaml_file)
- if environment_pair:
- with open(yaml_file, "r") as f:
- yml = yaml.load(f)
- environment_pair["yyml"] = yml
- else:
+ if not environment_pair: # this is a nested file perhaps?
+ environment_pair = find_environment_file(
+ yaml_file
+ ) # we want to check parent env
+ if not environment_pair:
pytest.skip("unable to determine environment file for nested yaml file")
- if check_resource:
- invalid_parameters = check_resource_parameter(
- environment_pair, prop, DESIRED, resource_type, **kwargs
- )
- else:
- invalid_parameters = check_param_in_env_file(environment_pair, prop, DESIRED)
+ env_yaml = environment_pair.get("eyml")
+ parameters = env_yaml.get("parameters", {})
+ in_env = False
+ for param, value in parameters.items():
+ if re.match(parameter, parameter):
+ in_env = True
+ break
- if kwargs.get("resource_type_inverse"):
- resource_type = "non-{}".format(resource_type)
+ # confusing return. This function is looking for a violation.
+ return not persistent == in_env
- params = (
- ": {}".format(", ".join(invalid_parameters))
- if isinstance(invalid_parameters, Iterable)
- else ""
- )
- assert not invalid_parameters, (
- "{} {} parameters in template {}{}"
- " found in {} environment file{}".format(
- resource_type,
- prop,
- filename,
- " not" if DESIRED else "",
- environment_pair.get("name"),
- params,
- )
- )
+def param_helper(spec, param, rid):
+ # helper function that has some predefined additional
+ # checkers, mainly to figure out if internal/external network
+ keeper = True
+ for k, v in spec.get("kwargs").items():
+ if k == "exclude_resource" and re.match(v, rid):
+ keeper = False
+ break
+ elif k == "exclude_parameter" and re.match(v, param):
+ keeper = False
+ break
+
+ return keeper
@validates("R-91125")
def test_nova_server_image_parameter_exists_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "image", True, "OS::Nova::Server")
+ run_test_parameter(yaml_file, "OS::Nova::Server", "image")
@validates("R-69431")
def test_nova_server_flavor_parameter_exists_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "flavor", True, "OS::Nova::Server")
+ run_test_parameter(yaml_file, "OS::Nova::Server", "flavor")
@categories("environment_file")
-@validates("R-22838")
+@validates("R-22838", "R-99812")
def test_nova_server_name_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "name", False, "OS::Nova::Server")
+ run_test_parameter(yaml_file, "ALL", "name")
@categories("environment_file")
@validates("R-59568")
def test_nova_server_az_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file, "availability_zone", False, "OS::Nova::Server"
- )
+ run_test_parameter(yaml_file, "OS::Nova::Server", "availability_zone")
@categories("environment_file")
@validates("R-20856")
def test_nova_server_vnf_id_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "vnf_id", False, "", check_resource=False)
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "vnf_id")
@categories("environment_file")
@validates("R-72871")
def test_nova_server_vf_module_id_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file, "vf_module_id", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "vf_module_id")
@categories("environment_file")
def test_nova_server_vf_module_index_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
- yaml_file, "vf_module_index", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "vf_module_index")
@categories("environment_file")
@validates("R-36542")
def test_nova_server_vnf_name_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "vnf_name", False, "", check_resource=False)
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "vnf_name")
@categories("environment_file")
def test_nova_server_vf_module_name_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
- yaml_file, "vf_module_name", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "vf_module_name")
@categories("environment_file")
def test_nova_server_workload_context_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
- yaml_file, "workload_context", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "workload_context")
@categories("environment_file")
def test_nova_server_environment_context_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
- yaml_file, "environment_context", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", "environment_context")
@categories("environment_file")
@validates("R-29872")
def test_neutron_port_network_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(yaml_file, "network", False, "OS::Neutron::Port")
+ run_test_parameter(yaml_file, "OS::Neutron::Port", "network")
@categories("environment_file")
def test_neutron_port_external_fixedips_ipaddress_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "fixed_ips",
- False,
"OS::Neutron::Port",
- nested_prop="ip_address",
- exclude_parameter=re.compile(r"^(.+?)_int_(.+?)$"),
+ "fixed_ips",
+ "ip_address",
+ network_type="external",
)
def test_neutron_port_internal_fixedips_ipaddress_parameter_exists_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "fixed_ips",
- True,
"OS::Neutron::Port",
- nested_prop="ip_address",
- exclude_parameter=re.compile(r"^((?!_int_).)*$"),
+ "fixed_ips",
+ "ip_address",
+ network_type="internal",
)
def test_neutron_port_fixedips_subnet_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
- yaml_file, "fixed_ips", False, "OS::Neutron::Port", nested_prop="subnet"
+ run_test_parameter(
+ yaml_file, "OS::Neutron::Port", "fixed_ips", "subnet", network_type="internal"
)
def test_neutron_port_external_aap_ip_parameter_doesnt_exist_in_environment_file(
yaml_file
):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "allowed_address_pairs",
- False,
"OS::Neutron::Port",
- nested_prop="ip_address",
- exclude_parameter=re.compile(r"^(.+?)_int_(.+?)$"),
- )
-
-
-@categories("environment_file")
-@validates("R-99812")
-def test_non_nova_server_name_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file, "name", False, "OS::Nova::Server", resource_type_inverse=True
+ "allowed_address_pairs",
+ "subnet",
+ network_type="external",
)
@categories("environment_file")
@validates("R-92193")
def test_network_fqdn_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file, r"^(.+?)_net_fqdn$", False, "", check_resource=False
- )
+ run_test_parameter(yaml_file, "PLATFORM PROVIDED", r"^(.+?)_net_fqdn$")
@categories("environment_file")
@validates("R-76682")
def test_contrail_route_prefixes_parameter_doesnt_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "interface_route_table_routes",
- False,
"OS::ContrailV2::InterfaceRouteTable",
- nested_prop="interface_route_table_routes_route",
+ "interface_route_table_routes",
+ "interface_route_table_routes_route",
)
@validates("R-50011")
def test_heat_rg_count_parameter_exists_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file,
- "count",
- True,
- "OS::Heat::ResourceGroup",
- exclude_resource=re.compile(r"^(.+?)_subint_(.+?)_port_(.+?)_subinterfaces$"),
- )
+ run_test_parameter(yaml_file, "OS::Heat::ResourceGroup", "count")
@categories("environment_file")
@validates("R-100020", "R-100040", "R-100060", "R-100080", "R-100170")
def test_contrail_external_instance_ip_does_not_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "instance_ip_address",
- False,
"OS::ContrailV2::InstanceIp",
- exclude_resource=re.compile(r"^.*_int_.*$"), # exclude internal IPs
+ "instance_ip_address",
+ network_type="external",
)
@validates("R-100100", "R-100120", "R-100140", "R-100160", "R-100180")
def test_contrail_internal_instance_ip_does_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
+ run_test_parameter(
yaml_file,
- "instance_ip_address",
- True,
"OS::ContrailV2::InstanceIp",
- exclude_resource=re.compile(r"(?!.*_int_.*)"), # exclude external IPs
+ "instance_ip_address",
+ network_type="internal",
)
@categories("environment_file")
@validates("R-100210", "R-100230", "R-100250", "R-100270")
def test_contrail_subnet_uuid_does_not_exist_in_environment_file(yaml_file):
- run_check_resource_parameter(
- yaml_file, "subnet_uuid", False, "OS::ContrailV2::InstanceIp"
- )
+ run_test_parameter(yaml_file, "OS::ContrailV2::InstanceIp", "subnet_uuid")
@categories("environment_file")
@validates("R-100320", "R-100340")
def test_contrail_vmi_aap_does_not_exist_in_environment_file(yaml_file):
- # This test needs to check a more complex structure. Rather than try to force
- # that into the existing run_check_resource_parameter logic we'll just check it
- # directly
- pairs = get_environment_pair(yaml_file)
- if not pairs:
- pytest.skip("No matching env file found")
- heat = Heat(filepath=yaml_file)
- env_parameters = pairs["eyml"].get("parameters") or {}
- vmis = heat.get_resource_by_type("OS::ContrailV2::VirtualMachineInterface")
- external_vmis = {rid: data for rid, data in vmis.items() if "_int_" not in rid}
- invalid_params = []
- for r_id, vmi in external_vmis.items():
- aap_value = nested_dict.get(
- vmi,
- "properties",
- "virtual_machine_interface_allowed_address_pairs",
- "virtual_machine_interface_allowed_address_pairs_allowed_address_pair",
- )
- if not aap_value or not isinstance(aap_value, list):
- # Skip if aap not used or is not a list.
- continue
- for pair_ip in aap_value:
- if not isinstance(pair_ip, dict):
- continue # Invalid Heat will be detected by another test
- settings = (
- pair_ip.get(
- "virtual_machine_interface_allowed_address"
- "_pairs_allowed_address_pair_ip"
- )
- or {}
- )
- if isinstance(settings, dict):
- ip_prefix = (
- settings.get(
- "virtual_machine_interface_allowed_address"
- "_pairs_allowed_address_pair_ip_ip_prefix"
- )
- or {}
- )
- ip_prefix_param = get_param(ip_prefix)
- if ip_prefix_param and ip_prefix_param in env_parameters:
- invalid_params.append(ip_prefix_param)
-
- msg = (
- "OS::ContrailV2::VirtualMachineInterface "
- "virtual_machine_interface_allowed_address_pairs"
- "_allowed_address_pair_ip_ip_prefix "
- "parameters found in environment file {}: {}"
- ).format(pairs.get("name"), ", ".join(invalid_params))
- assert not invalid_params, msg
+ run_test_parameter(
+ yaml_file,
+ "OS::ContrailV2::VirtualMachineInterface",
+ "virtual_machine_interface_allowed_address_pairs",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair_ip",
+ "virtual_machine_interface_allowed_address_pairs_allowed_address_pair_ip_ip_prefix",
+ )
import appdirs
import os
import pytest
-import sys
import version
import yaml
import contextlib
import tempfile
import webbrowser
import zipfile
+import platform
+import subprocess # nosec
from collections import MutableMapping
from configparser import ConfigParser
from tkinter.scrolledtext import ScrolledText
from typing import Optional, List, Dict, TextIO, Callable, Iterator
+import preload
+
VERSION = version.VERSION
PATH = os.path.dirname(os.path.realpath(__file__))
OUT_DIR = "output"
pass
-def get_plugins() -> Optional[List]:
- """When running in a frozen bundle, plugins to be registered
- explicitly. This method will return the required plugins to register
- based on the run mode"""
- if hasattr(sys, "frozen"):
- import pytest_tap.plugin
-
- return [pytest_tap.plugin]
- else:
- return None
-
-
def run_pytest(
template_dir: str,
log: TextIO,
args.extend(("--category", category))
if not halt_on_failure:
args.append("--continue-on-failure")
- pytest.main(args=args, plugins=get_plugins())
+ pytest.main(args=args)
result_queue.put((True, None))
except Exception as e:
result_queue.put((False, e))
def report_formats(self):
return ["CSV", "Excel", "HTML"]
+ @property
+ def preload_formats(self):
+ excluded = self._config.get("excluded-preloads", [])
+ formats = (cls.format_name() for cls in preload.get_generator_plugins())
+ return [f for f in formats if f not in excluded]
+
+ @property
+ def default_preload_format(self):
+ default = self._user_settings.get("preload_format")
+ if default and default in self.preload_formats:
+ return default
+ else:
+ return self.preload_formats[0]
+
+ @staticmethod
+ def get_subdir_for_preload(preload_format):
+ for gen in preload.get_generator_plugins():
+ if gen.format_name() == preload_format:
+ return gen.output_sub_dir()
+ return ""
+
@property
def default_input_format(self):
requested_default = self._user_settings.get("input_format") or self._config[
category_checkbox.grid(row=x + 1, column=1, columnspan=2, sticky="w")
settings_frame = LabelFrame(actions, text="Settings")
+ settings_row = 1
settings_frame.grid(row=3, column=1, columnspan=3, pady=10, sticky="we")
verbosity_label = Label(settings_frame, text="Verbosity:")
- verbosity_label.grid(row=1, column=1, sticky=W)
+ verbosity_label.grid(row=settings_row, column=1, sticky=W)
self.verbosity = StringVar(self._root, name="verbosity")
self.verbosity.set(self.config.default_verbosity(self.VERBOSITY_LEVELS))
verbosity_menu = OptionMenu(
settings_frame, self.verbosity, *tuple(self.VERBOSITY_LEVELS.keys())
)
verbosity_menu.config(width=25)
- verbosity_menu.grid(row=1, column=2, columnspan=3, sticky=E, pady=5)
+ verbosity_menu.grid(row=settings_row, column=2, columnspan=3, sticky=E, pady=5)
+ settings_row += 1
+
+ if self.config.preload_formats:
+ preload_format_label = Label(settings_frame, text="Preload Template:")
+ preload_format_label.grid(row=settings_row, column=1, sticky=W)
+ self.preload_format = StringVar(self._root, name="preload_format")
+ self.preload_format.set(self.config.default_preload_format)
+ preload_format_menu = OptionMenu(
+ settings_frame, self.preload_format, *self.config.preload_formats
+ )
+ preload_format_menu.config(width=25)
+ preload_format_menu.grid(
+ row=settings_row, column=2, columnspan=3, sticky=E, pady=5
+ )
+ settings_row += 1
report_format_label = Label(settings_frame, text="Report Format:")
- report_format_label.grid(row=2, column=1, sticky=W)
+ report_format_label.grid(row=settings_row, column=1, sticky=W)
self.report_format = StringVar(self._root, name="report_format")
self.report_format.set(self.config.default_report_format)
report_format_menu = OptionMenu(
settings_frame, self.report_format, *self.config.report_formats
)
report_format_menu.config(width=25)
- report_format_menu.grid(row=2, column=2, columnspan=3, sticky=E, pady=5)
+ report_format_menu.grid(
+ row=settings_row, column=2, columnspan=3, sticky=E, pady=5
+ )
+ settings_row += 1
input_format_label = Label(settings_frame, text="Input Format:")
- input_format_label.grid(row=3, column=1, sticky=W)
+ input_format_label.grid(row=settings_row, column=1, sticky=W)
self.input_format = StringVar(self._root, name="input_format")
self.input_format.set(self.config.default_input_format)
input_format_menu = OptionMenu(
settings_frame, self.input_format, *self.config.input_formats
)
input_format_menu.config(width=25)
- input_format_menu.grid(row=3, column=2, columnspan=3, sticky=E, pady=5)
+ input_format_menu.grid(
+ row=settings_row, column=2, columnspan=3, sticky=E, pady=5
+ )
+ settings_row += 1
self.halt_on_failure = BooleanVar(self._root, name="halt_on_failure")
self.halt_on_failure.set(self.config.default_halt_on_failure)
halt_on_failure_label = Label(settings_frame, text="Halt on Basic Failures:")
- halt_on_failure_label.grid(row=4, column=1, sticky=E, pady=5)
+ halt_on_failure_label.grid(row=settings_row, column=1, sticky=E, pady=5)
halt_checkbox = Checkbutton(
settings_frame, offvalue=False, onvalue=True, variable=self.halt_on_failure
)
- halt_checkbox.grid(row=4, column=2, columnspan=2, sticky=W, pady=5)
+ halt_checkbox.grid(row=settings_row, column=2, columnspan=2, sticky=W, pady=5)
directory_label = Label(actions, text="Template Location:")
directory_label.grid(row=4, column=1, pady=5, sticky=W)
)
self.underline(self.result_label)
self.result_label.bind("<Button-1>", self.open_report)
+
+ self.preload_label = Label(
+ self.result_panel, text="View Preloads", fg="blue", cursor="hand2"
+ )
+ self.underline(self.preload_label)
+ self.preload_label.bind("<Button-1>", self.open_preloads)
+
self.result_panel.grid(row=6, column=1, columnspan=2)
control_panel.pack(fill=BOTH, expand=1)
# room for them
self.completion_label.pack()
self.result_label.pack() # Show report link
+ self.preload_label.pack() # Show preload link
self._root.after_idle(
lambda: (
self.completion_label.pack_forget(),
self.result_label.pack_forget(),
+ self.preload_label.pack_forget(),
)
)
self.report_format,
self.halt_on_failure,
)
+ if self.config.preload_formats:
+ self.config.watch(self.preload_format)
self.schedule(self.execute_pollers)
if self.config.terms_link_text and not self.config.are_terms_accepted:
TermsAndConditionsDialog(parent_frame, self.config)
def create_footer(self, parent_frame):
footer = Frame(parent_frame)
- disclaimer = Message(
- footer, text=self.config.disclaimer_text, anchor=CENTER
- )
+ disclaimer = Message(footer, text=self.config.disclaimer_text, anchor=CENTER)
disclaimer.grid(row=0, pady=2)
parent_frame.bind(
"<Configure>", lambda e: disclaimer.configure(width=e.width - 20)
self.clear_log()
self.completion_label.pack_forget()
self.result_label.pack_forget()
+ self.preload_label.pack_forget()
self.task = multiprocessing.Process(
target=run_pytest,
args=(
if is_success:
self.completion_label.pack()
self.result_label.pack() # Show report link
+ if hasattr(self, "preload_format"):
+ self.preload_label.pack() # Show preload link
else:
self.log_panel.insert(END, str(e))
"""Open the report in the user's default browser"""
webbrowser.open_new("file://{}".format(self.report_file_path))
+ def open_preloads(self, event):
+ """Open the report in the user's default browser"""
+ path = os.path.join(
+ PATH,
+ OUT_DIR,
+ "preloads",
+ self.config.get_subdir_for_preload(self.preload_format.get()),
+ )
+ if platform.system() == "Windows":
+ os.startfile(path) # nosec
+ elif platform.system() == "Darwin":
+ subprocess.Popen(["open", path]) # nosec
+ else:
+ subprocess.Popen(["xdg-open", path]) # nosec
+
def open_requirements(self):
"""Open the report in the user's default browser"""
webbrowser.open_new(self.config.requirement_link_url)
Code Review / vvp / validation-scripts.git / commitdiff