path: root/robotframework-onap/ONAPLibrary/HeatVNFValidation.py

# 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"),
                    }
                )