# ============LICENSE_END============================================
#
#
-"""structures
-"""
import collections
import inspect
import os
import sys
from tests import cached_yaml as yaml
-from tests.helpers import load_yaml
+from tests.helpers import load_yaml, get_param
from .utils import nested_dict
VERSION = "4.2.0"
network_flavor = cls.network_flavor_internal
else:
p = param.get("get_param")
- if isinstance(p, str):
- if "_int_" in p or p.startswith("int_"):
- network_flavor = cls.network_flavor_internal
- elif "_subint_" in p:
- network_flavor = cls.network_flavor_subint
- else:
- network_flavor = cls.network_flavor_external
+ network_flavor = cls.get_network_format(p)
return network_flavor
+ @classmethod
+ def get_network_format(cls, param):
+ if isinstance(param, str):
+ if "_int_" in param or param.startswith("int_"):
+ return cls.network_flavor_internal
+ elif "_subint_" in param:
+ return cls.network_flavor_subint
+ else:
+ return cls.network_flavor_external
+
class ContrailV2InstanceIpProcessor(ContrailV2NetworkFlavorBaseProcessor):
""" ContrailV2 InstanceIp
An OS::Nova:Port with the property binding:vnic_type
"""
resource_properties = nested_dict.get(resource, "properties", default={})
- if nested_dict.get(resource, "type") == cls.resource_type and resource_properties.get("binding:vnic_type", "") == "direct":
+ if (
+ nested_dict.get(resource, "type") == cls.resource_type
+ and resource_properties.get("binding:vnic_type", "") == "direct"
+ ):
return True
return False
d = dict(
flavor=cls.get_flavor(resource),
image=cls.get_image(resource),
- networks=cls.get_network(resource),
+ network_role=cls.get_network(resource),
)
if all(d.values()):
vm_class.update(d)
self.load_env(envpath)
self.heat_processors = self.get_heat_processors()
+ @property
+ def is_heat(self):
+ return "heat_template_version" in self.yml
+
@property
def contrail_resources(self):
"""This attribute is a dict of Contrail resources.
resource_type=ContrailV2VirtualMachineInterfaceProcessor.resource_type
)
- def get_all_resources(self, base_dir):
+ def get_all_resources(self, base_dir=None, count=1):
"""
- Like ``resources``,
- but this returns all the resources definitions
+ Like ``resources``, but this returns all the resources definitions
defined in the template, resource groups, and nested YAML files.
+
+ A special variable will be added to all resource properties (__count__).
+ This will normally be 1, but if the resource is generated by a
+ ResourceGroup **and** an env file is present, then the count will be
+ the value from the env file (assuming this follows standard VNF Heat
+ Guidelines)
"""
+ base_dir = base_dir or self.dirname
resources = {}
for r_id, r_data in self.resources.items():
+ r_data["__count__"] = count
resources[r_id] = r_data
resource = Resource(r_id, r_data)
if resource.is_nested():
+ nested_count = resource.get_count(self.env)
nested = Heat(os.path.join(base_dir, resource.get_nested_filename()))
- resources.update(nested.get_all_resources(base_dir))
+ nested_resources = nested.get_all_resources(count=nested_count)
+ resources.update(nested_resources)
return resources
@staticmethod
"""
return _HEAT_PROCESSORS
- def get_resource_by_type(self, resource_type):
+ def get_resource_by_type(self, resource_type, all_resources=False):
"""Return dict of resources whose type is `resource_type`.
key is resource_id, value is resource.
"""
+ resources = self.get_all_resources() if all_resources else self.resources
return {
rid: resource
- for rid, resource in self.resources.items()
+ for rid, resource in resources.items()
if self.nested_get(resource, "type") == resource_type
}
re.search("(^(%(x)s)_)|(_(%(x)s)_)|(_(%(x)s)$)" % dict(x=part), name)
)
+ def iter_nested_heat(self):
+ """
+ Returns an iterable of tuples (int, heat) where the first parameter is the
+ depth of the nested file and the second item is an instance of Heat
+ """
+
+ def walk_nested(heat, level=1):
+ resources = [Resource(r_id, data) for r_id, data in heat.resources.items()]
+ for resource in resources:
+ if resource.is_nested():
+ nested_path = os.path.join(
+ self.dirname, resource.get_nested_filename()
+ )
+ nested_heat = Heat(nested_path)
+ yield level, nested_heat
+ yield from walk_nested(nested_heat, level + 1)
+
+ yield from walk_nested(self)
+
+ def __str__(self):
+ return "Heat({})".format(self.filepath)
+
+ def __repr__(self):
+ return str(self)
+
class Env(Heat):
"""An Environment file
else:
return self.properties
+ def get_count(self, env):
+ if self.resource_type == "OS::Heat::ResourceGroup":
+ if not env:
+ return 1
+ env_params = env.parameters
+ count_param = get_param(self.properties["count"])
+ count_value = env_params.get(count_param) if count_param else 1
+ try:
+ return int(count_value)
+ except (ValueError, TypeError):
+ print(
+ (
+ "WARNING: Invalid value for count parameter {}. Expected "
+ "an integer, but got {}. Defaulting to 1"
+ ).format(count_param, count_value)
+ )
+ return 1
+
@property
def depends_on(self):
"""
else:
return {}
+ def __str__(self):
+ return "Resource(id={}, type={})".format(self.resource_id, self.resource_type)
+
+ def __repr__(self):
+ return str(self)
+
def get_all_resources(yaml_files):
"""Return a dict, resource id: resource
Code Review / vvp / validation-scripts.git / blobdiff