ice_validator/tests/test_neutron_port_addresses.py

   1 # -*- coding: utf8 -*-
   2 # ============LICENSE_START====================================================
   3 # org.onap.vvp/validation-scripts
   4 # ===================================================================
   5 # Copyright © 2019 AT&T Intellectual Property. All rights reserved.
   6 # ===================================================================
   7 #
   8 # Unless otherwise specified, all software contained herein is licensed
   9 # under the Apache License, Version 2.0 (the "License");
  10 # you may not use this software except in compliance with the License.
  11 # You may obtain a copy of the License at
  12 #
  13 #             http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 # Unless required by applicable law or agreed to in writing, software
  16 # distributed under the License is distributed on an "AS IS" BASIS,
  17 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18 # See the License for the specific language governing permissions and
  19 # limitations under the License.
  20 #
  21 #
  22 #
  23 # Unless otherwise specified, all documentation contained herein is licensed
  24 # under the Creative Commons License, Attribution 4.0 Intl. (the "License");
  25 # you may not use this documentation except in compliance with the License.
  26 # You may obtain a copy of the License at
  27 #
  28 #             https://creativecommons.org/licenses/by/4.0/
  29 #
  30 # Unless required by applicable law or agreed to in writing, documentation
  31 # distributed under the License is distributed on an "AS IS" BASIS,
  32 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  33 # See the License for the specific language governing permissions and
  34 # limitations under the License.
  35 #
  36 # ============LICENSE_END============================================
  37 #
  38 #
  39
  40 """
  41 OS::Neutron::Port connecting to external network
  42 must have at most one ip_address and at most one v6_ip_address.
  43 """
  44
  45 import collections
  46 import os.path
  47
  48 from .structures import Heat
  49 from .helpers import validates
  50
  51 VERSION = "1.1.0"
  52
  53
  54 def get_neutron_ports(heat):
  55     """Return dict of resource_id: resource, whose type is
  56     OS::Neutron::Port.
  57     """
  58     return {
  59         rid: resource
  60         for rid, resource in heat.resources.items()
  61         if heat.nested_get(resource, "type") == "OS::Neutron::Port"
  62     }
  63
  64
  65 def get_port_addresses(filepath):
  66     """Return dict:
  67     key is field name, value is dict:
  68         key is parameter name, value is dict:
  69             key is filepath, value is set of rid
  70     """
  71     port_addresses = collections.defaultdict(
  72         lambda: collections.defaultdict(lambda: collections.defaultdict(set))
  73     )
  74     heat = Heat(filepath=filepath)
  75     basename = os.path.basename(filepath)
  76     for rid, port in get_neutron_ports(heat).items():
  77         allowed_address_pairs = heat.nested_get(
  78             port, "properties", "allowed_address_pairs"
  79         )
  80         if not isinstance(allowed_address_pairs, list):
  81             continue
  82         field = "ip_address"
  83         for aa_pair in allowed_address_pairs:
  84             param = heat.nested_get(aa_pair, field, "get_param")
  85             if param is None:
  86                 continue
  87             else:
  88                 param = param[0] if isinstance(param, list) else param
  89             port_addresses[field][param][basename].add(rid)
  90     return port_addresses
  91
  92
  93 def nested_update(out_dict, in_dict):
  94     """Recursively update out_dict from in_dict.
  95     """
  96     for key, value in in_dict.items():
  97         if key not in out_dict:
  98             out_dict[key] = value
  99         elif isinstance(value, dict) and isinstance(out_dict[key], dict):
 100             out_dict[key] = nested_update(out_dict[key], value)
 101         elif isinstance(value, set) and isinstance(out_dict[key], set):
 102             out_dict[key].update(value)
 103         else:
 104             out_dict[key] = value
 105     return out_dict
 106
 107
 108 @validates("R-10754")
 109 def test_neutron_port_floating(yaml_files):
 110     """
 111     If a VNF has two or more ports that
 112     attach to an external network that require a Virtual IP Address (VIP),
 113     and the VNF requires ONAP automation to assign the IP address,
 114     all the Virtual Machines using the VIP address **MUST**
 115     be instantiated in the same Base Module Heat Orchestration Template
 116     or in the same Incremental Module Heat Orchestration Template.
 117     """
 118     fields = {}
 119     for filepath in yaml_files:
 120         fields = nested_update(fields, get_port_addresses(filepath))
 121     bad = []
 122     for field, params in fields.items():
 123         for param, files in params.items():
 124             if len(files) > 1:
 125                 error = ["{} {} assigned in multiple templates: ".format(field, param)]
 126                 for file_name, r_ids in files.items():
 127                     error.append(
 128                         "In {} it's assigned to {}. ".format(
 129                             file_name, ", ".join(r_ids)
 130                         )
 131                     )
 132                 bad.append("".join(error))
 133     assert not bad, "; ".join(bad)