ice_validator/tests/structures.py

   1 # -*- coding: utf8 -*-
   2 # ============LICENSE_START====================================================
   3 # org.onap.vvp/validation-scripts
   4 # ===================================================================
   5 # Copyright © 2019 AT&T Intellectual Property. All rights reserved.
   6 # ===================================================================
   7 #
   8 # Unless otherwise specified, all software contained herein is licensed
   9 # under the Apache License, Version 2.0 (the "License");
  10 # you may not use this software except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 #             http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 #
  21 #
  22 #
  23 # Unless otherwise specified, all documentation contained herein is licensed
  24 # under the Creative Commons License, Attribution 4.0 Intl. (the "License");
  25 # you may not use this documentation except in compliance with the License.
  26 # You may obtain a copy of the License at
  27 #
  28 #             https://creativecommons.org/licenses/by/4.0/
  29 #
  30 # Unless required by applicable law or agreed to in writing, documentation
  31 # distributed under the License is distributed on an "AS IS" BASIS,
  32 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  33 # See the License for the specific language governing permissions and
  34 # limitations under the License.
  35 #
  36 # ============LICENSE_END============================================
  37 #
  38 #
  39 """structures
  40 """
  41 import collections
  42 import inspect
  43 import os
  44 import re
  45 import sys
  46
  47 from tests import cached_yaml as yaml
  48 from tests.helpers import load_yaml, get_param
  49 from .utils import nested_dict
  50
  51 VERSION = "4.2.0"
  52
  53 # key = pattern, value = regex compiled from pattern
  54 _REGEX_CACHE = {}
  55
  56
  57 def _get_regex(pattern):
  58     """Return a compiled version of pattern.
  59     Keep result in _REGEX_CACHE to avoid re-compiling.
  60     """
  61     regex = _REGEX_CACHE.get(pattern, None)
  62     if regex is None:
  63         regex = re.compile(pattern)
  64         _REGEX_CACHE[pattern] = regex
  65     return regex
  66
  67
  68 class Hashabledict(dict):
  69     """A hashable dict.
  70     dicts with the same keys and whose keys have the same values
  71     are assigned the same hash.
  72     """
  73
  74     def __hash__(self):
  75         return hash((frozenset(self), frozenset(self.values())))
  76
  77
  78 class HeatProcessor(object):
  79     """base class for xxxx::xxxx::xxxx processors
  80     """
  81
  82     resource_type = None  # string 'xxxx::xxxx::xxxx'
  83     re_rids = collections.OrderedDict()  # OrderedDict of name: regex
  84     # name is a string to name the regex.
  85     # regex parses the proper resource id format.
  86
  87     @staticmethod
  88     def get_param_value(value, withIndex=False):
  89         """Return get_param value of `value`
  90         """
  91         if isinstance(value, dict) and len(value) == 1:
  92             v = value.get("get_param")
  93             if isinstance(v, list) and v:
  94                 if withIndex and len(v) > 1:
  95                     idx = v[1]
  96                     if isinstance(idx, dict):
  97                         idx = idx.get("get_param", idx)
  98                     v = "{}{}".format(v[0], idx)
  99                 else:
 100                     v = v[0]
 101         else:
 102             v = None
 103         return v
 104
 105     @classmethod
 106     def get_resource_or_param_value(cls, value):
 107         """Return the get_resource or get_param value of `value`
 108         """
 109         if isinstance(value, dict) and len(value) == 1:
 110             v = value.get("get_resource") or cls.get_param_value(value)
 111         else:
 112             v = None
 113         return v
 114
 115     @classmethod
 116     def get_rid_match_tuple(cls, rid):
 117         """find the first regex matching `rid` and return the tuple
 118         (name, match object) or ('', None) if no match.
 119         """
 120         rid = "" if rid is None else rid
 121         for name, regex in cls.re_rids.items():
 122             match = regex.match(rid)
 123             if match:
 124                 return name, match
 125         return "", None
 126
 127     @classmethod
 128     def get_rid_patterns(cls):
 129         """Return OrderedDict of name: friendly regex.pattern
 130         "friendly" means the group notation is replaced with
 131         braces, and the trailing "$" is removed.
 132
 133         NOTE
 134         nested parentheses in any rid_pattern will break this parser.
 135         The final character is ASSUMED to be a dollar sign.
 136         """
 137         friendly_pattern = _get_regex(r"\(\?P<(.*?)>.*?\)")
 138         rid_patterns = collections.OrderedDict()
 139         for name, regex in cls.re_rids.items():
 140             rid_patterns[name] = friendly_pattern.sub(
 141                 r"{\1}", regex.pattern  # replace groups with braces
 142             )[
 143                 :-1
 144             ]  # remove trailing $
 145         return rid_patterns
 146
 147     @classmethod
 148     def get_str_replace_name(cls, resource_dict):
 149         """Return the name modified by str_replace of `resource_dict`,
 150         a resource (i.e. a value in some template's resources).
 151         Return None, if there is no name, str_replace, its template,
 152         or any missing parameters.
 153         """
 154         str_replace = Heat.nested_get(
 155             resource_dict, "properties", "name", "str_replace"
 156         )
 157         if not str_replace:
 158             return None
 159         template = Heat.nested_get(str_replace, "template")
 160         if not isinstance(template, str):
 161             return None
 162         params = Heat.nested_get(str_replace, "params", default={})
 163         if not isinstance(params, dict):
 164             return None
 165         # WARNING
 166         # The user must choose non-overlapping keys for params since they
 167         # are replaced in the template in arbitrary order.
 168         name = template
 169         for key, value in params.items():
 170             param = cls.get_param_value(value, withIndex=True)
 171             if param is None:
 172                 return None
 173             name = name.replace(key, str(param))
 174         return name
 175
 176
 177 class CinderVolumeAttachmentProcessor(HeatProcessor):
 178     """ Cinder VolumeAttachment
 179     """
 180
 181     resource_type = "OS::Cinder::VolumeAttachment"
 182
 183     @classmethod
 184     def get_config(cls, resources):
 185         """Return a tuple (va_config, va_count)
 186         va_config - Hashabledict of Cinder Volume Attachment config
 187                     indexed by rid.
 188         va_count - dict of attachment counts indexed by rid.
 189         """
 190         va_count = collections.defaultdict(int)
 191         va_config = Hashabledict()
 192         for resource in resources.values():
 193             resource_type = nested_dict.get(resource, "type")
 194             if resource_type == cls.resource_type:
 195                 config, rids = cls.get_volume_attachment_config(resource)
 196                 for rid in rids:
 197                     va_config[rid] = config
 198                     va_count[rid] += 1
 199         return va_config, va_count
 200
 201     @classmethod
 202     def get_volume_attachment_config(cls, resource):
 203         """Returns the cinder volume attachment configuration
 204         of `resource` as a tuple (config, rids)
 205         where:
 206         - config is a Hashabledict whose keys are the keys of the
 207             properties of resource, and whose values are the
 208             corresponding property values (nova server resource ids)
 209             replaced with the vm-type they reference.
 210         - rids is the set of nova server resource ids referenced by
 211             the property values.
 212         """
 213         config = Hashabledict()
 214         rids = set()
 215         for key, value in (resource.get("properties") or {}).items():
 216             rid = cls.get_resource_or_param_value(value)
 217             if rid:
 218                 name, match = NovaServerProcessor.get_rid_match_tuple(rid)
 219                 if name == "server":
 220                     vm_type = match.groupdict()["vm_type"]
 221                     config[key] = vm_type
 222                     rids.add(rid)
 223         return config, rids
 224
 225
 226 class ContrailV2NetworkFlavorBaseProcessor(HeatProcessor):
 227     """ContrailV2 objects which have network_flavor
 228     """
 229
 230     network_flavor_external = "external"
 231     network_flavor_internal = "internal"
 232     network_flavor_subint = "subinterface"
 233
 234     @classmethod
 235     def get_network_flavor(cls, resource):
 236         """Return the network flavor of resource, one of
 237         "internal" - get_resource, or get_param contains _int_
 238         "subint" - get_param contains _subint_
 239         "external" - otherwise
 240         None - no parameters found to decide the flavor.
 241
 242         resource.properties.virtual_network_refs should be a list.
 243         All the parameters in the list should have the same "flavor"
 244         so the flavor is determined from the first item.
 245         """
 246         network_flavor = None
 247         network_refs = nested_dict.get(resource, "properties", "virtual_network_refs")
 248         if network_refs and isinstance(network_refs, list):
 249             param = network_refs[0]
 250             if isinstance(param, dict):
 251                 if "get_resource" in param:
 252                     network_flavor = cls.network_flavor_internal
 253                 else:
 254                     p = param.get("get_param")
 255                     network_flavor = cls.get_network_format(p)
 256         return network_flavor
 257
 258     @classmethod
 259     def get_network_format(cls, param):
 260         if isinstance(param, str):
 261             if "_int_" in param or param.startswith("int_"):
 262                 return cls.network_flavor_internal
 263             elif "_subint_" in param:
 264                 return cls.network_flavor_subint
 265             else:
 266                 return cls.network_flavor_external
 267
 268
 269 class ContrailV2InstanceIpProcessor(ContrailV2NetworkFlavorBaseProcessor):
 270     """ ContrailV2 InstanceIp
 271     """
 272
 273     resource_type = "OS::ContrailV2::InstanceIp"
 274     re_rids = collections.OrderedDict(
 275         [
 276             (
 277                 "internal",
 278                 _get_regex(
 279                     r"(?P<vm_type>.+)"
 280                     r"_(?P<vm_type_index>\d+)"
 281                     r"_int"
 282                     r"_(?P<network_role>.+)"
 283                     r"_vmi"
 284                     r"_(?P<vmi_index>\d+)"
 285                     r"(_v6)?"
 286                     r"_IP"
 287                     r"_(?P<index>\d+)"
 288                     r"$"
 289                 ),
 290             ),
 291             (
 292                 "subinterface",
 293                 _get_regex(
 294                     r"(?P<vm_type>.+)"
 295                     r"_(?P<vm_type_index>\d+)"
 296                     r"_subint"
 297                     r"_(?P<network_role>.+)"
 298                     r"_vmi"
 299                     r"_(?P<vmi_index>\d+)"
 300                     r"(_v6)?"
 301                     r"_IP"
 302                     r"_(?P<index>\d+)"
 303                     r"$"
 304                 ),
 305             ),
 306             (
 307                 "external",
 308                 _get_regex(
 309                     r"(?P<vm_type>.+)"
 310                     r"_(?P<vm_type_index>\d+)"
 311                     r"_(?P<network_role>.+)"
 312                     r"_vmi"
 313                     r"_(?P<vmi_index>\d+)"
 314                     r"(_v6)?"
 315                     r"_IP"
 316                     r"_(?P<index>\d+)"
 317                     r"$"
 318                 ),
 319             ),
 320         ]
 321     )
 322
 323
 324 class ContrailV2InterfaceRouteTableProcessor(HeatProcessor):
 325     """ ContrailV2 InterfaceRouteTable
 326     """
 327
 328     resource_type = "OS::ContrailV2::InterfaceRouteTable"
 329
 330
 331 class ContrailV2NetworkIpamProcessor(HeatProcessor):
 332     """ ContrailV2 NetworkIpam
 333     """
 334
 335     resource_type = "OS::ContrailV2::NetworkIpam"
 336
 337
 338 class ContrailV2PortTupleProcessor(HeatProcessor):
 339     """ ContrailV2 PortTuple
 340     """
 341
 342     resource_type = "OS::ContrailV2::PortTuple"
 343
 344
 345 class ContrailV2ServiceHealthCheckProcessor(HeatProcessor):
 346     """ ContrailV2 ServiceHealthCheck
 347     """
 348
 349     resource_type = "OS::ContrailV2::ServiceHealthCheck"
 350
 351
 352 class ContrailV2ServiceInstanceProcessor(HeatProcessor):
 353     """ ContrailV2 ServiceInstance
 354     """
 355
 356     resource_type = "OS::ContrailV2::ServiceInstance"
 357
 358
 359 class ContrailV2ServiceInstanceIpProcessor(HeatProcessor):
 360     """ ContrailV2 ServiceInstanceIp
 361     """
 362
 363     resource_type = "OS::ContrailV2::ServiceInstanceIp"
 364
 365
 366 class ContrailV2ServiceTemplateProcessor(HeatProcessor):
 367     """ ContrailV2 ServiceTemplate
 368     """
 369
 370     resource_type = "OS::ContrailV2::ServiceTemplate"
 371
 372
 373 class ContrailV2VirtualMachineInterfaceProcessor(ContrailV2NetworkFlavorBaseProcessor):
 374     """ ContrailV2 Virtual Machine Interface resource
 375     """
 376
 377     resource_type = "OS::ContrailV2::VirtualMachineInterface"
 378     re_rids = collections.OrderedDict(
 379         [
 380             (
 381                 "internal",
 382                 _get_regex(
 383                     r"(?P<vm_type>.+)"
 384                     r"_(?P<vm_type_index>\d+)"
 385                     r"_int"
 386                     r"_(?P<network_role>.+)"
 387                     r"_vmi"
 388                     r"_(?P<vmi_index>\d+)"
 389                     r"$"
 390                 ),
 391             ),
 392             (
 393                 "subinterface",
 394                 _get_regex(
 395                     r"(?P<vm_type>.+)"
 396                     r"_(?P<vm_type_index>\d+)"
 397                     r"_subint"
 398                     r"_(?P<network_role>.+)"
 399                     r"_vmi"
 400                     r"_(?P<vmi_index>\d+)"
 401                     r"$"
 402                 ),
 403             ),
 404             (
 405                 "external",
 406                 _get_regex(
 407                     r"(?P<vm_type>.+)"
 408                     r"_(?P<vm_type_index>\d+)"
 409                     r"_(?P<network_role>.+)"
 410                     r"_vmi"
 411                     r"_(?P<vmi_index>\d+)"
 412                     r"$"
 413                 ),
 414             ),
 415         ]
 416     )
 417
 418
 419 class ContrailV2VirtualNetworkProcessor(HeatProcessor):
 420     """ ContrailV2 VirtualNetwork
 421     """
 422
 423     resource_type = "OS::ContrailV2::VirtualNetwork"
 424     re_rids = collections.OrderedDict(
 425         [
 426             ("network", _get_regex(r"int" r"_(?P<network_role>.+)" r"_network" r"$")),
 427             # ("rvn", _get_regex(r"int" r"_(?P<network_role>.+)" r"_RVN" r"$")),
 428         ]
 429     )
 430
 431
 432 class HeatResourceGroupProcessor(HeatProcessor):
 433     """ Heat ResourceGroup
 434     """
 435
 436     resource_type = "OS::Heat::ResourceGroup"
 437     re_rids = collections.OrderedDict(
 438         [
 439             (
 440                 "subint",
 441                 _get_regex(
 442                     r"(?P<vm_type>.+)"
 443                     r"_(?P<vm_type_index>\d+)"
 444                     r"_subint"
 445                     r"_(?P<network_role>.+)"
 446                     r"_port_(?P<port_index>\d+)"
 447                     r"_subinterfaces"
 448                     r"$"
 449                 ),
 450             )
 451         ]
 452     )
 453
 454
 455 class NeutronNetProcessor(HeatProcessor):
 456     """ Neutron Net resource
 457     """
 458
 459     resource_type = "OS::Neutron::Net"
 460     re_rids = collections.OrderedDict(
 461         [("network", _get_regex(r"int" r"_(?P<network_role>.+)" r"_network" r"$"))]
 462     )
 463
 464
 465 class NeutronPortProcessor(HeatProcessor):
 466     """ Neutron Port resource
 467     """
 468
 469     resource_type = "OS::Neutron::Port"
 470     re_rids = collections.OrderedDict(
 471         [
 472             (
 473                 "internal",
 474                 _get_regex(
 475                     r"(?P<vm_type>.+)"
 476                     r"_(?P<vm_type_index>\d+)"
 477                     r"_int"
 478                     r"_(?P<network_role>.+)"
 479                     r"_port_(?P<port_index>\d+)"
 480                     r"$"
 481                 ),
 482             ),
 483             (
 484                 "external",
 485                 _get_regex(
 486                     r"(?P<vm_type>.+)"
 487                     r"_(?P<vm_type_index>\d+)"
 488                     r"_(?P<network_role>.+)"
 489                     r"_port_(?P<port_index>\d+)"
 490                     r"$"
 491                 ),
 492             ),
 493         ]
 494     )
 495
 496     @classmethod
 497     def uses_sr_iov(cls, resource):
 498         """Returns True/False as `resource` is/not
 499         An OS::Nova:Port with the property binding:vnic_type
 500         """
 501         resource_properties = nested_dict.get(resource, "properties", default={})
 502         if (
 503             nested_dict.get(resource, "type") == cls.resource_type
 504             and resource_properties.get("binding:vnic_type", "") == "direct"
 505         ):
 506             return True
 507
 508         return False
 509
 510
 511 class NovaServerProcessor(HeatProcessor):
 512     """ Nova Server resource
 513     """
 514
 515     resource_type = "OS::Nova::Server"
 516     re_rids = collections.OrderedDict(
 517         [
 518             (
 519                 "server",
 520                 _get_regex(r"(?P<vm_type>.+)" r"_server_(?P<vm_type_index>\d+)" r"$"),
 521             )
 522         ]
 523     )
 524
 525     @classmethod
 526     def get_flavor(cls, resource):
 527         """Return the flavor property of `resource`
 528         """
 529         return cls.get_param_value(nested_dict.get(resource, "properties", "flavor"))
 530
 531     @classmethod
 532     def get_image(cls, resource):
 533         """Return the image property of `resource`
 534         """
 535         return cls.get_param_value(nested_dict.get(resource, "properties", "image"))
 536
 537     @classmethod
 538     def get_network(cls, resource):
 539         """Return the network configuration of `resource` as a
 540         frozenset of network-roles.
 541         """
 542         network = set()
 543         networks = nested_dict.get(resource, "properties", "networks")
 544         if isinstance(networks, list):
 545             for port in networks:
 546                 value = cls.get_resource_or_param_value(nested_dict.get(port, "port"))
 547                 name, match = NeutronPortProcessor.get_rid_match_tuple(value)
 548                 if name:
 549                     network_role = match.groupdict().get("network_role")
 550                     if network_role:
 551                         network.add(network_role)
 552         return frozenset(network)
 553
 554     @classmethod
 555     def get_vm_class(cls, resource):
 556         """Return the vm_class of `resource`, a Hashabledict (of
 557         hashable values) whose keys are only the required keys.
 558         Return empty Hashabledict if `resource` is not a NovaServer.
 559         """
 560         vm_class = Hashabledict()
 561         resource_type = nested_dict.get(resource, "type")
 562         if resource_type == cls.resource_type:
 563             d = dict(
 564                 flavor=cls.get_flavor(resource),
 565                 image=cls.get_image(resource),
 566                 network_role=cls.get_network(resource),
 567             )
 568             if all(d.values()):
 569                 vm_class.update(d)
 570         return vm_class
 571
 572
 573 class Heat(object):
 574     """A Heat template.
 575     filepath - absolute path to template file.
 576     envpath - absolute path to environmnt file.
 577     """
 578
 579     type_bool = "boolean"
 580     type_boolean = "boolean"
 581     type_cdl = "comma_delimited_list"
 582     type_comma_delimited_list = "comma_delimited_list"
 583     type_json = "json"
 584     type_num = "number"
 585     type_number = "number"
 586     type_str = "string"
 587     type_string = "string"
 588
 589     def __init__(self, filepath=None, envpath=None):
 590         self.filepath = None
 591         self.basename = None
 592         self.dirname = None
 593         self.yml = None
 594         self.heat_template_version = None
 595         self.description = None
 596         self.parameter_groups = None
 597         self.parameters = None
 598         self.resources = None
 599         self.outputs = None
 600         self.conditions = None
 601         if filepath:
 602             self.load(filepath)
 603         self.env = None
 604         if envpath:
 605             self.load_env(envpath)
 606         self.heat_processors = self.get_heat_processors()
 607
 608     @property
 609     def is_heat(self):
 610         return "heat_template_version" in self.yml
 611
 612     @property
 613     def contrail_resources(self):
 614         """This attribute is a dict of Contrail resources.
 615         """
 616         return self.get_resource_by_type(
 617             resource_type=ContrailV2VirtualMachineInterfaceProcessor.resource_type
 618         )
 619
 620     def get_all_resources(self, base_dir=None, count=1):
 621         """
 622         Like ``resources``, but this returns all the resources definitions
 623         defined in the template, resource groups, and nested YAML files.
 624
 625         A special variable will be added to all resource properties (__count__).
 626         This will normally be 1, but if the resource is generated by a
 627         ResourceGroup **and** an env file is present, then the count will be
 628         the value from the env file (assuming this follows standard VNF Heat
 629         Guidelines)
 630         """
 631         base_dir = base_dir or self.dirname
 632         resources = {}
 633         for r_id, r_data in self.resources.items():
 634             r_data["__count__"] = count
 635             resources[r_id] = r_data
 636             resource = Resource(r_id, r_data)
 637             if resource.is_nested():
 638                 nested_count = resource.get_count(self.env)
 639                 nested = Heat(os.path.join(base_dir, resource.get_nested_filename()))
 640                 nested_resources = nested.get_all_resources(count=nested_count)
 641                 resources.update(nested_resources)
 642         return resources
 643
 644     @staticmethod
 645     def get_heat_processors():
 646         """Return a dict, key is resource_type, value is the
 647         HeatProcessor subclass whose resource_type is the key.
 648         """
 649         return _HEAT_PROCESSORS
 650
 651     def get_resource_by_type(self, resource_type, all_resources=False):
 652         """Return dict of resources whose type is `resource_type`.
 653         key is resource_id, value is resource.
 654         """
 655         resources = self.get_all_resources() if all_resources else self.resources
 656         return {
 657             rid: resource
 658             for rid, resource in resources.items()
 659             if self.nested_get(resource, "type") == resource_type
 660         }
 661
 662     def get_rid_match_tuple(self, rid, resource_type):
 663         """return get_rid_match_tuple(rid) called on the class
 664         corresponding to the given resource_type.
 665         """
 666         processor = self.heat_processors.get(resource_type, HeatProcessor)
 667         return processor.get_rid_match_tuple(rid)
 668
 669     def get_vm_type(self, rid, resource=None):
 670         """return the vm_type
 671         """
 672         if resource is None:
 673             resource = self
 674         resource_type = self.nested_get(resource, "type")
 675         match = self.get_rid_match_tuple(rid, resource_type)[1]
 676         vm_type = match.groupdict().get("vm_type") if match else None
 677         return vm_type
 678
 679     def load(self, filepath):
 680         """Load the Heat template given a filepath.
 681         """
 682         self.filepath = filepath
 683         self.basename = os.path.basename(self.filepath)
 684         self.dirname = os.path.dirname(self.filepath)
 685         with open(self.filepath) as fi:
 686             self.yml = yaml.load(fi)
 687         self.heat_template_version = self.yml.get("heat_template_version", None)
 688         self.description = self.yml.get("description", "")
 689         self.parameter_groups = self.yml.get("parameter_groups") or {}
 690         self.parameters = self.yml.get("parameters") or {}
 691         self.resources = self.yml.get("resources") or {}
 692         self.outputs = self.yml.get("outputs") or {}
 693         self.conditions = self.yml.get("conditions") or {}
 694
 695     def load_env(self, envpath):
 696         """Load the Environment template given a envpath.
 697         """
 698         self.env = Env(filepath=envpath)
 699
 700     @staticmethod
 701     def nested_get(dic, *keys, **kwargs):
 702         """make utils.nested_dict.get available as a class method.
 703         """
 704         return nested_dict.get(dic, *keys, **kwargs)
 705
 706     @property
 707     def neutron_port_resources(self):
 708         """This attribute is a dict of Neutron Ports
 709         """
 710         return self.get_resource_by_type(
 711             resource_type=NeutronPortProcessor.resource_type
 712         )
 713
 714     @property
 715     def nova_server_resources(self):
 716         """This attribute is a dict of Nova Servers
 717         """
 718         return self.get_resource_by_type(
 719             resource_type=NovaServerProcessor.resource_type
 720         )
 721
 722     @staticmethod
 723     def part_is_in_name(part, name):
 724         """
 725         Return True if any of
 726         - name starts with part + '_'
 727         - name contains '_' + part + '_'
 728         - name ends with '_' + part
 729         False otherwise
 730         """
 731         return bool(
 732             re.search("(^(%(x)s)_)|(_(%(x)s)_)|(_(%(x)s)$)" % dict(x=part), name)
 733         )
 734
 735
 736 class Env(Heat):
 737     """An Environment file
 738     """
 739
 740     pass
 741
 742
 743 class Resource(object):
 744     """A Resource
 745     """
 746
 747     def __init__(self, resource_id=None, resource=None):
 748         self.resource_id = resource_id or ""
 749         self.resource = resource or {}
 750         self.properties = self.resource.get("properties", {})
 751         self.resource_type = self.resource.get("type", "")
 752
 753     @staticmethod
 754     def get_index_var(resource):
 755         """Return the index_var for this resource.
 756         """
 757         index_var = nested_dict.get(resource, "properties", "index_var") or "index"
 758         return index_var
 759
 760     def get_nested_filename(self):
 761         """Returns the filename of the nested YAML file if the
 762         resource is a nested YAML or ResourceDef, returns '' otherwise."""
 763         typ = self.resource.get("type", "")
 764         if typ == "OS::Heat::ResourceGroup":
 765             rd = nested_dict.get(self.resource, "properties", "resource_def")
 766             typ = rd.get("type", "") if rd else ""
 767         ext = os.path.splitext(typ)[1]
 768         ext = ext.lower()
 769         if ext == ".yml" or ext == ".yaml":
 770             return typ
 771         else:
 772             return ""
 773
 774     def get_nested_properties(self):
 775         """
 776         Returns {} if not nested
 777         Returns resource: properties if nested
 778         Returns resource: properties: resource_def: properties if RG
 779         """
 780         if not bool(self.get_nested_filename()):
 781             return {}
 782         elif self.resource_type == "OS::Heat::ResourceGroup":
 783             return nested_dict.get(
 784                 self.properties, "resource_def", "properties", default={}
 785             )
 786         else:
 787             return self.properties
 788
 789     def get_count(self, env):
 790         if self.resource_type == "OS::Heat::ResourceGroup":
 791             if not env:
 792                 return 1
 793             env_params = env.parameters
 794             count_param = get_param(self.properties["count"])
 795             count_value = env_params.get(count_param) if count_param else 1
 796             try:
 797                 return int(count_value)
 798             except (ValueError, TypeError):
 799                 print(
 800                     (
 801                         "WARNING: Invalid value for count parameter {}. Expected "
 802                         "an integer, but got {}. Defaulting to 1"
 803                     ).format(count_param, count_value)
 804                 )
 805         return 1
 806
 807     @property
 808     def depends_on(self):
 809         """
 810         Returns the list of resources this resource depends on.  Always
 811         returns a list.
 812
 813         :return: list of all resource IDs this resource depends on.  If none,
 814                  then returns an empty list
 815         """
 816         parents = self.resource.get("depends_on", [])
 817         return parents if isinstance(parents, list) else [parents]
 818
 819     def is_nested(self):
 820         """Returns True if the resource represents a Nested YAML resource
 821         using either type: {filename} or ResourceGroup -> resource_def"""
 822         return bool(self.get_nested_filename())
 823
 824     def get_nested_yaml(self, base_dir):
 825         """If the resource represents a Nested YAML resource, then it
 826         returns the loaded YAML.  If the resource is not nested or the
 827         file cannot be found, then an empty dict is returned"""
 828         filename = self.get_nested_filename()
 829         if filename:
 830             file_path = os.path.join(base_dir, filename)
 831             return load_yaml(file_path) if os.path.exists(file_path) else {}
 832         else:
 833             return {}
 834
 835
 836 def get_all_resources(yaml_files):
 837     """Return a dict, resource id: resource
 838     of the union of resources across all files.
 839     """
 840     resources = {}
 841     for heat_template in yaml_files:
 842         heat = Heat(filepath=heat_template)
 843         dirname = os.path.dirname(heat_template)
 844         resources.update(heat.get_all_resources(dirname))
 845     return resources
 846
 847
 848 def _get_heat_processors():
 849     """Introspect this module and return a
 850     dict of all HeatProcessor sub-classes with a (True) resource_type.
 851     Key is the resource_type, value is the corrresponding class.
 852     """
 853     mod_classes = inspect.getmembers(sys.modules[__name__], inspect.isclass)
 854     heat_processors = {
 855         c.resource_type: c
 856         for _, c in mod_classes
 857         if issubclass(c, HeatProcessor) and c.resource_type
 858     }
 859     return heat_processors
 860
 861
 862 _HEAT_PROCESSORS = _get_heat_processors()