--- /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"),
+ }
+ )
--- /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 os
+import subprocess
+
+VVP_BRANCH = "master"
+VVP_URL = "https://gerrit.onap.org/r/vvp/validation-scripts"
+
+
+# use this to import and run validation from robot
+class HeatValidationScripts:
+ def __init__(self):
+ pass
+
+ @keyword
+ def validate(self, build_dir, template_directory, output_directory):
+ """
+ keyword invoked by robot to execute VVP validation scripts
+
+ :build_dir: directory to install virtualenv
+ and clone validation scripts
+ :template_directory: directory with heat templates
+ :output_directory: directory to store output files
+ """
+ t = VVP(build_dir, template_directory, output_directory)
+ t.install_requirements()
+ status = t.run_vvp()
+
+ return status
+
+
+class VVP:
+ def __init__(self, build_dir, template_directory, output_directory):
+ self._build_dir = build_dir
+ self.initialize()
+
+ self.virtualenv = "{}/test_env".format(build_dir)
+ self.vvp = "{}/validation_scripts".format(build_dir)
+ self.template_directory = template_directory
+ self.output_directory = output_directory
+
+ def initialize(self):
+ self.create_venv(self._build_dir)
+ self.clone_vvp(self._build_dir)
+
+ def create_venv(self, build_dir):
+ try:
+ subprocess.call(
+ ["python3.7", "-m", "virtualenv", "--clear", "{}/test_env".format(build_dir)]
+ )
+ except OSError as e:
+ print("error creating virtual environment for vvp {}".format(e))
+ raise
+
+ def clone_vvp(self, build_dir):
+ if not os.path.exists("{}/validation_scripts".format(build_dir)):
+ try:
+ subprocess.call(
+ [
+ "git",
+ "clone",
+ "-b",
+ VVP_BRANCH,
+ VVP_URL,
+ "{}/validation_scripts".format(build_dir),
+ ]
+ )
+ except OSError as e:
+ print("error cloning vvp validation scripts {}".format(e))
+ raise
+
+ def install_requirements(self):
+ try:
+ subprocess.call(
+ [
+ "{}/bin/python".format(self.virtualenv),
+ "-m",
+ "pip",
+ "install",
+ "--upgrade",
+ "pip",
+ "wheel",
+ ]
+ )
+ subprocess.call(
+ [
+ "{}/bin/python".format(self.virtualenv),
+ "-m",
+ "pip",
+ "install",
+ "wheel",
+ "-r",
+ "{}/requirements.txt".format(self.vvp),
+ ]
+ )
+ except OSError as e:
+ print("error installing vvp requirements {}".format(e))
+ raise
+
+ def run_vvp(self):
+ try:
+ ret = subprocess.call(
+ [
+ "{}/bin/python".format(self.virtualenv),
+ "-m",
+ "pytest",
+ "--rootdir={}/ice_validator/".format(self.vvp),
+ "--template-directory={}".format(self.template_directory),
+ "--output-directory={}".format(self.output_directory),
+ "{}/ice_validator/tests/".format(self.vvp),
+ ]
+ )
+ except OSError as e:
+ print("error running vvp validation scripts {}".format(e))
+ raise
+
+ if ret != 0:
+ raise ValueError("Validation Script error detected")
--- /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.
+import datetime
+import hashlib
+import json
+import os
+from copy import deepcopy
+from robot.libraries.BuiltIn import BuiltIn
+from zipfile import ZipFile
+
+OUTPUT_DATA = {
+ "vnf_checksum": "",
+ "build_tag": "",
+ "version": "2019.09",
+ "test_date": "",
+ "duration": "",
+ "vnf_type": "heat",
+ "testcases_list": [
+ {
+ "mandatory": "true",
+ "name": "onap-vvp.validate.heat",
+ "result": "NOT_STARTED",
+ "objective": "onap heat template validation",
+ "sub_testcase": [],
+ "portal_key_file": "report.json",
+ },
+ {
+ "mandatory": "true",
+ "name": "onap-vvp.lifecycle_validate.heat",
+ "result": "NOT_STARTED",
+ "objective": "onap vnf lifecycle validation",
+ "sub_testcase": [
+ {"name": "model-and-distribute", "result": "NOT_STARTED"},
+ {"name": "instantiation", "result": "NOT_STARTED"},
+ ],
+ "portal_key_file": "log.html",
+ },
+ {
+ "mandatory": "true",
+ "name": "stack_validation",
+ "result": "NOT_STARTED",
+ "objective": "onap vnf openstack validation",
+ "sub_testcase": [],
+ "portal_key_file": "stack_report.json",
+ },
+ ],
+}
+
+
+class OVPListener:
+ ROBOT_LISTENER_API_VERSION = 2
+
+ def __init__(self):
+ self.report = deepcopy(OUTPUT_DATA)
+
+ self.build_number = ""
+ self.build_directory = ""
+ self.output_directory = ""
+ self.template_directory = ""
+
+ def initialize(self):
+ self.build_number = BuiltIn().get_variable_value("${GLOBAL_BUILD_NUMBER}")
+ self.build_directory = BuiltIn().get_variable_value("${BUILD_DIR}")
+ self.output_directory = BuiltIn().get_variable_value("${OUTPUTDIR}")
+
+ self.template_directory = "{}/templates".format(self.build_directory)
+ self.report["build_tag"] = "vnf-validation-{}".format(self.build_number)
+ self.report["vnf_checksum"] = sha256(self.template_directory)
+
+ def start_test(self, name, attrs):
+ self.initialize()
+ date = datetime.datetime.strptime(attrs["starttime"], '%Y%m%d %H:%M:%S.%f').strftime('%Y-%m-%d %H:%M:%S')
+ self.report["test_date"] = date
+
+ def end_keyword(self, name, attrs):
+ kwname = attrs["kwname"]
+ status = attrs["status"]
+
+ if kwname == "Run VVP Validation Scripts":
+ self.report["testcases_list"][0]["result"] = status
+ elif kwname == "Model Distribution For Directory":
+ self.report["testcases_list"][1]["sub_testcase"][0]["result"] = status
+ elif kwname == "Instantiate VNF":
+ self.report["testcases_list"][1]["sub_testcase"][1]["result"] = status
+ self.report["testcases_list"][1]["result"] = status
+ elif kwname == "Run VNF Instantiation Report":
+ self.report["testcases_list"][2]["result"] = status
+
+ def end_test(self, name, attrs):
+ self.report["duration"] = attrs["elapsedtime"] / 1000
+
+ def close(self):
+ with open("{}/summary/results.json".format(self.output_directory), "w") as f:
+ json.dump(self.report, f, indent=4)
+
+
+def sha256(template_directory):
+ heat_sha = None
+
+ if os.path.exists(template_directory):
+ zip_file = "{}/tmp_heat.zip".format(template_directory)
+ with ZipFile(zip_file, "w") as zip_obj:
+ for folder_name, subfolders, filenames in os.walk(template_directory):
+ for filename in filenames:
+ file_path = os.path.join(folder_name, filename)
+ zip_obj.write(file_path)
+
+ with open(zip_file, "rb") as f:
+ bytes = f.read()
+ heat_sha = hashlib.sha256(bytes).hexdigest()
+
+ os.remove(zip_file)
+
+ return heat_sha
Code Review / testsuite / python-testing-utils.git / commitdiff