ice_validator/tests/test_neutron_port_network_attachment.py

   1 import os
   2 import re
   3
   4 import pytest
   5
   6 from tests.helpers import validates, get_base_template_from_yaml_files
   7 from tests.parametrizers import get_nested_files
   8 from tests.structures import Heat
   9
  10 INTERNAL_UUID_PATTERN = re.compile(r"^int_(?P<network_role>.+?)_net_id$")
  11 INTERNAL_NAME_PATTERN = re.compile(r"^int_(?P<network_role>.+?)_net_name$")
  12 INTERNAL_PORT = re.compile(r"^(?P<vm_type>.+)_(?P<vm_type_index>\d+)_int_"
  13                            r"(?P<network_role>.+)_port_(?P<port_index>\d+)$")
  14
  15 EXTERNAL_PORT = re.compile(r"^(?P<vm_type>.+)_(?P<vm_type_index>\d+)_(?!int_)"
  16                            r"(?P<network_role>.+)_port_(?P<port_index>\d+)$")
  17
  18 EXTERNAL_UUID_PATTERN = re.compile(r"^(?!int_)(?P<network_role>.+?)_net_id$")
  19 EXTERNAL_NAME_PATTERN = re.compile(r"^(?!int_)(?P<network_role>.+?)_net_name$")
  20
  21 INTERNAL_NETWORK_PATTERN = re.compile(r"^int_(?P<network_role>.+?)"
  22                                       r"_(network|RVN)$")
  23
  24
  25 def is_incremental_module(yaml_file, base_path, nested_paths):
  26     return yaml_file != base_path and yaml_file not in nested_paths
  27
  28
  29 def get_param(prop_val):
  30     if not isinstance(prop_val, dict):
  31         return None
  32     param = prop_val.get("get_param")
  33     return param if isinstance(param, str) else None
  34
  35
  36 @validates("R-86182", "R-22688")
  37 def test_internal_network_parameters(yaml_files):
  38     base_path = get_base_template_from_yaml_files(yaml_files)
  39     if not base_path:
  40         pytest.skip("No base module found")
  41     base_heat = Heat(filepath=base_path)
  42     nested_paths = get_nested_files(yaml_files)
  43     incremental_modules = [f for f in yaml_files
  44                            if is_incremental_module(f, base_path, nested_paths)]
  45     errors = []
  46     for module in incremental_modules:
  47         heat = Heat(filepath=module)
  48         for rid, port in heat.neutron_port_resources.items():
  49             rid_match = INTERNAL_PORT.match(rid)
  50             if not rid_match:
  51                 continue
  52
  53             network = (port.get("properties") or {}).get("network") or {}
  54             if isinstance(network, dict) and (
  55                     "get_resource" in network or "get_attr" in network):
  56                 continue
  57
  58             param = get_param(network)
  59             if not param:
  60                 errors.append((
  61                     "The internal port ({}) must either connect to a network "
  62                     "in the base module using get_param or to a network "
  63                     "created in this module ({})"
  64                 ).format(rid, os.path.split(module)[1]))
  65                 continue
  66
  67             param_match = (
  68                 INTERNAL_UUID_PATTERN.match(param)
  69                 or INTERNAL_NAME_PATTERN.match(param)
  70             )
  71             if not param_match:
  72                 errors.append((
  73                     "The internal port ({}) network parameter ({}) does not "
  74                     "match one of the required naming conventions of "
  75                     "int_{{network-role}}_net_id or "
  76                     "int_{{network-role}}_net_name "
  77                     "for connecting to an internal network. "
  78                     "If this is not an internal port, then change the resource "
  79                     "ID to adhere to the external port naming convention."
  80                 ).format(rid, param))
  81                 continue
  82
  83             if param not in base_heat.yml.get("outputs", {}):
  84                 base_module = os.path.split(base_path)[1]
  85                 errors.append((
  86                     "The internal network parameter ({}) attached to port ({}) "
  87                     "must be defined in the output section of the base module ({})."
  88                 ).format(param, rid, base_module))
  89                 continue
  90
  91             param_network_role = param_match.groupdict().get("network_role")
  92             rid_network_role = rid_match.groupdict().get("network_role")
  93             if param_network_role != rid_network_role:
  94                 errors.append((
  95                     "The network role ({}) extracted from the resource ID ({}) "
  96                     "does not match network role ({}) extracted from the "
  97                     "network parameter ({})"
  98                 ).format(rid_network_role, rid, param_network_role, param))
  99
 100             resources = base_heat.get_all_resources(os.path.split(base_path)[0])
 101             networks = {rid: resource for rid, resource in resources.items()
 102                         if resource.get("type")
 103                         in {"OS::Neutron::Net",
 104                             "OS::ContrailV2::VirtualNetwork"}}
 105             matches = (INTERNAL_NETWORK_PATTERN.match(n) for n in networks)
 106             roles = {m.groupdict()["network_role"] for m in matches if m}
 107             if param_network_role not in roles:
 108                 errors.append((
 109                     "No internal network with a network role of {} was "
 110                     "found in the base modules networks: {}"
 111                 ).format(param_network_role, ", ".join(networks)))
 112
 113     assert not errors, ". ".join(errors)
 114
 115
 116 @validates("R-62983")
 117 def test_external_network_parameter(heat_template):
 118     heat = Heat(filepath=heat_template)
 119     errors = []
 120     for rid, port in heat.neutron_port_resources.items():
 121         rid_match = EXTERNAL_PORT.match(rid)
 122         if not rid_match:
 123             continue   # only test external ports
 124         network = (port.get("properties") or {}).get("network") or {}
 125         if not isinstance(network, dict) or "get_param" not in network:
 126             errors.append((
 127                 "The external port ({}) must assign the network property "
 128                 "using get_param.  If this port is for an internal network, "
 129                 "then change the resource ID format to the external format."
 130             ).format(rid))
 131             continue
 132         param = get_param(network)
 133         if not param:
 134             errors.append((
 135                 "The get_param function on the network property of port ({}) "
 136                 "must only take a single, string parameter."
 137             ).format(rid))
 138             continue
 139
 140         param_match = (
 141             EXTERNAL_NAME_PATTERN.match(param)
 142             or EXTERNAL_UUID_PATTERN.match(param)
 143         )
 144         if not param_match:
 145             errors.append((
 146                 "The network parameter ({}) on port ({}) does not match one of "
 147                 "{{network-role}}_net_id or {{network-role}}_net_name."
 148             ).format(param, rid))
 149             continue
 150         rid_network_role = rid_match.groupdict()["network_role"]
 151         param_network_role = param_match.groupdict()["network_role"]
 152         if rid_network_role != param_network_role:
 153             errors.append((
 154                 "The network role ({}) extracted from the resource ID ({}) "
 155                 "does not match network role ({}) extracted from the "
 156                 "network parameter ({})"
 157             ).format(rid_network_role, rid, param_network_role, param))
 158
 159     assert not errors, ". ".join(errors)