--- /dev/null
+# Copyright 2019 AT&T Intellectual Property. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from robot.api.deco import keyword
+
+import json
+import yaml
+import requests
+
+stack_url = "{}/stacks/{}"
+stack_resources_url = "{}/stacks/{}/resources"
+
+
+# use this to import and run validation from robot
+class HeatVNFValidation:
+ def __init__(self):
+ pass
+
+ @keyword
+ def validate(self, orchestration_url, token, manifest, vnf_name):
+ validator = StackValidation(orchestration_url, token, manifest, vnf_name)
+ validator.create_summary()
+ validator.validate_summary()
+
+ return validator.report
+
+
+class StackValidation:
+ def __init__(self, orchestration_url, token, manifest, vnf_name):
+ """retrieves stack and template details, and creates
+ a report for submission to OVP portal.
+
+ :orchestration_url heat service endpoint in openstack
+ :token keystone auth token
+ :manifest json that contains list of heat templates, env,
+ preloads, and stack names for each module in
+ a VNF
+ """
+ self.modules = []
+ self.url = orchestration_url
+ self.token = token
+ self.manifest = manifest
+ self.vnf_name = vnf_name
+ self.report = {}
+
+ self.load_manifest()
+
+ def load_manifest(self):
+ for entry in self.manifest:
+ template = entry.get("template_name")
+ env_file = template.replace(".yaml", ".env").replace(".yml", ".env")
+ preload = entry.get("preload_name")
+ stack = entry.get("stack_name")
+ module = HeatModule(
+ template,
+ env_file,
+ stack,
+ preload
+ )
+ module.get_data(self.url, self.token)
+ self.modules.append(module)
+
+ def create_summary(self):
+ """creates a report dictionary to compare stack
+ resources, parameters, outputs w/ template"""
+ self.report["modules"] = []
+ self.report["VNF Name"] = self.vnf_name
+ for module in self.modules:
+ stack = module.stack
+ preload = module.preload
+ template = module.template
+
+ module_report = {}
+ module_report["stack_details"] = stack.stack_details
+
+ module_report["resources"] = {}
+ module_report["resources"]["summary"] = ""
+
+ module_report["parameters"] = {}
+ module_report["parameters"]["summary"] = ""
+
+ module_report["outputs"] = {}
+ module_report["outputs"]["summary"] = ""
+
+ module_report["resources"]["stack_resources"] = stack.resources
+ module_report["resources"]["template_resources"] = template.resources
+
+ module_report["parameters"]["stack_parameters"] = stack.parameters
+ module_report["parameters"]["template_parameters"] = dict(template.parameters, **preload.parameters)
+
+ module_report["outputs"]["stack_outputs"] = stack.outputs
+ module_report["outputs"]["template_outputs"] = template.outputs
+
+ self.report["modules"].append(module_report)
+
+ def validate_summary(self):
+ # validates resources, parameters, and outputs
+ self.validate_resources()
+ self.validate_parameters()
+ self.validate_outputs()
+
+ self.report["summary"] = "SUCCESS"
+ for module in self.report["modules"]:
+ if module["resources"]["summary"] != "SUCCESS":
+ self.report["summary"] = "FAILED"
+ break
+ if module["parameters"]["summary"] != "SUCCESS":
+ self.report["summary"] = "FAILED"
+ break
+ if module["outputs"]["summary"] != "SUCCESS":
+ self.report["summary"] = "FAILED"
+ break
+
+ def validate_resources(self):
+ """validates that all resources from a heat template
+ are present in instantiated heat stack"""
+ report = self.report
+ for module in report["modules"]:
+ module["resources"]["summary"] = "SUCCESS"
+ resources = module.get("resources", {})
+ template_resources = resources.get("template_resources", [])
+ stack_resources = resources.get("stack_resources", [])
+
+ if len(stack_resources) != len(template_resources):
+ module["resources"]["summary"] = "FAILED"
+ continue
+
+ stack_rids = []
+ for s_resource in stack_resources:
+ stack_rids.append(s_resource.get("resource_id"))
+
+ template_rids = []
+ for t_resource in template_resources:
+ template_rids.append(t_resource.get("resource_id"))
+
+ if stack_rids.sort() != template_rids.sort():
+ module["resources"]["summary"] = "FAILED"
+ continue
+
+ def validate_parameters(self):
+ """validates that parameter name/value from template
+ == values from instantiated heat stack"""
+ report = self.report
+ for module in report["modules"]:
+ module["parameters"]["summary"] = "SUCCESS"
+ parameters = module.get("parameters", {})
+ template_parameters = parameters.get("template_parameters", {})
+ stack_parameters = parameters.get("stack_parameters", {})
+
+ for parameter, parameter_value in template_parameters.items():
+ stack_parameter = stack_parameters.get(parameter)
+ if not stack_parameter:
+ module["parameters"]["summary"] = "FAILED"
+ break
+
+ if stack_parameter != parameter_value:
+ module["parameters"]["summary"] = "FAILED"
+ break
+
+ def validate_outputs(self):
+ """validates that all outputs from a heat template
+ are present in instantiated heat stack"""
+ report = self.report
+ for module in report["modules"]:
+ module["outputs"]["summary"] = "SUCCESS"
+ outputs = module.get("outputs", {})
+ template_outputs = outputs.get("template_outputs", {})
+ stack_outputs = outputs.get("stack_outputs", [])
+
+ for output in stack_outputs:
+ output_key = output.get("output_key")
+ if output_key not in template_outputs:
+ module["outputs"]["summary"] = "FAILED"
+ break
+
+
+class HeatModule:
+ def __init__(self, heat_template, environment_file, stack_name, preload):
+ """
+ creates module object that has stack, preload, and template objects
+
+ :heat_template /path/to/heat/template.yaml
+ :environment_file /path/to/heat/env.env
+ :preload /path/to/preloads/file.json
+ :stack_name name of heat stack in openstack
+ """
+ self.stack = HeatStack(stack_name)
+ self.template = HeatTemplate(heat_template, environment_file)
+ self.preload = HeatPreload(preload)
+
+ def get_data(self, url, token):
+ self.stack.get_data(url, token)
+ self.template.get_data()
+ self.preload.get_data()
+
+
+class HeatTemplate:
+ def __init__(self, heat_template, environment_file):
+ """
+ creates template object that holds template resources,
+ parameters, and outputs of a heat template/env pair.
+
+ :heat_template /path/to/heat/template.yaml
+ :environment_file /path/to/heat/env.env
+ """
+ self.template = heat_template
+ self.env = environment_file
+ self.resources = []
+ self.parameters = {}
+ self.outputs = []
+
+ def get_data(self):
+ with open(self.template, "r") as f:
+ ydata = yaml.safe_load(f)
+
+ resources = ydata.get("resources", {})
+
+ for rid, resource in resources.items():
+ self.resources.append(
+ {"resource_id": rid, "resource_type": resource.get("type", "")}
+ )
+
+ outputs = ydata.get("outputs", {})
+
+ for output, output_value in outputs.items():
+ self.outputs.append(output)
+
+ with open(self.env, "r") as f:
+ ydata = yaml.safe_load(f)
+
+ self.parameters = ydata.get("parameters", {})
+
+
+class HeatPreload:
+ def __init__(self, preload):
+ """
+ creates preload object that holds parameter name/values
+
+ :preload /path/to/preloads/file.json
+ """
+ self.preload = preload
+ self.parameters = {}
+
+ def get_data(self):
+ with open(self.preload, "r") as f:
+ jdata = json.loads(f.read())
+
+ # get parameters regardless of API version
+
+ vnf_api_parameters = (
+ jdata.get("input", {})
+ .get("vnf-topology-information", {})
+ .get("vnf-parameters", [])
+ )
+
+ for parameter in vnf_api_parameters:
+ p_name = parameter.get("vnf-parameter-name")
+ p_value = parameter.get("vnf-parameter-value")
+ self.parameters[p_name] = p_value
+
+ gr_api_parameters = (
+ jdata.get("input", {})
+ .get("preload-vf-module-topology-information", {})
+ .get("vf-module-topology", {})
+ .get("vf-module-parameters", {})
+ .get("param", [])
+ )
+
+ for parameter in gr_api_parameters:
+ p_name = parameter.get("name")
+ p_value = parameter.get("value")
+ self.parameters[p_name] = p_value
+
+
+class HeatStack:
+ def __init__(self, stack_name):
+ """
+ creates stack object that hold stack resources,
+ parameters, and outputs
+
+ :stack_name name of heat stack in openstack
+ """
+ self.stack_name = stack_name
+ self.resources = []
+ self.parameters = {}
+ self.outputs = []
+ self.status = ""
+ self.stack_details = {}
+
+ def get_data(self, orchestration_url, token):
+ url = stack_url.format(orchestration_url, self.stack_name)
+ r = requests.get(headers={"X-Auth-Token": token}, url=url)
+
+ if r.status_code == 200:
+ response = r.json()
+ self.parameters = response.get("stack", {}).get("parameters", {})
+ self.outputs = response.get("stack", {}).get("outputs", {})
+ self.status = response.get("stack", {}).get("stack_status", "")
+ self.stack_details = response.get("stack", {})
+
+ url = stack_resources_url.format(orchestration_url, self.stack_name)
+ r = requests.get(headers={"X-Auth-Token": token}, url=url)
+ if r.status_code == 200:
+ response = r.json()
+ resources = response.get("resources", [])
+ for resource in resources:
+ self.resources.append(
+ {
+ "resource_id": resource.get("resource_name"),
+ "resource_type": resource.get("resource_type"),
+ "resource_status": resource.get("resource_status"),
+ }
+ )
Code Review / testsuite / python-testing-utils.git / blobdiff