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"
94 def vf_module_resource_names(self):
96 Returns the resource names for all VfModules (these can be used
97 to find the resource templates as they will be part of the filename)
100 module.get("metadata", {}).get("vfModuleModelName")
101 for module in self.vf_modules.values()
103 return [name.split(".")[0] for name in names if name]
105 def get_vf_module_resource_name(self, vf_module):
107 Retrieves the resource name of the module or None if vf_module is not
108 found (see get_vf_module)
110 :param vf_module: name of Heat module (no path or file extension)
111 :return: The value if resource nae as string or None if not found
113 vf_model_name = self.get_vf_module_model_name(vf_module)
114 if not vf_model_name:
116 resource_name = vf_model_name.split(".")[0]
117 resource = self._resources.get(resource_name, {})
118 return resource.get("metadata", {}).get("name")
121 def _get_definition_files(csar_dir):
123 Returns a list of all files in the CSAR's Definitions directory
125 def_dir = csar_dir / "Definitions"
128 f"CSAR is invalid. {csar_dir.as_posix()} does not contain a "
129 f"Definitions directory.",
131 return yaml_files(def_dir)
133 def _get_service_template(self, csar_dir):
135 Returns the service template as a dict. Assumes there is only one.
137 files = map(str, self._get_definition_files(csar_dir))
138 service_template = first(files, SERVICE_TEMPLATE_PATTERN.match)
139 return load_yaml(service_template) if service_template else {}
141 def _get_vf_module_resource_templates(self, csar_dir):
143 Returns a mapping of resource name to resource definition (as a dict)
144 (Only loads resource templates that correspond to VF Modules
146 def_dir = csar_dir / "Definitions"
148 (name, def_dir / "resource-{}-template.yml".format(name))
149 for name in self.vf_module_resource_names
151 return {name: load_yaml(path) for name, path in mapping if path.exists()}
154 def service_name(self):
156 Name of the service (extracted from the service template
158 return self._service.get("metadata", {}).get("name")
161 return f"CSAR (path={self.csar_path.name}, name={self.service_name})"
167 class PreloadEnvironment:
172 def __init__(self, env_dir, parent=None):
173 self.base_dir = Path(env_dir)
175 self._modules = self._load_modules()
176 self._sub_env = self._load_envs()
177 self._defaults = self._load_defaults()
179 def _load_defaults(self):
180 defaults = self.base_dir / "defaults.yaml"
181 return load_yaml(defaults) if defaults.exists() else {}
183 def _load_modules(self):
186 for p in self.base_dir.iterdir()
187 if p.is_file() and p.suffix.lower().endswith(".env")
189 return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
191 def _load_envs(self):
193 p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
195 return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
199 csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
201 return CloudServiceArchive(csar_path)
203 return self.parent.csar if self.parent else None
209 result.update(self.parent.defaults)
210 result.update(self._defaults)
214 def environments(self):
216 for env in self._sub_env.values():
218 all_envs.extend(env.environments)
219 return [e for e in all_envs if e.is_leaf]
221 def get_module(self, name):
222 name = name if name.lower().endswith(".env") else f"{name}.env".lower()
223 if name not in self.module_names:
226 parent_module = self.parent.get_module(name) if self.parent else None
227 module = self._modules.get(name)
228 for m in (parent_module, self.defaults, module):
234 def module_names(self):
235 parent_modules = self.parent.module_names if self.parent else set()
237 result.update(self._modules.keys())
238 result.update(parent_modules)
243 return {name: self.get_module(name) for name in self.module_names}
245 def get_environment(self, env_name):
246 for name, env in self._sub_env.items():
249 result = env.get_environment(env_name)
256 return self.parent is None
260 return not self._sub_env
264 return self.base_dir.name
267 return f"PreloadEnvironment(name={self.name})"