3 from pathlib import Path
5 from cached_property import cached_property
7 from tests.helpers import check, first, unzip, load_yaml
9 SERVICE_TEMPLATE_PATTERN = re.compile(r".*service-.*?-template.yml")
10 RESOURCE_TEMPLATE_PATTERN = re.compile(r".*resource-(.*?)-template.yml")
15 Return files that are YAML (end with .yml or .yaml)
17 :param path: Directory path object
18 :return: list of paths to YAML files
22 for p in path.iterdir()
23 if p.is_file() and p.suffix.lower() in (".yml", ".yaml")
27 class CloudServiceArchive:
29 Wrapper to extract information from a CSAR file.
32 def __init__(self, csar_path):
33 self.csar_path = Path(csar_path)
34 with tempfile.TemporaryDirectory() as csar_dir:
35 csar_dir = Path(csar_dir)
36 unzip(self.csar_path, csar_dir)
37 self._service = self._get_service_template(csar_dir)
38 self._resources = self._get_vf_module_resource_templates(csar_dir)
40 def get_vf_module(self, vf_module):
42 Retrieve the VF Module definition from the CSAR for the given heat
43 module name (should not include the file extension - ex: base)
45 :param vf_module: name of Heat module (no path or file extension)
46 :return: The definition of the module as a dict or None if not found
48 groups = self._service.get("topology_template", {}).get("groups", {})
49 for props in groups.values():
50 module_label = props.get("properties", {}).get("vf_module_label", "")
51 if module_label.lower() == vf_module.lower():
55 def get_vf_module_model_name(self, vf_module):
57 Retrieves the vfModuleModelName of the module or None if vf_module is not
58 found (see get_vf_module)
60 :param vf_module: name of Heat module (no path or file extension)
61 :return: The value if vfModuleModelName as string or None if not found
63 module = self.get_vf_module(vf_module)
64 return module.get("metadata", {}).get("vfModuleModelName") if module else None
67 def topology_template(self):
69 Return dict representing the topology_template node of the service
72 return self._service.get("topology_template") or {}
77 Return dict representing the groups node of the service
80 return self.topology_template.get("groups") or {}
85 Returns mapping of group ID to VfModule present in the service template
89 for group_id, props in self.groups.items()
90 if props.get("type") == "org.openecomp.groups.VfModule"
93 def get_vnf_type(self, module):
95 Concatenation of service and VF instance name
97 service_name = self.service_name
98 instance_name = self.get_vf_module_resource_name(module)
99 if service_name and instance_name:
100 return "{}/{}".format(service_name, instance_name)
103 def vf_module_resource_names(self):
105 Returns the resource names for all VfModules (these can be used
106 to find the resource templates as they will be part of the filename)
109 module.get("metadata", {}).get("vfModuleModelName")
110 for module in self.vf_modules.values()
112 return [name.split(".")[0] for name in names if name]
114 def get_vf_module_resource_name(self, vf_module):
116 Retrieves the resource name of the module or None if vf_module is not
117 found (see get_vf_module)
119 :param vf_module: name of Heat module (no path or file extension)
120 :return: The value if resource nae as string or None if not found
122 vf_model_name = self.get_vf_module_model_name(vf_module)
123 if not vf_model_name:
125 resource_name = vf_model_name.split(".")[0]
126 resource = self._resources.get(resource_name, {})
127 return resource.get("metadata", {}).get("name")
130 def _get_definition_files(csar_dir):
132 Returns a list of all files in the CSAR's Definitions directory
134 def_dir = csar_dir / "Definitions"
137 "CSAR is invalid. {} does not contain a Definitions directory.".format(
141 return yaml_files(def_dir)
143 def _get_service_template(self, csar_dir):
145 Returns the service template as a dict. Assumes there is only one.
147 files = map(str, self._get_definition_files(csar_dir))
148 service_template = first(files, SERVICE_TEMPLATE_PATTERN.match)
149 return load_yaml(service_template) if service_template else {}
151 def _get_vf_module_resource_templates(self, csar_dir):
153 Returns a mapping of resource name to resource definition (as a dict)
154 (Only loads resource templates that correspond to VF Modules
156 def_dir = csar_dir / "Definitions"
158 (name, def_dir / "resource-{}-template.yml".format(name))
159 for name in self.vf_module_resource_names
161 return {name: load_yaml(path) for name, path in mapping if path.exists()}
164 def service_name(self):
166 Name of the service (extracted from the service template
168 return self._service.get("metadata", {}).get("name")
171 return f"CSAR (path={self.csar_path.name}, name={self.service_name})"
177 class PreloadEnvironment:
179 def __init__(self, env_dir, parent=None):
180 self.base_dir = Path(env_dir)
182 self._modules = self._load_modules()
183 self._sub_env = self._load_envs()
184 self._defaults = self._load_defaults()
186 def _load_defaults(self):
187 defaults = self.base_dir / "defaults.yaml"
188 return load_yaml(defaults) if defaults.exists() else {}
190 def _load_modules(self):
193 for p in self.base_dir.iterdir()
194 if p.is_file() and p.suffix.lower().endswith(".env")
196 return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
198 def _load_envs(self):
200 p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
202 return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
206 csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
208 return CloudServiceArchive(csar_path)
210 return self.parent.csar if self.parent else None
216 result.update(self.parent.defaults)
217 result.update(self._defaults)
221 def environments(self):
223 for env in self._sub_env.values():
225 all_envs.extend(env.environments)
226 return [e for e in all_envs if e.is_leaf]
228 def get_module(self, name):
229 name = name if name.lower().endswith(".env") else f"{name}.env".lower()
230 if name not in self.module_names:
233 parent_module = self.parent.get_module(name) if self.parent else None
234 module = self._modules.get(name)
235 for m in (parent_module, self.defaults, module):
239 vnf_type = self.csar.get_vnf_type(name)
241 result["vnf-type"] = vnf_type
242 model_name = self.csar.get_vf_module_model_name(name)
244 result["vf-module-model-name"] = model_name
248 def module_names(self):
249 parent_modules = self.parent.module_names if self.parent else set()
251 result.update(self._modules.keys())
252 result.update(parent_modules)
257 return {name: self.get_module(name) for name in self.module_names}
259 def get_environment(self, env_name):
260 for name, env in self._sub_env.items():
263 result = env.get_environment(env_name)
270 return self.parent is None
274 return not self._sub_env
278 return self.base_dir.name
281 return f"PreloadEnvironment(name={self.name})"