vnfmarket-be/deployment/vtp2ovp/vtp2ovp-result.py

   1 #!/usr/bin/python
   2 # Copyright 2019 Huawei Technologies Co., Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import json
  17 import os
  18 import re
  19 import shutil
  20 import sys
  21 import tarfile
  22 from datetime import datetime
  23 import argparse
  24 import pprint
  25 from argparse import RawTextHelpFormatter
  26
  27
  28 class VTP2OVPResult:
  29     execution_id = None
  30     tar_path = None
  31     vtp_home = None
  32
  33     result_json = {
  34         "testcases_list": [],
  35         "build_tag": "",
  36         "version": "2019.12",
  37         "test_date": "",
  38         "duration": 0,
  39         "vnf_type": "TOSCA",
  40         "vnf_checksum": "",
  41         "validation": "enabled"
  42
  43     }
  44
  45     sub_test = {
  46         "objective": "",
  47         "portal_key_file": "",
  48         "sub_testcase": [],
  49         "mandatory": "true",
  50         "name": "",
  51         "result": ""
  52     }
  53
  54     def __init__(self, execution_id, tar_path, vtp_home):
  55         self.execution_id = execution_id
  56         self.tar_path = tar_path
  57         self.vtp_home = vtp_home + "/data/executions"
  58
  59     def find_folder_by_execution_id(self):
  60         ret_dir_name = None
  61
  62         # TODO: Improve by better seraching.
  63         dir_obj = os.listdir(self.vtp_home)
  64         for dir_name in dir_obj:
  65             if dir_name.__contains__(self.execution_id):
  66                 ret_dir_name = dir_name
  67                 break
  68
  69         if not ret_dir_name:
  70             raise Exception('Given execution-id does not exist\n')
  71
  72         return ret_dir_name
  73
  74     def extract_execution_id_from_name(self, dir_name=None):
  75
  76         temp_arr = dir_name.split("__")
  77         exec_id = temp_arr[0]
  78         temp_arr.remove(exec_id)
  79         nm_arr = []
  80         nm_arr.extend(temp_arr)
  81         objective = "__".join(nm_arr)
  82         return exec_id + "," + objective
  83
  84     def extract_request_id_from_execution_id(self, exec_id):
  85         request_id = exec_id.split("-")[0]
  86         return request_id
  87
  88     def find_folder_by_request_id(self, request_id):
  89         file_list = []
  90         if os.path.exists(self.vtp_home):
  91             for dir_name in os.listdir(self.vtp_home):
  92                 if re.match('^' + request_id + '-', dir_name):
  93                     if not dir_name.__contains__(self.execution_id):
  94                         file_list.append(dir_name)
  95         return file_list
  96
  97     def check_pass_fail(self, dir_path):
  98         completed_path = dir_path + "/completed"
  99         if os.path.isfile(completed_path):
 100             return "PASS"
 101         else:
 102             return "FAIL"
 103
 104     def get_testdate(self, dir_path=None):
 105         t = os.path.getmtime(dir_path)
 106         return str(datetime.utcfromtimestamp(t)) + " UTC"
 107
 108     def get_duration(self, dir_path=None):
 109         input_file = dir_path + "/input"
 110         output_file = dir_path + "/output"
 111         t1 = os.path.getmtime(input_file)
 112         input_timestamp = datetime.utcfromtimestamp(t1)
 113         t2 = os.path.getmtime(output_file)
 114         output_timestamp = datetime.utcfromtimestamp(t2)
 115         time_diff = output_timestamp - input_timestamp
 116         return time_diff.seconds * 1000
 117
 118     def generate_ovp_result_json(self):
 119         arr_temp = [self.sub_test]
 120         self.result_json.update({"testcases_list": arr_temp})
 121         with open('./result.json', 'w') as json_file:
 122             json.dump(self.result_json, json_file, default=str, indent=4)
 123         return json_file
 124
 125     def generate_result_tar(self):
 126         arr_sub_testcase = []
 127         dir_name = None
 128         try:
 129             dir_name = self.find_folder_by_execution_id()
 130             exec_id = self.extract_execution_id_from_name(dir_name)
 131             objective = exec_id.split(',')[1]
 132             self.sub_test['objective'] = objective
 133             output_file_path = "/data/executions/" + dir_name + "/output"
 134             self.sub_test.update({'portal_key_file': output_file_path})
 135             self.sub_test['name'] = objective
 136             self.result_json.update({'duration': str(self.get_duration(self.vtp_home + "/" + dir_name)) + " ms"})
 137             self.sub_test['result'] = self.check_pass_fail(self.vtp_home + "/" + dir_name)
 138             test_date = self.get_testdate(self.vtp_home + "/" + dir_name)
 139             self.result_json.update({'test_date': test_date})
 140             self.result_json.update({'build_tag': self.execution_id})
 141             request_id = self.extract_request_id_from_execution_id(exec_id)
 142             file_list = self.find_folder_by_request_id(request_id)
 143             for sub_test in file_list:
 144                 sub_test_case = {}
 145                 sub_exec_id = self.extract_execution_id_from_name(sub_test)
 146                 exec_id_name = sub_exec_id.split(",")
 147                 sub_test_name = exec_id_name[1]
 148                 sub_test_case['name'] = sub_test_name
 149                 full_name = "__".join(exec_id_name)
 150                 pass_fail = self.check_pass_fail(self.vtp_home + "/" + full_name)
 151                 sub_test_case['result'] = pass_fail
 152                 arr_sub_testcase.append(sub_test_case)
 153             self.sub_test['sub_testcase'] = arr_sub_testcase
 154             self.generate_ovp_result_json()
 155             self.make_tar_dir()
 156             self.display_result_json()
 157
 158         except Exception as e:
 159             print str(e)
 160
 161     def display_result_json(self):
 162         print "\nOVP result JSON: "
 163         pretty_printer = pprint.PrettyPrinter(indent=4)
 164         pretty_printer.pprint(self.result_json)
 165
 166     def make_tar_dir(self):
 167         tar_file_name = "vtp-ovp-result-" + self.execution_id + ".tar"
 168         head, tail = os.path.split(self.vtp_home)
 169         with tarfile.open(tar_file_name, "w:gz") as tar_handle:
 170             tar_handle.add('./result.json')
 171             tar_handle.add(head, arcname='./data')
 172
 173         if not os.path.exists(self.tar_path):
 174             os.mkdir(self.tar_path)
 175         elif not os.path.isdir(self.tar_path):
 176             raise Exception('Given path for OVP output tar file is not a directory\n')
 177         shutil.move(tar_file_name, os.path.join(self.tar_path, tar_file_name))
 178         if os.path.exists('result.json'):
 179             os.remove('result.json')
 180         print "\n\nOVP Result Tar file: " + os.path.abspath(self.tar_path + '/' + tar_file_name)
 181
 182
 183 def main():
 184     text = 'This command helps to produce VTP results in LFV OVP format. ' \
 185            'Run this command from the server machine, where VTP is running'
 186     parser = argparse.ArgumentParser(description=text, formatter_class=RawTextHelpFormatter)
 187     parser.add_argument('--execution-id', action='store', dest='execution_id',
 188                         help='VTP test case execution id')
 189     parser.add_argument('--tar-path', action='store', dest='tar_path',
 190                         help='Location to OVP output tar file')
 191     parser.add_argument('--vtp-home', default=os.getenv('OPEN_CLI_HOME'), action='store', dest='vtp_home',
 192                         help='VTP installation home directory')
 193     args = parser.parse_args()
 194
 195     if not args.execution_id or not args.tar_path or not args.vtp_home:
 196         sys.stderr.write(str('Too few arguments\n'))
 197     else:
 198         if args.execution_id:
 199             execution_id = args.execution_id
 200         if args.tar_path:
 201             tar_path = args.tar_path
 202         if args.vtp_home:
 203             vtp_home = args.vtp_home
 204
 205         if os.path.exists(vtp_home):
 206             if not os.path.isdir(vtp_home):
 207                 print 'Given VTP path is not a directory\n'
 208             else:
 209                 vtp2ovp = VTP2OVPResult(execution_id, tar_path, vtp_home)
 210                 vtp2ovp.generate_result_tar()
 211         else:
 212             print 'Given VTP path does not exist\n'
 213
 214
 215 """
 216     usage: vtp2ovp-result.py [-h] [--execution-id EXECUTION_ID]
 217                              [--tar-path TAR_PATH] [--vtp-home VTP_HOME]
 218
 219     This command helps to produce VTP results in LFV OVP format.
 220     Run this command from the server machine, where VTP is running
 221
 222     optional arguments:
 223       -h, --help            show this help message and exit
 224       --execution-id EXECUTION_ID
 225                             VTP test case execution id
 226       --tar-path TAR_PATH   Location to OVP output tar file
 227       --vtp-home VTP_HOME   VTP installation home directory
 228
 229 """
 230 if __name__ == '__main__':
 231     main()