robotframework-onap/listeners/OVPListener.py

   1 # Copyright 2019 AT&T Intellectual Property. All rights reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 import datetime
  15 import hashlib
  16 import json
  17 import os
  18 from copy import deepcopy
  19 from robot.libraries.BuiltIn import BuiltIn
  20 from zipfile import ZipFile
  21
  22 OUTPUT_DATA = {
  23     "vnf_checksum": "",
  24     "build_tag": "",
  25     "version": "2019.12",
  26     "test_date": "",
  27     "duration": "",
  28     "vnf_type": "heat",
  29     "testcases_list": [
  30         {
  31             "mandatory": "true",
  32             "name": "onap-vvp.validate.heat",
  33             "result": "NOT_STARTED",
  34             "objective": "onap heat template validation",
  35             "sub_testcase": [],
  36             "portal_key_file": "report.json",
  37         },
  38         {
  39             "mandatory": "true",
  40             "name": "onap-vvp.lifecycle_validate.heat",
  41             "result": "NOT_STARTED",
  42             "objective": "onap vnf lifecycle validation",
  43             "sub_testcase": [
  44                 {"name": "model-and-distribute", "result": "NOT_STARTED"},
  45                 {"name": "instantiation", "result": "NOT_STARTED"},
  46             ],
  47             "portal_key_file": "log.html",
  48         },
  49         {
  50             "mandatory": "true",
  51             "name": "stack_validation",
  52             "result": "NOT_STARTED",
  53             "objective": "onap vnf openstack validation",
  54             "sub_testcase": [],
  55             "portal_key_file": "stack_report.json",
  56         },
  57     ],
  58 }
  59
  60
  61 class OVPListener:
  62     ROBOT_LISTENER_API_VERSION = 2
  63
  64     def __init__(self):
  65         self.report = deepcopy(OUTPUT_DATA)
  66
  67         self.build_number = ""
  68         self.build_directory = ""
  69         self.output_directory = ""
  70         self.template_directory = ""
  71
  72     def initialize(self):
  73         self.build_number = BuiltIn().get_variable_value("${GLOBAL_BUILD_NUMBER}")
  74         self.build_directory = BuiltIn().get_variable_value("${BUILD_DIR}")
  75         self.output_directory = BuiltIn().get_variable_value("${OUTPUTDIR}")
  76
  77         self.template_directory = "{}/templates".format(self.build_directory)
  78         self.report["build_tag"] = "vnf-validation-{}".format(self.build_number)
  79         self.report["vnf_checksum"] = sha256(self.template_directory)
  80
  81     def start_test(self, name, attrs):
  82         self.initialize()
  83         date = datetime.datetime.strptime(attrs["starttime"], '%Y%m%d %H:%M:%S.%f').strftime('%Y-%m-%d %H:%M:%S')
  84         self.report["test_date"] = date
  85
  86     def end_keyword(self, name, attrs):
  87         kwname = attrs["kwname"]
  88         status = attrs["status"]
  89
  90         if kwname == "Run VVP Validation Scripts":
  91             self.report["testcases_list"][0]["result"] = status
  92         elif kwname == "Model Distribution For Directory":
  93             self.report["testcases_list"][1]["sub_testcase"][0]["result"] = status
  94         elif kwname == "Instantiate VNF":
  95             self.report["testcases_list"][1]["sub_testcase"][1]["result"] = status
  96             self.report["testcases_list"][1]["result"] = status
  97         elif kwname == "Run VNF Instantiation Report":
  98             self.report["testcases_list"][2]["result"] = status
  99
 100     def end_test(self, name, attrs):
 101         self.report["duration"] = attrs["elapsedtime"] / 1000
 102
 103     def close(self):
 104         with open("{}/summary/results.json".format(self.output_directory), "w") as f:
 105             json.dump(self.report, f, indent=4)
 106
 107
 108 def sha256(template_directory):
 109     heat_sha = None
 110     zip_file = "{}/tmp.zip".format(template_directory)
 111     onlyfiles = [f for f in os.listdir(template_directory) if os.path.isfile(os.path.join(template_directory, f))]
 112
 113     with ZipFile(zip_file, 'w') as zipObj:
 114         for filename in onlyfiles:
 115             zipObj.write(os.path.join(template_directory, filename), arcname=filename)
 116
 117     with open(zip_file, "rb") as f:
 118         bytes = f.read()
 119         heat_sha = hashlib.sha256(bytes).hexdigest()
 120
 121     if os.path.exists(zip_file):
 122         os.remove(zip_file)
 123
 124     return heat_sha