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