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============================================
39 from abc import ABC, abstractmethod
40 from collections import OrderedDict
42 from preload.generator import yield_by_count
43 from preload.environment import PreloadEnvironment
44 from tests.helpers import (
52 from tests.parametrizers import parametrize_heat_templates
53 from tests.structures import NeutronPortProcessor, Heat
54 from tests.test_environment_file_parameters import get_preload_excluded_parameters
55 from tests.utils import nested_dict
56 from tests.utils.vm_types import get_vm_type_for_nova_server
57 from config import Config, get_generator_plugins
59 from tests.test_environment_file_parameters import ENV_PARAMETER_SPEC
64 # This is only used to fake out parametrizers
66 def __init__(self, config):
70 def parametrize(self, name, file_list):
71 self.inputs[name] = file_list
74 def get_heat_templates(config):
76 Returns the Heat template paths discovered by the pytest parameterizers
77 :param config: pytest config
78 :return: list of heat template paths
80 meta = DummyMetafunc(config)
81 parametrize_heat_templates(meta)
82 heat_templates = meta.inputs.get("heat_templates", [])
83 if isinstance(heat_templates, list) and len(heat_templates) > 0:
84 heat_templates = heat_templates[0]
90 class FilterBaseOutputs(ABC):
92 Invoked to remove parameters in an object that appear in the base module.
93 Base output parameters can be passed to incremental modules
94 so they do not need to be defined in a preload. This method can be
95 invoked on a module to pre-filter the parameters before a preload is
98 The method should remove the parameters that exist in the base module from
99 both itself and any sub-objects.
103 def filter_output_params(self, base_outputs):
104 raise NotImplementedError()
108 def __init__(self, ip_addr_param, port):
109 self.param = ip_addr_param or ""
113 def ip_version(self):
114 return 6 if "_v6_" in self.param else 4
117 return hash(self.param)
119 def __eq__(self, other):
120 return hash(self) == hash(other)
123 return "{}(v{})".format(self.param, self.ip_version)
129 class Network(FilterBaseOutputs):
130 def __init__(self, role, name_param):
131 self.network_role = role
132 self.name_param = name_param
133 self.subnet_params = set()
135 def filter_output_params(self, base_outputs):
136 self.subnet_params = remove(self.subnet_params, base_outputs)
139 return hash(self.network_role)
141 def __eq__(self, other):
142 return hash(self) == hash(other)
145 class Port(FilterBaseOutputs):
146 def __init__(self, vm, network):
148 self.network = network
150 self.floating_ips = []
151 self.uses_dhcp = True
153 def add_ips(self, props):
154 props = props.get("properties") or props
155 for fixed_ip in props.get("fixed_ips") or []:
156 if not isinstance(fixed_ip, dict):
158 ip_address = get_param(fixed_ip.get("ip_address"))
159 subnet = get_param(fixed_ip.get("subnet") or fixed_ip.get("subnet_id"))
161 self.uses_dhcp = False
162 self.fixed_ips.append(IpParam(ip_address, self))
164 self.network.subnet_params.add(subnet)
165 for ip in prop_iterator(props, "allowed_address_pairs", "ip_address"):
166 self.uses_dhcp = False
167 param = get_param(ip) if ip else ""
169 self.floating_ips.append(IpParam(param, self))
171 def filter_output_params(self, base_outputs):
172 self.fixed_ips = remove(self.fixed_ips, base_outputs, key=lambda ip: ip.param)
173 self.floating_ips = remove(
174 self.floating_ips, base_outputs, key=lambda ip: ip.param
178 class VirtualMachineType(FilterBaseOutputs):
179 def __init__(self, vm_type, vnf_module):
180 self.vm_type = vm_type
184 self.vnf_module = vnf_module
186 def filter_output_params(self, base_outputs):
187 self.names = remove(self.names, base_outputs)
188 for port in self.ports:
189 port.filter_output_params(base_outputs)
193 return {port.network for port in self.ports}
196 def floating_ips(self):
197 for port in self.ports:
198 for ip in port.floating_ips:
203 for port in self.ports:
204 for ip in port.fixed_ips:
207 def update_ports(self, network, props):
208 port = self.get_or_create_port(network)
211 def get_or_create_port(self, network):
212 for port in self.ports:
213 if port.network == network:
215 port = Port(self, network)
216 self.ports.append(port)
221 def __init__(self, templates):
222 self.modules = [VnfModule(t, self) for t in templates]
223 self.uses_contrail = self._uses_contrail()
224 self.base_module = next(
225 (mod for mod in self.modules if mod.is_base_module), None
227 self.incremental_modules = [m for m in self.modules if not m.is_base_module]
229 def _uses_contrail(self):
230 for mod in self.modules:
231 resources = mod.heat.get_all_resources()
232 types = (r.get("type", "") for r in resources.values())
233 if any(t.startswith("OS::ContrailV2") for t in types):
238 def base_output_params(self):
239 return self.base_module.heat.outputs if self.base_module else {}
241 def filter_base_outputs(self):
242 non_base_modules = (m for m in self.modules if not m.is_base_module)
243 for mod in non_base_modules:
244 mod.filter_output_params(self.base_output_params)
247 def env_path(heat_path):
249 Create the path to the env file for the give heat path.
250 :param heat_path: path to heat file
251 :return: path to env file (assumes it is present and named correctly)
253 base_path = os.path.splitext(heat_path)[0]
254 env_path = "{}.env".format(base_path)
255 return env_path if os.path.exists(env_path) else None
258 class VnfModule(FilterBaseOutputs):
259 def __init__(self, template_file, vnf):
261 self.vnf_name = os.path.splitext(os.path.basename(template_file))[0]
262 self.template_file = template_file
263 self.heat = Heat(filepath=template_file, envpath=env_path(template_file))
264 env_pair = get_environment_pair(self.template_file)
265 env_yaml = env_pair.get("eyml") if env_pair else {}
266 self.parameters = {key: "" for key in self.heat.parameters}
267 self.parameters.update(env_yaml.get("parameters") or {})
269 self.virtual_machine_types = self._create_vm_types()
271 self.outputs_filtered = False
273 def filter_output_params(self, base_outputs):
274 for vm in self.virtual_machine_types:
275 vm.filter_output_params(base_outputs)
276 for network in self.networks:
277 network.filter_output_params(base_outputs)
279 k: v for k, v in self.parameters.items() if k not in base_outputs
283 for network in self.networks
284 if network.name_param not in base_outputs or network.subnet_params
286 self.outputs_filtered = True
288 def _create_vm_types(self):
289 servers = self.heat.get_resource_by_type("OS::Nova::Server", all_resources=True)
291 for _, props in yield_by_count(servers):
292 vm_type = get_vm_type_for_nova_server(props)
293 vm = vm_types.setdefault(vm_type, VirtualMachineType(vm_type, self))
295 name = nested_dict.get(props, "properties", "name", default={})
296 vm_name = get_param(name) if name else ""
297 vm.names.append(vm_name)
298 return list(vm_types.values())
300 def _add_networks(self):
301 ports = self.heat.get_resource_by_type("OS::Neutron::Port", all_resources=True)
302 for rid, props in yield_by_count(ports):
303 resource_type, port_match = NeutronPortProcessor.get_rid_match_tuple(rid)
304 if resource_type != "external":
306 network_role = port_match.group("network_role")
307 vm = self._get_vm_type(port_match.group("vm_type"))
308 network = self._get_network(network_role, props)
309 vm.update_ports(network, props)
312 def is_base_module(self):
313 return is_base_module(self.template_file)
316 def availability_zones(self):
317 """Returns a list of all availability zone parameters found in the template"""
319 p for p in self.heat.parameters if p.startswith("availability_zone")
325 Label for the VF module that will appear in the CSAR
331 """Return available Environment Spec definitions"""
333 return Config().env_specs
334 except FileNotFoundError:
335 return [ENV_PARAMETER_SPEC]
338 def platform_provided_params(self):
340 for spec in self.env_specs:
341 for props in spec["PLATFORM PROVIDED"]:
342 result.add(props["property"][-1])
346 def env_template(self):
348 Returns a a template .env file that can be completed to enable
351 params = OrderedDict()
352 params["vnf-type"] = CHANGE
353 params["vf-module-model-name"] = CHANGE
354 params["vf_module_name"] = CHANGE
355 for az in self.availability_zones:
357 for network in self.networks:
358 params[network.name_param] = CHANGE
359 for param in set(network.subnet_params):
360 params[param] = CHANGE
361 for vm in self.virtual_machine_types:
362 for name in set(vm.names):
363 params[name] = CHANGE
364 for ip in vm.floating_ips:
365 params[ip.param] = CHANGE
366 for ip in vm.fixed_ips:
367 params[ip.param] = CHANGE
368 excluded = get_preload_excluded_parameters(
369 self.template_file, persistent_only=True
371 excluded.update(self.platform_provided_params)
372 for name, value in self.parameters.items():
375 params[name] = value if value else CHANGE
376 return {"parameters": params}
379 def preload_parameters(self):
381 Subset of parameters from the env file that can be overridden in
382 tag values. Per VNF Heat Guidelines, specific parameters such as
383 flavor, image, etc. must not be overridden so they are excluded.
385 :return: dict of parameters suitable for the preload
387 excluded = get_preload_excluded_parameters(self.template_file)
388 params = {k: v for k, v in self.parameters.items() if k not in excluded}
391 def _get_vm_type(self, vm_type):
392 for vm in self.virtual_machine_types:
393 if vm_type.lower() == vm.vm_type.lower():
395 raise RuntimeError("Encountered unknown VM type: {}".format(vm_type))
397 def _get_network(self, network_role, props):
398 network_prop = nested_dict.get(props, "properties", "network") or {}
399 name_param = get_param(network_prop) if network_prop else ""
400 for network in self.networks:
401 if network.network_role.lower() == network_role.lower():
403 new_network = Network(network_role, name_param)
404 self.networks.append(new_network)
408 return "VNF Module ({})".format(os.path.basename(self.template_file))
414 return hash(self.vnf_name)
416 def __eq__(self, other):
417 return hash(self) == hash(other)
420 def create_preloads(config, exitstatus):
422 Create preloads in every format that can be discovered by get_generator_plugins
424 if config.getoption("self_test"):
426 print("+===================================================================+")
427 print("| Preload Template Generation |")
428 print("+===================================================================+")
430 preload_dir = os.path.join(get_output_dir(config), "preloads")
431 if os.path.exists(preload_dir):
432 shutil.rmtree(preload_dir)
433 env_directory = config.getoption("env_dir")
434 preload_env = PreloadEnvironment(env_directory) if env_directory else None
435 plugins = get_generator_plugins()
436 available_formats = [p.format_name() for p in plugins]
437 selected_formats = config.getoption("preload_formats") or available_formats
438 heat_templates = get_heat_templates(config)
440 for plugin_class in plugins:
441 if plugin_class.format_name() not in selected_formats:
443 vnf = Vnf(heat_templates)
444 generator = plugin_class(vnf, preload_dir, preload_env)
446 if vnf and vnf.uses_contrail:
448 "\nWARNING: Preload template generation does not support Contrail\n"
449 "at this time, but Contrail resources were detected. The preload \n"
450 "template may be incomplete."
454 "\nWARNING: Heat violations detected. Preload templates may be\n"