#
# ============LICENSE_END============================================
#
-# ECOMP is a trademark and service mark of AT&T Intellectual Property.
-#
-'''structures
-'''
+"""structures
+"""
+import sys
+
+import collections
+import inspect
import os
+import re
from tests import cached_yaml as yaml
+from tests.helpers import load_yaml
from .utils import nested_dict
-VERSION = '1.4.0'
+VERSION = "3.5.0"
+
+# key = pattern, value = regex compiled from pattern
+_REGEX_CACHE = {}
+
+
+def _get_regex(pattern):
+ """Return a compiled version of pattern.
+ Keep result in _REGEX_CACHE to avoid re-compiling.
+ """
+ regex = _REGEX_CACHE.get(pattern, None)
+ if regex is None:
+ regex = re.compile(pattern)
+ _REGEX_CACHE[pattern] = regex
+ return regex
+
+
+class HeatObject(object):
+ """base class for xxxx::xxxx::xxxx objects
+ """
+
+ resource_type = None
+
+ def __init__(self):
+ self.re_rids = self.get_re_rids()
+
+ @staticmethod
+ def get_re_rids():
+ """Return OrderedDict of name: regex
+ Each regex parses the proper format for a given rid
+ (resource id).
+ """
+ return collections.OrderedDict()
+
+ def get_rid_match_tuple(self, rid):
+ """find the first regex matching `rid` and return the tuple
+ (name, match object) or ('', None) if no match.
+ """
+ for name, regex in self.re_rids.items():
+ match = regex.match(rid)
+ if match:
+ return name, match
+ return "", None
+
+ def get_rid_patterns(self):
+ """Return OrderedDict of name: friendly regex.pattern
+ "friendly" means the group notation is replaced with
+ braces, and the trailing "$" is removed.
+
+ NOTE
+ nested parentheses in any rid_pattern will break this parser.
+ The final character is ASSUMED to be a dollar sign.
+ """
+ friendly_pattern = _get_regex(r"\(\?P<(.*?)>.*?\)")
+ rid_patterns = collections.OrderedDict()
+ for name, regex in self.re_rids.items():
+ rid_patterns[name] = friendly_pattern.sub(
+ r"{\1}", regex.pattern # replace groups with braces
+ )[
+ :-1
+ ] # remove trailing $
+ return rid_patterns
+
+
+class ContrailV2NetworkHeatObject(HeatObject):
+ """ContrailV2 objects which have network_flavor
+ """
+
+ network_flavor_external = "external"
+ network_flavor_internal = "internal"
+ network_flavor_subint = "subint"
+
+ def get_network_flavor(self, resource):
+ """Return the network flavor of resource, one of
+ "internal" - get_resource, or get_param contains _int_
+ "subint" - get_param contains _subint_
+ "external" - otherwise
+ None - no parameters found to decide the flavor.
+
+ resource.properties.virtual_network_refs should be a list.
+ All the parameters in the list should have the same "flavor"
+ so the flavor is determined from the first item.
+ """
+ network_flavor = None
+ network_refs = nested_dict.get(resource, "properties", "virtual_network_refs")
+ if network_refs and isinstance(network_refs, list):
+ param = network_refs[0]
+ if isinstance(param, dict):
+ if "get_resource" in param:
+ network_flavor = self.network_flavor_internal
+ else:
+ p = param.get("get_param")
+ if isinstance(p, str):
+ if "_int_" in p or p.startswith("int_"):
+ network_flavor = self.network_flavor_internal
+ elif "_subint_" in p:
+ network_flavor = self.network_flavor_subint
+ else:
+ network_flavor = self.network_flavor_external
+ return network_flavor
+
+
+class ContrailV2InstanceIp(ContrailV2NetworkHeatObject):
+ """ ContrailV2 InstanceIp
+ """
+
+ resource_type = "OS::ContrailV2::InstanceIp"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [
+ (
+ "int_ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_int"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "int_v6_ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_int"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_v6_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "subint_ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_subint"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "subint_v6_ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_subint"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_v6_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "v6_ip",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"_v6_IP"
+ r"_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ ]
+ )
+
+
+class ContrailV2InterfaceRouteTable(HeatObject):
+ """ ContrailV2 InterfaceRouteTable
+ """
+
+ resource_type = "OS::ContrailV2::InterfaceRouteTable"
+
+
+class ContrailV2NetworkIpam(HeatObject):
+ """ ContrailV2 NetworkIpam
+ """
+
+ resource_type = "OS::ContrailV2::NetworkIpam"
+
+
+class ContrailV2PortTuple(HeatObject):
+ """ ContrailV2 PortTuple
+ """
+
+ resource_type = "OS::ContrailV2::PortTuple"
+
+
+class ContrailV2ServiceHealthCheck(HeatObject):
+ """ ContrailV2 ServiceHealthCheck
+ """
+
+ resource_type = "OS::ContrailV2::ServiceHealthCheck"
+
+
+class ContrailV2ServiceInstance(HeatObject):
+ """ ContrailV2 ServiceInstance
+ """
+
+ resource_type = "OS::ContrailV2::ServiceInstance"
+
+
+class ContrailV2ServiceInstanceIp(HeatObject):
+ """ ContrailV2 ServiceInstanceIp
+ """
+
+ resource_type = "OS::ContrailV2::ServiceInstanceIp"
+
+
+class ContrailV2ServiceTemplate(HeatObject):
+ """ ContrailV2 ServiceTemplate
+ """
+
+ resource_type = "OS::ContrailV2::ServiceTemplate"
+
+
+class ContrailV2VirtualMachineInterface(ContrailV2NetworkHeatObject):
+ """ ContrailV2 Virtual Machine Interface resource
+ """
+
+ resource_type = "OS::ContrailV2::VirtualMachineInterface"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [
+ (
+ "vmi_internal",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_int"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "vmi_subint",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_subint"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "vmi_external",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_(?P<network_role>.+)"
+ r"_vmi"
+ r"_(?P<vmi_index>\d+)"
+ r"$"
+ ),
+ ),
+ ]
+ )
+
+
+class ContrailV2VirtualNetwork(HeatObject):
+ """ ContrailV2 VirtualNetwork
+ """
+
+ resource_type = "OS::ContrailV2::VirtualNetwork"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [
+ (
+ "network",
+ _get_regex(r"int" r"_(?P<network_role>.+)" r"_network" r"$"),
+ ),
+ ("rvn", _get_regex(r"int" r"_(?P<network_role>.+)" r"_RVN" r"$")),
+ ]
+ )
+
+
+class NeutronNet(HeatObject):
+ """ Neutron Net resource
+ """
+
+ resource_type = "OS::Neutron::Net"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [("network", _get_regex(r"int" r"_(?P<network_role>.+)" r"_network" r"$"))]
+ )
+
+
+class NeutronPort(HeatObject):
+ """ Neutron Port resource
+ """
+
+ resource_type = "OS::Neutron::Port"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [
+ (
+ "internal_port",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_int"
+ r"_(?P<network_role>.+)"
+ r"_port_(?P<port_index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "port",
+ _get_regex(
+ r"(?P<vm_type>.+)"
+ r"_(?P<vm_type_index>\d+)"
+ r"_(?P<network_role>.+)"
+ r"_port_(?P<port_index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "floating_ip",
+ _get_regex(
+ r"reserve_port"
+ r"_(?P<vm_type>.+)"
+ r"_(?P<network_role>.+)"
+ r"_floating_ip_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ (
+ "floating_v6_ip",
+ _get_regex(
+ r"reserve_port"
+ r"_(?P<vm_type>.+)"
+ r"_(?P<network_role>.+)"
+ r"_floating_v6_ip_(?P<index>\d+)"
+ r"$"
+ ),
+ ),
+ ]
+ )
+
+
+class NovaServer(HeatObject):
+ """ Nova Server resource
+ """
+
+ resource_type = "OS::Nova::Server"
+
+ def get_re_rids(self):
+ """Return OrderedDict of name: regex
+ """
+ return collections.OrderedDict(
+ [
+ (
+ "server",
+ _get_regex(
+ r"(?P<vm_type>.+)" r"_server_(?P<vm_type_index>\d+)" r"$"
+ ),
+ )
+ ]
+ )
class Heat(object):
filepath - absolute path to template file.
envpath - absolute path to environmnt file.
"""
+
+ type_cdl = "comma_delimited_list"
+ type_num = "number"
+ type_str = "string"
+
def __init__(self, filepath=None, envpath=None):
self.filepath = None
self.basename = None
self.env = None
if envpath:
self.load_env(envpath)
+ self.heat_objects = self.get_heat_objects()
+
+ @property
+ def contrail_resources(self):
+ """This attribute is a dict of Contrail resources.
+ """
+ return self.get_resource_by_type(
+ resource_type=ContrailV2VirtualMachineInterface.resource_type
+ )
+
+ @staticmethod
+ def get_heat_objects():
+ """Return a dict, key is resource_type, value is the
+ HeatObject subclass whose resource_type is the key.
+ """
+ return _HEAT_OBJECTS
+
+ def get_resource_by_type(self, resource_type):
+ """Return dict of resources whose type is `resource_type`.
+ key is resource_id, value is resource.
+ """
+ return {
+ rid: resource
+ for rid, resource in self.resources.items()
+ if self.nested_get(resource, "type") == resource_type
+ }
+
+ def get_rid_match_tuple(self, rid, resource_type):
+ """return get_rid_match_tuple(rid) called on the class
+ corresponding to the given resource_type.
+ """
+ hoc = self.heat_objects.get(resource_type, HeatObject)
+ return hoc().get_rid_match_tuple(rid)
+
+ def get_vm_type(self, rid, resource=None):
+ """return the vm_type
+ """
+ if resource is None:
+ resource = self
+ resource_type = self.nested_get(resource, "type")
+ match = self.get_rid_match_tuple(rid, resource_type)[1]
+ vm_type = match.groupdict().get("vm_type") if match else None
+ return vm_type
def load(self, filepath):
"""Load the Heat template given a filepath.
self.dirname = os.path.dirname(self.filepath)
with open(self.filepath) as fi:
self.yml = yaml.load(fi)
- self.heat_template_version = self.yml.get('heat_template_version', None)
- self.description = self.yml.get('description', '')
- self.parameter_groups = self.yml.get('parameter_groups', {})
- self.parameters = self.yml.get('parameters', {})
- self.resources = self.yml.get('resources', {})
- self.outputs = self.yml.get('outputs', {})
- self.conditions = self.yml.get('conditions', {})
+ self.heat_template_version = self.yml.get("heat_template_version", None)
+ self.description = self.yml.get("description", "")
+ self.parameter_groups = self.yml.get("parameter_groups", {})
+ self.parameters = self.yml.get("parameters") or {}
+ self.resources = self.yml.get("resources", {})
+ self.outputs = self.yml.get("outputs", {})
+ self.conditions = self.yml.get("conditions", {})
+
+ def get_all_resources(self, base_dir):
+ """
+ Like ``resources``, but this returns all the resources definitions
+ defined in the template, resource groups, and nested YAML files.
+ """
+ resources = {}
+ for r_id, r_data in self.resources.items():
+ resources[r_id] = r_data
+ resource = Resource(r_id, r_data)
+ if resource.is_nested():
+ nested = Heat(os.path.join(base_dir, resource.get_nested_filename()))
+ resources.update(nested.get_all_resources(base_dir))
+ return resources
def load_env(self, envpath):
- """Load the Environment template given a envpath.
+ """
+ Load the Environment template given a envpath.
"""
self.env = Env(filepath=envpath)
@staticmethod
- def nested_get(dic, *keys):
+ def nested_get(dic, *keys, **kwargs):
"""make utils.nested_dict.get available as a class method.
"""
- return nested_dict.get(dic, *keys)
+ return nested_dict.get(dic, *keys, **kwargs)
+
+ @property
+ def neutron_port_resources(self):
+ """This attribute is a dict of Neutron Ports
+ """
+ return self.get_resource_by_type(resource_type=NeutronPort.resource_type)
+
+ @property
+ def nova_server_resources(self):
+ """This attribute is a dict of Nova Servers
+ """
+ return self.get_resource_by_type(resource_type=NovaServer.resource_type)
+
+ @staticmethod
+ def part_is_in_name(part, name):
+ """
+ Return True if any of
+ - name starts with part + '_'
+ - name contains '_' + part + '_'
+ - name ends with '_' + part
+ False otherwise
+ """
+ return bool(
+ re.search("(^(%(x)s)_)|(_(%(x)s)_)|(_(%(x)s)$)" % dict(x=part), name)
+ )
class Env(Heat):
"""An Environment file
"""
+
pass
class Resource(object):
"""A Resource
"""
+
def __init__(self, resource_id=None, resource=None):
- self.resource_id = resource_id or ''
+ self.resource_id = resource_id or ""
self.resource = resource or {}
+ self.properties = self.resource.get("properties", {})
+ self.resource_type = resource.get("type", "")
@staticmethod
def get_index_var(resource):
"""Return the index_var for this resource.
"""
- index_var = nested_dict.get(resource,
- 'properties',
- 'index_var') or 'index'
+ index_var = nested_dict.get(resource, "properties", "index_var") or "index"
return index_var
+ def get_nested_filename(self):
+ """Returns the filename of the nested YAML file if the
+ resource is a nested YAML or ResourceDef, returns '' otherwise."""
+ typ = self.resource.get("type", "")
+ if typ == "OS::Heat::ResourceGroup":
+ rd = nested_dict.get(self.resource, "properties", "resource_def")
+ typ = rd.get("type", "") if rd else ""
+ ext = os.path.splitext(typ)[1]
+ ext = ext.lower()
+ if ext == ".yml" or ext == ".yaml":
+ return typ
+ else:
+ return ""
+
+ def get_nested_properties(self):
+ """
+ Returns {} if not nested
+ Returns resource: properties if nested
+ Returns resource: properties: resource_def: properties if RG
+ """
+ if not bool(self.get_nested_filename()):
+ return {}
+ elif self.resource_type == "OS::Heat::ResourceGroup":
+ return nested_dict.get(
+ self.properties, "resource_def", "properties", default={}
+ )
+ else:
+ return self.properties
+
+ @property
+ def depends_on(self):
+ """
+ Returns the list of resources this resource depends on. Always
+ returns a list.
+
+ :return: list of all resource IDs this resource depends on. If none,
+ then returns an empty list
+ """
+ parents = self.resource.get("depends_on", [])
+ return parents if isinstance(parents, list) else [parents]
+
+ def is_nested(self):
+ """Returns True if the resource represents a Nested YAML resource
+ using either type: {filename} or ResourceGroup -> resource_def"""
+ return bool(self.get_nested_filename())
+
+ def get_nested_yaml(self, base_dir):
+ """If the resource represents a Nested YAML resource, then it
+ returns the loaded YAML. If the resource is not nested or the
+ file cannot be found, then an empty dict is returned"""
+ filename = self.get_nested_filename()
+ if filename:
+ file_path = os.path.join(base_dir, filename)
+ return load_yaml(file_path) if os.path.exists(file_path) else {}
+ else:
+ return {}
+
+
+def _get_heat_objects():
+ """
+ Introspect this module and return a dict of all HeatObject sub-classes with
+ a (True) resource_type. Key is the resource_type, value is the
+ corresponding class.
+ """
+ mod_classes = inspect.getmembers(sys.modules[__name__], inspect.isclass)
+ heat_objects = {
+ c.resource_type: c
+ for _, c in mod_classes
+ if issubclass(c, HeatObject) and c.resource_type
+ }
+ return heat_objects
+
+
+_HEAT_OBJECTS = _get_heat_objects()
Code Review / vvp / validation-scripts.git / blobdiff