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