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"
# regex parses the proper resource id format.
@staticmethod
- def get_param_value(value):
+ def get_param_value(value, withIndex=False):
"""Return get_param value of `value`
"""
if isinstance(value, dict) and len(value) == 1:
v = value.get("get_param")
if isinstance(v, list) and v:
- v = v[0]
+ if withIndex and len(v) > 1:
+ idx = v[1]
+ if isinstance(idx, dict):
+ idx = idx.get("get_param", idx)
+ v = "{}{}".format(v[0], idx)
+ else:
+ v = v[0]
else:
v = None
return v
# are replaced in the template in arbitrary order.
name = template
for key, value in params.items():
- param = cls.get_param_value(value)
+ param = cls.get_param_value(value, withIndex=True)
if param is None:
return None
name = name.replace(key, str(param))
network_flavor_external = "external"
network_flavor_internal = "internal"
- network_flavor_subint = "subint"
+ network_flavor_subint = "subinterface"
@classmethod
def get_network_flavor(cls, resource):
re_rids = collections.OrderedDict(
[
(
- "int_ip",
+ "internal",
_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)?"
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",
+ "subinterface",
_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)?"
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",
+ "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"(_v6)?"
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"$"
- ),
- ),
]
)
re_rids = collections.OrderedDict(
[
(
- "vmi_internal",
+ "internal",
_get_regex(
r"(?P<vm_type>.+)"
r"_(?P<vm_type_index>\d+)"
),
),
(
- "vmi_subint",
+ "subinterface",
_get_regex(
r"(?P<vm_type>.+)"
r"_(?P<vm_type_index>\d+)"
),
),
(
- "vmi_external",
+ "external",
_get_regex(
r"(?P<vm_type>.+)"
r"_(?P<vm_type_index>\d+)"
re_rids = 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"$")),
+ # ("rvn", _get_regex(r"int" r"_(?P<network_role>.+)" r"_RVN" r"$")),
]
)
re_rids = collections.OrderedDict(
[
(
- "internal_port",
+ "internal",
_get_regex(
r"(?P<vm_type>.+)"
r"_(?P<vm_type_index>\d+)"
),
),
(
- "port",
+ "external",
_get_regex(
r"(?P<vm_type>.+)"
r"_(?P<vm_type_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"$"
- ),
- ),
]
)
"""Returns True/False as `resource` is/not
An OS::Nova:Port with the property binding:vnic_type
"""
- return nested_dict.get(
- resource, "type"
- ) == cls.resource_type and "binding:vnic_type" in nested_dict.get(
- resource, "properties", default={}
- )
+ 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":
+ return True
+
+ return False
class NovaServerProcessor(HeatProcessor):
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
}
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):
"""
Code Review / vvp / validation-scripts.git / blobdiff