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
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 return "{}.env".format(base_path)
257 class VnfModule(FilterBaseOutputs):
258 def __init__(self, template_file, vnf):
260 self.vnf_name = os.path.splitext(os.path.basename(template_file))[0]
261 self.template_file = template_file
262 self.heat = Heat(filepath=template_file, envpath=env_path(template_file))
263 env_pair = get_environment_pair(self.template_file)
264 env_yaml = env_pair.get("eyml") if env_pair else {}
265 self.parameters = {key: "" for key in self.heat.parameters}
266 self.parameters.update(env_yaml.get("parameters") or {})
268 self.virtual_machine_types = self._create_vm_types()
270 self.outputs_filtered = False
272 def filter_output_params(self, base_outputs):
273 for vm in self.virtual_machine_types:
274 vm.filter_output_params(base_outputs)
275 for network in self.networks:
276 network.filter_output_params(base_outputs)
278 k: v for k, v in self.parameters.items() if k not in base_outputs
282 for network in self.networks
283 if network.name_param not in base_outputs or network.subnet_params
285 self.outputs_filtered = True
287 def _create_vm_types(self):
288 servers = self.heat.get_resource_by_type("OS::Nova::Server", all_resources=True)
290 for _, props in yield_by_count(servers):
291 vm_type = get_vm_type_for_nova_server(props)
292 vm = vm_types.setdefault(vm_type, VirtualMachineType(vm_type, self))
294 name = nested_dict.get(props, "properties", "name", default={})
295 vm_name = get_param(name) if name else ""
296 vm.names.append(vm_name)
297 return list(vm_types.values())
299 def _add_networks(self):
300 ports = self.heat.get_resource_by_type("OS::Neutron::Port", all_resources=True)
301 for rid, props in yield_by_count(ports):
302 resource_type, port_match = NeutronPortProcessor.get_rid_match_tuple(rid)
303 if resource_type != "external":
305 network_role = port_match.group("network_role")
306 vm = self._get_vm_type(port_match.group("vm_type"))
307 network = self._get_network(network_role, props)
308 vm.update_ports(network, props)
311 def is_base_module(self):
312 return is_base_module(self.template_file)
315 def availability_zones(self):
316 """Returns a list of all availability zone parameters found in the template"""
318 p for p in self.heat.parameters if p.startswith("availability_zone")
324 Label for the VF module that will appear in the CSAR
330 """Return available Environment Spec definitions"""
332 return Config().env_specs
333 except FileNotFoundError:
334 return [ENV_PARAMETER_SPEC]
337 def platform_provided_params(self):
339 for spec in self.env_specs:
340 for props in spec["PLATFORM PROVIDED"]:
341 result.add(props["property"][-1])
345 def env_template(self):
347 Returns a a template .env file that can be completed to enable
350 params = OrderedDict()
351 params["vnf-type"] = CHANGE
352 params["vf-module-model-name"] = CHANGE
353 params["vf_module_name"] = CHANGE
354 for az in self.availability_zones:
356 for network in self.networks:
357 params[network.name_param] = CHANGE
358 for param in set(network.subnet_params):
359 params[param] = CHANGE
360 for vm in self.virtual_machine_types:
361 for name in set(vm.names):
362 params[name] = CHANGE
363 for ip in vm.floating_ips:
364 params[ip.param] = CHANGE
365 for ip in vm.fixed_ips:
366 params[ip.param] = CHANGE
367 excluded = get_preload_excluded_parameters(
368 self.template_file, persistent_only=True
370 excluded.update(self.platform_provided_params)
371 for name, value in self.parameters.items():
374 params[name] = value if value else CHANGE
375 return {"parameters": params}
378 def preload_parameters(self):
380 Subset of parameters from the env file that can be overridden in
381 tag values. Per VNF Heat Guidelines, specific parameters such as
382 flavor, image, etc. must not be overridden so they are excluded.
384 :return: dict of parameters suitable for the preload
386 excluded = get_preload_excluded_parameters(self.template_file)
387 params = {k: v for k, v in self.parameters.items() if k not in excluded}
390 def _get_vm_type(self, vm_type):
391 for vm in self.virtual_machine_types:
392 if vm_type.lower() == vm.vm_type.lower():
394 raise RuntimeError("Encountered unknown VM type: {}".format(vm_type))
396 def _get_network(self, network_role, props):
397 network_prop = nested_dict.get(props, "properties", "network") or {}
398 name_param = get_param(network_prop) if network_prop else ""
399 for network in self.networks:
400 if network.network_role.lower() == network_role.lower():
402 new_network = Network(network_role, name_param)
403 self.networks.append(new_network)
407 return "VNF Module ({})".format(os.path.basename(self.template_file))
413 return hash(self.vnf_name)
415 def __eq__(self, other):
416 return hash(self) == hash(other)
419 def create_preloads(config, exitstatus):
421 Create preloads in every format that can be discovered by get_generator_plugins
423 if config.getoption("self_test"):
425 print("+===================================================================+")
426 print("| Preload Template Generation |")
427 print("+===================================================================+")
429 preload_dir = os.path.join(get_output_dir(config), "preloads")
430 if os.path.exists(preload_dir):
431 shutil.rmtree(preload_dir)
432 env_directory = config.getoption("env_dir")
433 preload_env = PreloadEnvironment(env_directory) if env_directory else None
434 plugins = get_generator_plugins()
435 available_formats = [p.format_name() for p in plugins]
436 selected_formats = config.getoption("preload_formats") or available_formats
437 heat_templates = get_heat_templates(config)
439 for plugin_class in plugins:
440 if plugin_class.format_name() not in selected_formats:
442 vnf = Vnf(heat_templates)
443 generator = plugin_class(vnf, preload_dir, preload_env)
445 if vnf and vnf.uses_contrail:
447 "\nWARNING: Preload template generation does not support Contrail\n"
448 "at this time, but Contrail resources were detected. The preload \n"
449 "template may be incomplete."
453 "\nWARNING: Heat violations detected. Preload templates may be\n"