2 # ============LICENSE_START====================================================
3 # org.onap.vvp/validation-scripts
4 # ===================================================================
5 # Copyright © 2019 AT&T Intellectual Property. All rights reserved.
6 # ===================================================================
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
13 # http://www.apache.org/licenses/LICENSE-2.0
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.
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
28 # https://creativecommons.org/licenses/by/4.0/
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.
36 # ============LICENSE_END============================================
38 from abc import ABC, abstractmethod
39 from collections import OrderedDict
40 from itertools import chain
41 from typing import Tuple, List
43 from tests.helpers import (
50 from tests.parametrizers import parametrize_heat_templates
51 from tests.structures import NeutronPortProcessor, Heat
52 from tests.test_environment_file_parameters import get_preload_excluded_parameters
53 from tests.utils import nested_dict
54 from tests.utils.vm_types import get_vm_type_for_nova_server
56 from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
61 # This is only used to fake out parametrizers
63 def __init__(self, config):
67 def parametrize(self, name, file_list):
68 self.inputs[name] = file_list
71 def get_heat_templates(config):
73 Returns the Heat template paths discovered by the pytest parameterizers
74 :param config: pytest config
75 :return: list of heat template paths
77 meta = DummyMetafunc(config)
78 parametrize_heat_templates(meta)
79 heat_templates = meta.inputs.get("heat_templates", [])
80 if isinstance(heat_templates, list) and len(heat_templates) > 0:
81 heat_templates = heat_templates[0]
87 class FilterBaseOutputs(ABC):
89 Invoked to remove parameters in an object that appear in the base module.
90 Base output parameters can be passed to incremental modules
91 so they do not need to be defined in a preload. This method can be
92 invoked on a module to pre-filter the parameters before a preload is
95 The method should remove the parameters that exist in the base module from
96 both itself and any sub-objects.
100 def filter_output_params(self, base_outputs):
101 raise NotImplementedError()
105 def __init__(self, ip_addr_param, port):
106 self.param = ip_addr_param or ""
110 def ip_version(self):
111 return 6 if "_v6_" in self.param else 4
114 return hash(self.param)
116 def __eq__(self, other):
117 return hash(self) == hash(other)
120 return "{}(v{})".format(self.param, self.ip_version)
126 class Network(FilterBaseOutputs):
127 def __init__(self, role, name_param):
128 self.network_role = role
129 self.name_param = name_param
130 self.subnet_params = set()
132 def filter_output_params(self, base_outputs):
133 self.subnet_params = remove(
134 self.subnet_params, base_outputs, key=lambda s: s.param_name
138 return hash(self.network_role)
140 def __eq__(self, other):
141 return hash(self) == hash(other)
145 def __init__(self, param_name: str):
146 self.param_name = param_name
149 def ip_version(self):
150 return 6 if "_v6_" in self.param_name else 4
153 return hash(self.param_name)
155 def __eq__(self, other):
156 return hash(self) == hash(other)
159 class Port(FilterBaseOutputs):
160 def __init__(self, vm, network):
162 self.network = network
164 self.floating_ips = set()
165 self.uses_dhcp = True
167 def add_ips(self, props):
168 props = props.get("properties") or props
169 for fixed_ip in props.get("fixed_ips") or []:
170 if not isinstance(fixed_ip, dict):
172 ip_address = get_param(fixed_ip.get("ip_address"))
173 subnet = get_param(fixed_ip.get("subnet") or fixed_ip.get("subnet_id"))
175 self.uses_dhcp = False
176 self.fixed_ips.append(IpParam(ip_address, self))
178 self.network.subnet_params.add(Subnet(subnet))
179 for ip in prop_iterator(props, "allowed_address_pairs", "ip_address"):
180 param = get_param(ip) if ip else ""
182 self.floating_ips.add(IpParam(param, self))
185 def ipv6_fixed_ips(self):
188 (ip for ip in self.fixed_ips if ip.ip_version == 6),
189 key=lambda ip: ip.param,
194 def ipv4_fixed_ips(self):
197 (ip for ip in self.fixed_ips if ip.ip_version == 4),
198 key=lambda ip: ip.param,
203 def fixed_ips_with_index(self) -> List[Tuple[int, IpParam]]:
204 ipv4s = enumerate(self.ipv4_fixed_ips)
205 ipv6s = enumerate(self.ipv6_fixed_ips)
206 return list(chain(ipv4s, ipv6s))
208 def filter_output_params(self, base_outputs):
209 self.fixed_ips = remove(self.fixed_ips, base_outputs, key=lambda ip: ip.param)
210 self.floating_ips = remove(
211 self.floating_ips, base_outputs, key=lambda ip: ip.param
215 class VirtualMachineType(FilterBaseOutputs):
216 def __init__(self, vm_type, vnf_module):
217 self.vm_type = vm_type
221 self.vnf_module = vnf_module
223 def filter_output_params(self, base_outputs):
224 self.names = remove(self.names, base_outputs)
225 for port in self.ports:
226 port.filter_output_params(base_outputs)
230 return {port.network for port in self.ports}
233 def floating_ips(self):
234 for port in self.ports:
235 for ip in port.floating_ips:
240 for port in self.ports:
241 for ip in port.fixed_ips:
244 def update_ports(self, network, props):
245 port = self.get_or_create_port(network)
248 def get_or_create_port(self, network):
249 for port in self.ports:
250 if port.network == network:
252 port = Port(self, network)
253 self.ports.append(port)
258 def __init__(self, templates, config=None):
259 self.modules = [VnfModule(t, self, config) for t in templates]
260 self.uses_contrail = self._uses_contrail()
262 self.base_module = next(
263 (mod for mod in self.modules if mod.is_base_module), None
265 self.incremental_modules = [m for m in self.modules if not m.is_base_module]
267 def _uses_contrail(self):
268 for mod in self.modules:
269 resources = mod.heat.get_all_resources()
270 types = (r.get("type", "") for r in resources.values())
271 if any(t.startswith("OS::ContrailV2") for t in types):
276 def base_output_params(self):
277 return self.base_module.heat.outputs if self.base_module else {}
279 def filter_base_outputs(self):
280 non_base_modules = (m for m in self.modules if not m.is_base_module)
281 for mod in non_base_modules:
282 mod.filter_output_params(self.base_output_params)
285 def env_path(heat_path):
287 Create the path to the env file for the give heat path.
288 :param heat_path: path to heat file
289 :return: path to env file (assumes it is present and named correctly)
291 base_path = os.path.splitext(heat_path)[0]
292 env_path = "{}.env".format(base_path)
293 return env_path if os.path.exists(env_path) else None
296 class VnfModule(FilterBaseOutputs):
297 def __init__(self, template_file, vnf, config):
300 self.vnf_name = os.path.splitext(os.path.basename(template_file))[0]
301 self.template_file = template_file
302 self.heat = Heat(filepath=template_file, envpath=env_path(template_file))
303 env_pair = get_environment_pair(self.template_file)
304 env_yaml = env_pair.get("eyml") if env_pair else {}
305 self.parameters = {key: "" for key in self.heat.parameters}
306 self.parameters.update(env_yaml.get("parameters") or {})
307 # Filter out any parameters passed from the volume module's outputs
310 for key, value in self.parameters.items()
311 if key not in self.volume_module_outputs
314 self.virtual_machine_types = self._create_vm_types()
316 self.outputs_filtered = False
319 def volume_module_outputs(self):
320 heat_dir = os.path.dirname(self.template_file)
321 heat_filename = os.path.basename(self.template_file)
322 basename, ext = os.path.splitext(heat_filename)
323 volume_template_name = "{}_volume{}".format(basename, ext)
324 volume_path = os.path.join(heat_dir, volume_template_name)
325 if os.path.exists(volume_path):
326 volume_mod = Heat(filepath=volume_path)
327 return volume_mod.outputs
331 def filter_output_params(self, base_outputs):
332 for vm in self.virtual_machine_types:
333 vm.filter_output_params(base_outputs)
334 for network in self.networks:
335 network.filter_output_params(base_outputs)
337 k: v for k, v in self.parameters.items() if k not in base_outputs
341 for network in self.networks
342 if network.name_param not in base_outputs or network.subnet_params
344 self.outputs_filtered = True
346 def _create_vm_types(self):
347 servers = self.heat.get_resource_by_type("OS::Nova::Server", all_resources=True)
349 for _, props in yield_by_count(servers):
350 vm_type = get_vm_type_for_nova_server(props)
351 vm = vm_types.setdefault(vm_type, VirtualMachineType(vm_type, self))
353 name = nested_dict.get(props, "properties", "name", default={})
354 vm_name = get_param(name) if name else ""
355 vm.names.append(vm_name)
356 return list(vm_types.values())
358 def _add_networks(self):
359 ports = self.heat.get_resource_by_type("OS::Neutron::Port", all_resources=True)
360 for rid, props in yield_by_count(ports):
361 resource_type, port_match = NeutronPortProcessor.get_rid_match_tuple(rid)
362 if resource_type != "external":
364 network_role = port_match.group("network_role")
365 vm = self._get_vm_type(port_match.group("vm_type"))
366 network = self._get_network(network_role, props)
367 vm.update_ports(network, props)
370 def is_base_module(self):
371 return is_base_module(self.template_file)
374 def availability_zones(self):
375 """Returns a list of all availability zone parameters found in the template"""
377 p for p in self.heat.parameters if p.startswith("availability_zone")
383 Label for the VF module that will appear in the CSAR
389 """Return available Environment Spec definitions"""
390 return [ENV_PARAMETER_SPEC] if not self.config else self.config.env_specs
393 def platform_provided_params(self):
395 for spec in self.env_specs:
396 for props in spec["PLATFORM PROVIDED"]:
397 result.add(props["property"][-1])
401 def env_template(self):
403 Returns a a template .env file that can be completed to enable
406 params = OrderedDict()
407 params["vnf-type"] = CHANGE
408 params["vf-module-model-name"] = CHANGE
409 params["vf_module_name"] = CHANGE
410 for az in self.availability_zones:
412 for network in self.networks:
413 params[network.name_param] = CHANGE
414 for param in set(s.param_name for s in network.subnet_params):
415 params[param] = CHANGE
416 for vm in self.virtual_machine_types:
417 for name in set(vm.names):
418 params[name] = CHANGE
419 for ip in vm.floating_ips:
420 params[ip.param] = CHANGE
421 for ip in vm.fixed_ips:
422 params[ip.param] = CHANGE
423 excluded = get_preload_excluded_parameters(
424 self.template_file, persistent_only=True
426 excluded.update(self.platform_provided_params)
427 for name, value in self.parameters.items():
430 params[name] = value if value else CHANGE
431 return {"parameters": params}
434 def preload_parameters(self):
436 Subset of parameters from the env file that can be overridden in
437 tag values. Per VNF Heat Guidelines, specific parameters such as
438 flavor, image, etc. must not be overridden so they are excluded.
440 :return: dict of parameters suitable for the preload
442 excluded = get_preload_excluded_parameters(self.template_file)
443 excluded.update(self.platform_provided_params)
444 params = {k: v for k, v in self.parameters.items() if k not in excluded}
447 def _get_vm_type(self, vm_type):
448 for vm in self.virtual_machine_types:
449 if vm_type.lower() == vm.vm_type.lower():
451 raise RuntimeError("Encountered unknown VM type: {}".format(vm_type))
453 def _get_network(self, network_role, props):
454 network_prop = nested_dict.get(props, "properties", "network") or {}
455 name_param = get_param(network_prop) if network_prop else ""
456 for network in self.networks:
457 if network.network_role.lower() == network_role.lower():
459 new_network = Network(network_role, name_param)
460 self.networks.append(new_network)
464 return "VNF Module ({})".format(os.path.basename(self.template_file))
470 return hash(self.vnf_name)
472 def __eq__(self, other):
473 return hash(self) == hash(other)
476 def yield_by_count(sequence):
478 Iterates through sequence and yields each item according to its __count__
479 attribute. If an item has a __count__ of it will be returned 3 times
480 before advancing to the next item in the sequence.
482 :param sequence: sequence of dicts (must contain __count__)
483 :returns: generator of tuple key, value pairs
485 for key, value in sequence.items():
486 for i in range(value["__count__"]):