X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=ice_validator%2Fpreload%2Fenvironment.py;h=120e3c8240299e106d095ccb77989a2710d731e4;hb=HEAD;hp=c0f357a9e313abde0cd5a1692c80003af0d33d7f;hpb=84db7f8f65cd0ec77f09cfde365599df9890ce6c;p=vvp%2Fvalidation-scripts.git diff --git a/ice_validator/preload/environment.py b/ice_validator/preload/environment.py index c0f357a..120e3c8 100644 --- a/ice_validator/preload/environment.py +++ b/ice_validator/preload/environment.py @@ -1,14 +1,20 @@ +import os import re import tempfile from pathlib import Path +from typing import Any, Optional, Mapping from cached_property import cached_property +from preload.data import AbstractPreloadInstance, AbstractPreloadDataSource +from preload.model import VnfModule from tests.helpers import check, first, unzip, load_yaml SERVICE_TEMPLATE_PATTERN = re.compile(r".*service-.*?-template.yml") RESOURCE_TEMPLATE_PATTERN = re.compile(r".*resource-(.*?)-template.yml") +ZONE_PARAMS = ("availability_zone_0", "availability_zone_1", "availability_zone_2") + def yaml_files(path): """ @@ -45,6 +51,12 @@ class CloudServiceArchive: :param vf_module: name of Heat module (no path or file extension) :return: The definition of the module as a dict or None if not found """ + if ( + vf_module.endswith(".env") + or vf_module.endswith(".yaml") + or vf_module.endswith(".yml") + ): + vf_module = os.path.splitext(vf_module)[0] groups = self._service.get("topology_template", {}).get("groups", {}) for props in groups.values(): module_label = props.get("properties", {}).get("vf_module_label", "") @@ -90,6 +102,15 @@ class CloudServiceArchive: if props.get("type") == "org.openecomp.groups.VfModule" } + def get_vnf_type(self, module): + """ + Concatenation of service and VF instance name + """ + service_name = self.service_name + instance_name = self.get_vf_module_resource_name(module) + if service_name and instance_name: + return "{}/{} 0".format(service_name, instance_name) + @property def vf_module_resource_names(self): """ @@ -125,8 +146,9 @@ class CloudServiceArchive: def_dir = csar_dir / "Definitions" check( def_dir.exists(), - f"CSAR is invalid. {csar_dir.as_posix()} does not contain a " - f"Definitions directory.", + "CSAR is invalid. {} does not contain a Definitions directory.".format( + csar_dir.as_posix() + ), ) return yaml_files(def_dir) @@ -158,17 +180,13 @@ class CloudServiceArchive: return self._service.get("metadata", {}).get("name") def __repr__(self): - return f"CSAR (path={self.csar_path.name}, name={self.service_name})" + return "CSAR (path={}, name={})".format(self.csar_path.name, self.service_name) def __str__(self): return repr(self) class PreloadEnvironment: - """ - A - """ - def __init__(self, env_dir, parent=None): self.base_dir = Path(env_dir) self.parent = parent @@ -219,7 +237,7 @@ class PreloadEnvironment: return [e for e in all_envs if e.is_leaf] def get_module(self, name): - name = name if name.lower().endswith(".env") else f"{name}.env".lower() + name = name if name.lower().endswith(".env") else "{}.env".format(name).lower() if name not in self.module_names: return {} result = {} @@ -228,6 +246,13 @@ class PreloadEnvironment: for m in (parent_module, self.defaults, module): if m: result.update(m) + if self.csar: + vnf_type = self.csar.get_vnf_type(name) + if vnf_type: + result["vnf-type"] = vnf_type + model_name = self.csar.get_vf_module_model_name(name) + if model_name: + result["vf-module-model-name"] = model_name return result @property @@ -264,4 +289,132 @@ class PreloadEnvironment: return self.base_dir.name def __repr__(self): - return f"PreloadEnvironment(name={self.name})" + return "PreloadEnvironment(name={})".format(self.name) + + +class EnvironmentFilePreloadInstance(AbstractPreloadInstance): + def __init__(self, env: PreloadEnvironment, module_label: str, module_params: dict): + self.module_params = module_params + self._module_label = module_label + self.env = env + self.env_cache = {} + + @property + def flag_incompletes(self) -> bool: + return True + + @property + def preload_basename(self) -> str: + return self.module_label + + @property + def output_dir(self) -> Path: + return self.env.base_dir.joinpath("preloads") + + @property + def module_label(self) -> str: + return self._module_label + + @property + def vf_module_name(self) -> str: + return self.get_param("vf_module_name") + + @property + def vnf_name(self) -> Optional[str]: + return self.get_param("vnf_name") + + @property + def vnf_type(self) -> Optional[str]: + return self.get_param("vnf-type") + + @property + def vf_module_model_name(self) -> Optional[str]: + return self.get_param("vf-module-model-name") + + def get_availability_zone(self, index: int, param_name: str) -> Optional[str]: + return self.get_param(param_name) + + def get_network_name(self, network_role: str, name_param: str) -> Optional[str]: + return self.get_param(name_param) + + def get_subnet_id( + self, network_role: str, ip_version: int, param_name: str + ) -> Optional[str]: + return self.get_param(param_name) + + def get_subnet_name( + self, network_role: str, ip_version: int, param_name: str + ) -> Optional[str]: + # Not supported with env files + return None + + def get_vm_name(self, vm_type: str, index: int, param_name: str) -> Optional[str]: + return self.get_param(param_name, single=True) + + def get_floating_ip( + self, vm_type: str, network_role: str, ip_version: int, param_name: str + ) -> Optional[str]: + return self.get_param(param_name) + + def get_fixed_ip( + self, vm_type: str, network_role: str, ip_version: int, index: int, param: str + ) -> Optional[str]: + return self.get_param(param, single=True) + + def get_vnf_parameter(self, key: str, value: Any) -> Optional[str]: + module_value = self.get_param(key) + return module_value or value + + def get_additional_parameters(self) -> Mapping[str, Any]: + return {} + + def get_param(self, param_name, single=False): + """ + Retrieves the value for the given param if it exists. If requesting a + single item, and the parameter is tied to a list then only one item from + the list will be returned. For each subsequent call with the same parameter + it will iterate/rotate through the values in that list. If single is False + then the full list will be returned. + + :param param_name: name of the parameter + :param single: If True returns single value from lists otherwises the full + list. This has no effect on non-list values + """ + value = self.env_cache.get(param_name) + if not value: + value = self.module_params.get(param_name) + if isinstance(value, list): + value = value.copy() + value.reverse() + self.env_cache[param_name] = value + + if value and single and isinstance(value, list): + result = value.pop() + else: + result = value + return result if result != "CHANGEME" else None + + +class EnvironmentFileDataSource(AbstractPreloadDataSource): + def __init__(self, path: Path): + super().__init__(path) + check(path.is_dir(), f"{path} must be an existing directory") + self.path = path + self.env = PreloadEnvironment(path) + + @classmethod + def get_source_type(cls) -> str: + return "DIR" + + @classmethod + def get_identifier(self) -> str: + return "envfiles" + + @classmethod + def get_name(self) -> str: + return "Environment Files" + + def get_module_preloads(self, module: VnfModule): + for env in self.env.environments: + module_params = env.get_module(module.label) + yield EnvironmentFilePreloadInstance(env, module.label, module_params)