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
|
import pytest
import os
import json
import pathlib
import shutil
import plugins.agl_test_utils as utils
import plugins.agl_test_conf as conf
import plugins.agl_test_log as log
import plugins.agl_test_report as report
class AGLBaseTest:
name: str
case_info_list = dict()
def __init__(self, name: str):
self.name=name
utils.create_dir(self.name)
def get_name(self):
return self.name
def get_temp_logdir(self):
return conf.TMP_LOGS_DIR + self.name
def get_temp_logfile(self):
return conf.get_log_file(self.name)
def get_logfile(self):
return conf.get_log_file(self.name)
def get_workdir(self):
return conf.WORK_DIR + self.name.replace('-','_') + "/resource/"
def get_spec_path(self):
return conf.WORK_DIR + self.name.replace('-','_') + "/spec.json"
def append_one_caseinfo(name, value, status):
self.case_info_list[name] = [name, value, status]
def get_all_caseinfo(self):
return self.case_info_list
def get_caseinfo_by_name(self, name):
return self.case_info_list[name]
def update_caseinfo_by_name(self, name, case_info):
self.case_info_list[name] = case_info
def log_process(self):
logfile = self.get_logfile()
self.case_info_list = log.log_process(logfile)
def init_case_status(self):
for key in self.case_info_list:
case_info = self.case_info_list[key]
case_info[2] = "skipped"
#if (case_info[1] == "TEST-PASS"):
# case_info[2] = "passed"
#if (case_info[1] == "TEST-FAIL"):
# case_info[2] = "failed"
#if (case_info[1] == "TEST-SKIP"):
# case_info[2] = "skipped"
self.update_caseinfo_by_name(key, case_info)
def log_report_json(self):
#Get case status list
#case_status format
# {
# 'test_id': 'status',
# 'test_id': 'status'
# }
case_status = report.format_caselist(self.case_info_list)
#Get the summary status of the test set.
#summary format
#{
# 'collected': collected_num,
# 'passed': passed_num,
# 'failed': failed_num,
# 'skipped": skipped_num
# }
summary = report.format_summary(self.case_info_list)
#Get test set status
test_set_status = self.get_test_set_status(summary)
#Format data for json log
data = report.format_json_data(self.name, test_set_status, summary, case_status)
report.write_data_to_file(data)
#Get test set status
#default output:
# passed: there is no failed case
# failed: there is one or more failed case
# skipped: all case is skipped or no case is run
def get_test_set_status(self, summary):
#Judge whether the test set passes
test_set_status = None
if (summary["collected"] == summary["skipped"]):
test_set_status = "skipped"
elif (summary["failed"] == 0):
test_set_status = "passed"
else:
test_set_status = "failed"
return test_set_status
def log_report(self):
self.log_report_json()
#Package log file
report.log_compress(self.name)
#Write json data to html
report.change_json_to_html(self.name)
#Clean temp files and folders
self.clean_self_tmp()
def precheck(self):
return True;
def write_skip_info(self):
test_info = {"status":"skip","path":""}
self.write_info_to_file(test_info)
def write_run_info(self):
path_str = self.name + "/report.json"
test_info = {"status":"run","path":path_str}
self.write_info_to_file(test_info)
def write_info_to_file(self, test_info):
time_stamp = os.getenv("TIME_STAMP")
test_list = "/var/run/agl-test/logs/tmp-log/test_list_" + time_stamp + ".json"
with open(test_list, 'r') as f:
test_suites = json.load(f)
f.close()
with open(test_list, 'w') as f:
test_suites[self.name] = test_info
json.dump(test_suites,f)
f.close()
def prepare_local_log_mode(self):
# Get env 'LOG_MODE'
env_dist=os.environ
local_log_mode = str(env_dist.get("LOG_MODE"))
return local_log_mode
def clean_self_tmp(self):
path = self.get_temp_logdir()
log_mode = self.prepare_local_log_mode()
if log_mode == "slim" or log_mode == "clear":
# leave <path>/log and report.json as essential
folderObj = pathlib.Path(path)
for element in folderObj.iterdir():
if element.name != "log" and element.name != "report.json":
# check delete_file_path file type
if element.is_file():
element.unlink()
elif element.is_dir():
shutil.rmtree(path)
|
