ice_validator/preload/environment.py

   1 import re
   2 import tempfile
   3 from pathlib import Path
   4
   5 from cached_property import cached_property
   6
   7 from tests.helpers import check, first, unzip, load_yaml
   8
   9 SERVICE_TEMPLATE_PATTERN = re.compile(r".*service-.*?-template.yml")
  10 RESOURCE_TEMPLATE_PATTERN = re.compile(r".*resource-(.*?)-template.yml")
  11
  12
  13 def yaml_files(path):
  14     """
  15     Return files that are YAML (end with .yml or .yaml)
  16
  17     :param path: Directory path object
  18     :return: list of paths to YAML files
  19     """
  20     return [
  21         p
  22         for p in path.iterdir()
  23         if p.is_file() and p.suffix.lower() in (".yml", ".yaml")
  24     ]
  25
  26
  27 class CloudServiceArchive:
  28     """
  29     Wrapper to extract information from a CSAR file.
  30     """
  31
  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)
  39
  40     def get_vf_module(self, vf_module):
  41         """
  42         Retrieve the VF Module definition from the CSAR for the given heat
  43         module name (should not include the file extension - ex: base)
  44
  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
  47         """
  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():
  52                 return props
  53         return None
  54
  55     def get_vf_module_model_name(self, vf_module):
  56         """
  57         Retrieves the vfModuleModelName of the module or None if vf_module is not
  58         found (see get_vf_module)
  59
  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
  62         """
  63         module = self.get_vf_module(vf_module)
  64         return module.get("metadata", {}).get("vfModuleModelName") if module else None
  65
  66     @property
  67     def topology_template(self):
  68         """
  69         Return dict representing the topology_template node of the service
  70         template
  71         """
  72         return self._service.get("topology_template") or {}
  73
  74     @property
  75     def groups(self):
  76         """
  77         Return dict representing the groups node of the service
  78         template
  79         """
  80         return self.topology_template.get("groups") or {}
  81
  82     @property
  83     def vf_modules(self):
  84         """
  85         Returns mapping of group ID to VfModule present in the service template
  86         """
  87         return {
  88             group_id: props
  89             for group_id, props in self.groups.items()
  90             if props.get("type") == "org.openecomp.groups.VfModule"
  91         }
  92
  93     @property
  94     def vf_module_resource_names(self):
  95         """
  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)
  98         """
  99         names = (
 100             module.get("metadata", {}).get("vfModuleModelName")
 101             for module in self.vf_modules.values()
 102         )
 103         return [name.split(".")[0] for name in names if name]
 104
 105     def get_vf_module_resource_name(self, vf_module):
 106         """
 107         Retrieves the resource name of the module or None if vf_module is not
 108         found (see get_vf_module)
 109
 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
 112         """
 113         vf_model_name = self.get_vf_module_model_name(vf_module)
 114         if not vf_model_name:
 115             return None
 116         resource_name = vf_model_name.split(".")[0]
 117         resource = self._resources.get(resource_name, {})
 118         return resource.get("metadata", {}).get("name")
 119
 120     @staticmethod
 121     def _get_definition_files(csar_dir):
 122         """
 123         Returns a list of all files in the CSAR's Definitions directory
 124         """
 125         def_dir = csar_dir / "Definitions"
 126         check(
 127             def_dir.exists(),
 128             f"CSAR is invalid. {csar_dir.as_posix()} does not contain a "
 129             f"Definitions directory.",
 130         )
 131         return yaml_files(def_dir)
 132
 133     def _get_service_template(self, csar_dir):
 134         """
 135         Returns the service template as a dict.  Assumes there is only one.
 136         """
 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 {}
 140
 141     def _get_vf_module_resource_templates(self, csar_dir):
 142         """
 143         Returns a mapping of resource name to resource definition (as a dict)
 144         (Only loads resource templates that correspond to VF Modules
 145         """
 146         def_dir = csar_dir / "Definitions"
 147         mapping = (
 148             (name, def_dir / "resource-{}-template.yml".format(name))
 149             for name in self.vf_module_resource_names
 150         )
 151         return {name: load_yaml(path) for name, path in mapping if path.exists()}
 152
 153     @property
 154     def service_name(self):
 155         """
 156         Name of the service (extracted from the service template
 157         """
 158         return self._service.get("metadata", {}).get("name")
 159
 160     def __repr__(self):
 161         return f"CSAR (path={self.csar_path.name}, name={self.service_name})"
 162
 163     def __str__(self):
 164         return repr(self)
 165
 166
 167 class PreloadEnvironment:
 168     """
 169     A
 170     """
 171
 172     def __init__(self, env_dir, parent=None):
 173         self.base_dir = Path(env_dir)
 174         self.parent = parent
 175         self._modules = self._load_modules()
 176         self._sub_env = self._load_envs()
 177         self._defaults = self._load_defaults()
 178
 179     def _load_defaults(self):
 180         defaults = self.base_dir / "defaults.yaml"
 181         return load_yaml(defaults) if defaults.exists() else {}
 182
 183     def _load_modules(self):
 184         files = [
 185             p
 186             for p in self.base_dir.iterdir()
 187             if p.is_file() and p.suffix.lower().endswith(".env")
 188         ]
 189         return {f.name.lower(): load_yaml(f).get("parameters", {}) for f in files}
 190
 191     def _load_envs(self):
 192         env_dirs = [
 193             p for p in self.base_dir.iterdir() if p.is_dir() and p.name != "preloads"
 194         ]
 195         return {d.name: PreloadEnvironment(d, self) for d in env_dirs}
 196
 197     @cached_property
 198     def csar(self):
 199         csar_path = first(self.base_dir.iterdir(), lambda p: p.suffix == ".csar")
 200         if csar_path:
 201             return CloudServiceArchive(csar_path)
 202         else:
 203             return self.parent.csar if self.parent else None
 204
 205     @property
 206     def defaults(self):
 207         result = {}
 208         if self.parent:
 209             result.update(self.parent.defaults)
 210         result.update(self._defaults)
 211         return result
 212
 213     @property
 214     def environments(self):
 215         all_envs = [self]
 216         for env in self._sub_env.values():
 217             all_envs.append(env)
 218             all_envs.extend(env.environments)
 219         return [e for e in all_envs if e.is_leaf]
 220
 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:
 224             return {}
 225         result = {}
 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):
 229             if m:
 230                 result.update(m)
 231         return result
 232
 233     @property
 234     def module_names(self):
 235         parent_modules = self.parent.module_names if self.parent else set()
 236         result = set()
 237         result.update(self._modules.keys())
 238         result.update(parent_modules)
 239         return result
 240
 241     @property
 242     def modules(self):
 243         return {name: self.get_module(name) for name in self.module_names}
 244
 245     def get_environment(self, env_name):
 246         for name, env in self._sub_env.items():
 247             if name == env_name:
 248                 return env
 249             result = env.get_environment(env_name)
 250             if result:
 251                 return result
 252         return None
 253
 254     @property
 255     def is_base(self):
 256         return self.parent is None
 257
 258     @property
 259     def is_leaf(self):
 260         return not self._sub_env
 261
 262     @property
 263     def name(self):
 264         return self.base_dir.name
 265
 266     def __repr__(self):
 267         return f"PreloadEnvironment(name={self.name})"