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 "CSAR (path={}, name={})".format(self.csar_path.name, self.service_name)
177 class PreloadEnvironment:
178 def __init__(self, env_dir, parent=None):
179 self.base_dir = Path(env_dir)
181 self._modules = self._load_modules()
182 self._sub_env = self._load_envs()
183 self._defaults = self._load_defaults()
185 def _load_defaults(self):
186 defaults = self.base_dir / "defaults.yaml"
187 return load_yaml(defaults) if defaults.exists() else {}
189 def _load_modules(self):
192 for p in self.base_dir.iterdir()
193 if p.is_file() and p.suffix.lower().endswith(".env")
195 return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
197 def _load_envs(self):
199 p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
201 return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
205 csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
207 return CloudServiceArchive(csar_path)
209 return self.parent.csar if self.parent else None
215 result.update(self.parent.defaults)
216 result.update(self._defaults)
220 def environments(self):
222 for env in self._sub_env.values():
224 all_envs.extend(env.environments)
225 return [e for e in all_envs if e.is_leaf]
227 def get_module(self, name):
228 name = name if name.lower().endswith(".env") else "{}.env".format(name).lower()
229 if name not in self.module_names:
232 parent_module = self.parent.get_module(name) if self.parent else None
233 module = self._modules.get(name)
234 for m in (parent_module, self.defaults, module):
238 vnf_type = self.csar.get_vnf_type(name)
240 result["vnf-type"] = vnf_type
241 model_name = self.csar.get_vf_module_model_name(name)
243 result["vf-module-model-name"] = model_name
247 def module_names(self):
248 parent_modules = self.parent.module_names if self.parent else set()
250 result.update(self._modules.keys())
251 result.update(parent_modules)
256 return {name: self.get_module(name) for name in self.module_names}
258 def get_environment(self, env_name):
259 for name, env in self._sub_env.items():
262 result = env.get_environment(env_name)
269 return self.parent is None
273 return not self._sub_env
277 return self.base_dir.name
280 return "PreloadEnvironment(name={})".format(self.name)