--- /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]
Code Review / vvp / validation-scripts.git / blobdiff