3 from pathlib import Path
4 from typing import Any, Optional, Mapping
6 from cached_property import cached_property
8 from preload.data import AbstractPreloadInstance, AbstractPreloadDataSource
9 from preload.model import VnfModule
10 from tests.helpers import check, first, unzip, load_yaml
12 SERVICE_TEMPLATE_PATTERN = re.compile(r".*service-.*?-template.yml")
13 RESOURCE_TEMPLATE_PATTERN = re.compile(r".*resource-(.*?)-template.yml")
15 ZONE_PARAMS = ("availability_zone_0", "availability_zone_1", "availability_zone_2")
20 Return files that are YAML (end with .yml or .yaml)
22 :param path: Directory path object
23 :return: list of paths to YAML files
27 for p in path.iterdir()
28 if p.is_file() and p.suffix.lower() in (".yml", ".yaml")
32 class CloudServiceArchive:
34 Wrapper to extract information from a CSAR file.
37 def __init__(self, csar_path):
38 self.csar_path = Path(csar_path)
39 with tempfile.TemporaryDirectory() as csar_dir:
40 csar_dir = Path(csar_dir)
41 unzip(self.csar_path, csar_dir)
42 self._service = self._get_service_template(csar_dir)
43 self._resources = self._get_vf_module_resource_templates(csar_dir)
45 def get_vf_module(self, vf_module):
47 Retrieve the VF Module definition from the CSAR for the given heat
48 module name (should not include the file extension - ex: base)
50 :param vf_module: name of Heat module (no path or file extension)
51 :return: The definition of the module as a dict or None if not found
53 groups = self._service.get("topology_template", {}).get("groups", {})
54 for props in groups.values():
55 module_label = props.get("properties", {}).get("vf_module_label", "")
56 if module_label.lower() == vf_module.lower():
60 def get_vf_module_model_name(self, vf_module):
62 Retrieves the vfModuleModelName of the module or None if vf_module is not
63 found (see get_vf_module)
65 :param vf_module: name of Heat module (no path or file extension)
66 :return: The value if vfModuleModelName as string or None if not found
68 module = self.get_vf_module(vf_module)
69 return module.get("metadata", {}).get("vfModuleModelName") if module else None
72 def topology_template(self):
74 Return dict representing the topology_template node of the service
77 return self._service.get("topology_template") or {}
82 Return dict representing the groups node of the service
85 return self.topology_template.get("groups") or {}
90 Returns mapping of group ID to VfModule present in the service template
94 for group_id, props in self.groups.items()
95 if props.get("type") == "org.openecomp.groups.VfModule"
98 def get_vnf_type(self, module):
100 Concatenation of service and VF instance name
102 service_name = self.service_name
103 instance_name = self.get_vf_module_resource_name(module)
104 if service_name and instance_name:
105 return "{}/{}".format(service_name, instance_name)
108 def vf_module_resource_names(self):
110 Returns the resource names for all VfModules (these can be used
111 to find the resource templates as they will be part of the filename)
114 module.get("metadata", {}).get("vfModuleModelName")
115 for module in self.vf_modules.values()
117 return [name.split(".")[0] for name in names if name]
119 def get_vf_module_resource_name(self, vf_module):
121 Retrieves the resource name of the module or None if vf_module is not
122 found (see get_vf_module)
124 :param vf_module: name of Heat module (no path or file extension)
125 :return: The value if resource nae as string or None if not found
127 vf_model_name = self.get_vf_module_model_name(vf_module)
128 if not vf_model_name:
130 resource_name = vf_model_name.split(".")[0]
131 resource = self._resources.get(resource_name, {})
132 return resource.get("metadata", {}).get("name")
135 def _get_definition_files(csar_dir):
137 Returns a list of all files in the CSAR's Definitions directory
139 def_dir = csar_dir / "Definitions"
142 "CSAR is invalid. {} does not contain a Definitions directory.".format(
146 return yaml_files(def_dir)
148 def _get_service_template(self, csar_dir):
150 Returns the service template as a dict. Assumes there is only one.
152 files = map(str, self._get_definition_files(csar_dir))
153 service_template = first(files, SERVICE_TEMPLATE_PATTERN.match)
154 return load_yaml(service_template) if service_template else {}
156 def _get_vf_module_resource_templates(self, csar_dir):
158 Returns a mapping of resource name to resource definition (as a dict)
159 (Only loads resource templates that correspond to VF Modules
161 def_dir = csar_dir / "Definitions"
163 (name, def_dir / "resource-{}-template.yml".format(name))
164 for name in self.vf_module_resource_names
166 return {name: load_yaml(path) for name, path in mapping if path.exists()}
169 def service_name(self):
171 Name of the service (extracted from the service template
173 return self._service.get("metadata", {}).get("name")
176 return "CSAR (path={}, name={})".format(self.csar_path.name, self.service_name)
182 class PreloadEnvironment:
183 def __init__(self, env_dir, parent=None):
184 self.base_dir = Path(env_dir)
186 self._modules = self._load_modules()
187 self._sub_env = self._load_envs()
188 self._defaults = self._load_defaults()
190 def _load_defaults(self):
191 defaults = self.base_dir / "defaults.yaml"
192 return load_yaml(defaults) if defaults.exists() else {}
194 def _load_modules(self):
197 for p in self.base_dir.iterdir()
198 if p.is_file() and p.suffix.lower().endswith(".env")
200 return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
202 def _load_envs(self):
204 p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
206 return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
210 csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
212 return CloudServiceArchive(csar_path)
214 return self.parent.csar if self.parent else None
220 result.update(self.parent.defaults)
221 result.update(self._defaults)
225 def environments(self):
227 for env in self._sub_env.values():
229 all_envs.extend(env.environments)
230 return [e for e in all_envs if e.is_leaf]
232 def get_module(self, name):
233 name = name if name.lower().endswith(".env") else "{}.env".format(name).lower()
234 if name not in self.module_names:
237 parent_module = self.parent.get_module(name) if self.parent else None
238 module = self._modules.get(name)
239 for m in (parent_module, self.defaults, module):
243 vnf_type = self.csar.get_vnf_type(name)
245 result["vnf-type"] = vnf_type
246 model_name = self.csar.get_vf_module_model_name(name)
248 result["vf-module-model-name"] = model_name
252 def module_names(self):
253 parent_modules = self.parent.module_names if self.parent else set()
255 result.update(self._modules.keys())
256 result.update(parent_modules)
261 return {name: self.get_module(name) for name in self.module_names}
263 def get_environment(self, env_name):
264 for name, env in self._sub_env.items():
267 result = env.get_environment(env_name)
274 return self.parent is None
278 return not self._sub_env
282 return self.base_dir.name
285 return "PreloadEnvironment(name={})".format(self.name)
288 class EnvironmentFilePreloadInstance(AbstractPreloadInstance):
290 def __init__(self, env: PreloadEnvironment, module_label: str, module_params: dict):
291 self.module_params = module_params
292 self._module_label = module_label
297 def flag_incompletes(self) -> bool:
301 def preload_basename(self) -> str:
302 return self.module_label
305 def output_dir(self) -> Path:
306 return self.env.base_dir.joinpath("preloads")
309 def module_label(self) -> str:
310 return self._module_label
313 def vf_module_name(self) -> str:
314 return self.get_param("vf_module_name")
317 def vnf_name(self) -> Optional[str]:
318 return self.get_param("vnf_name")
321 def vnf_type(self) -> Optional[str]:
322 return self.get_param("vnf-type")
325 def vf_module_model_name(self) -> Optional[str]:
326 return self.get_param("vf-module-model-name")
328 def get_availability_zone(self, index: int, param_name: str) -> Optional[str]:
329 return self.get_param(param_name)
331 def get_network_name(self, network_role: str, name_param: str) -> Optional[str]:
332 return self.get_param(name_param)
335 self, network_role: str, ip_version: int, param_name: str
337 return self.get_param(param_name)
340 self, network_role: str, ip_version: int, param_name: str
342 # Not supported with env files
345 def get_vm_name(self, vm_type: str, index: int, param_name: str) -> Optional[str]:
346 return self.get_param(param_name, single=True)
349 self, vm_type: str, network_role: str, ip_version: int, param_name: str
351 return self.get_param(param_name)
354 self, vm_type: str, network_role: str, ip_version: int, index: int, param: str
356 return self.get_param(param, single=True)
358 def get_vnf_parameter(self, key: str, value: Any) -> Optional[str]:
359 module_value = self.get_param(key)
360 return module_value or value
362 def get_additional_parameters(self) -> Mapping[str, Any]:
365 def get_param(self, param_name, single=False):
367 Retrieves the value for the given param if it exists. If requesting a
368 single item, and the parameter is tied to a list then only one item from
369 the list will be returned. For each subsequent call with the same parameter
370 it will iterate/rotate through the values in that list. If single is False
371 then the full list will be returned.
373 :param param_name: name of the parameter
374 :param single: If True returns single value from lists otherwises the full
375 list. This has no effect on non-list values
377 value = self.env_cache.get(param_name)
379 value = self.module_params.get(param_name)
380 if isinstance(value, list):
383 self.env_cache[param_name] = value
385 if value and single and isinstance(value, list):
389 return result if result != "CHANGEME" else None
392 class EnvironmentFileDataSource(AbstractPreloadDataSource):
394 def __init__(self, path: Path):
395 super().__init__(path)
396 check(path.is_dir(), f"{path} must be an existing directory")
398 self.env = PreloadEnvironment(path)
401 def get_source_type(cls) -> str:
405 def get_identifier(self) -> str:
409 def get_name(self) -> str:
410 return "Environment Files"
412 def get_module_preloads(self, module: VnfModule):
413 for env in self.env.environments:
414 module_params = env.get_module(module.label)
415 yield EnvironmentFilePreloadInstance(env, module.label, module_params)