3 # This program and the accompanying materials
4 # are made available under the terms of the Apache License, Version 2.0
5 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 from urllib2 import Request, urlopen, URLError
18 # ----------------------------------------------------------
22 # -----------------------------------------------------------
23 def get_parameter_from_yaml(parameter, file):
25 Returns the value of a given parameter in file.yaml
26 parameter must be given in string format with dots
27 Example: general.openstack.image_name
30 file_yaml = yaml.safe_load(f)
33 for element in parameter.split("."):
34 value = value.get(element)
36 raise ValueError("The parameter %s is not defined in"
37 " reporting.yaml" % parameter)
41 def get_config(parameter):
42 yaml_ = os.environ["CONFIG_REPORTING_YAML"]
43 return get_parameter_from_yaml(parameter, yaml_)
46 # ----------------------------------------------------------
50 # -----------------------------------------------------------
51 def getLogger(module):
52 logFormatter = logging.Formatter("%(asctime)s [" +
54 "] [%(levelname)-5.5s] %(message)s")
55 logger = logging.getLogger()
56 log_file = get_config('general.log.log_file')
57 log_level = get_config('general.log.log_level')
59 fileHandler = logging.FileHandler("{0}/{1}".format('.', log_file))
60 fileHandler.setFormatter(logFormatter)
61 logger.addHandler(fileHandler)
63 consoleHandler = logging.StreamHandler()
64 consoleHandler.setFormatter(logFormatter)
65 logger.addHandler(consoleHandler)
66 logger.setLevel(log_level)
70 # ----------------------------------------------------------
74 # -----------------------------------------------------------
75 def getApiResults(case, installer, scenario, version):
76 results = json.dumps([])
77 # to remove proxy (to be removed at the end for local test only)
78 # proxy_handler = urllib2.ProxyHandler({})
79 # opener = urllib2.build_opener(proxy_handler)
80 # urllib2.install_opener(opener)
81 # url = "http://127.0.0.1:8000/results?case=" + case + \
82 # "&period=30&installer=" + installer
83 period = get_config('general.period')
84 url_base = get_config('testapi.url')
85 nb_tests = get_config('general.nb_iteration_tests_success_criteria')
87 url = ("http://" + url_base + "?case=" + case +
88 "&period=" + str(period) + "&installer=" + installer +
89 "&scenario=" + scenario + "&version=" + version +
90 "&last=" + str(nb_tests))
91 request = Request(url)
94 response = urlopen(request)
96 results = json.loads(k)
98 print('No kittez. Got an error code:', e)
103 def getScenarios(case, installer, version):
106 case = case.getName()
108 # if case is not an object test case, try the string
109 if type(case) == str:
112 raise ValueError("Case cannot be evaluated")
114 period = get_config('general.period')
115 url_base = get_config('testapi.url')
117 url = ("http://" + url_base + "?case=" + case +
118 "&period=" + str(period) + "&installer=" + installer +
119 "&version=" + version)
120 request = Request(url)
123 response = urlopen(request)
125 results = json.loads(k)
126 test_results = results['results']
127 except URLError as e:
128 print('Got an error code:', e)
130 if test_results is not None:
131 test_results.reverse()
133 scenario_results = {}
135 for r in test_results:
136 # Retrieve all the scenarios per installer
137 if not r['scenario'] in scenario_results.keys():
138 scenario_results[r['scenario']] = []
139 # Do we consider results from virtual pods ...
140 # Do we consider results for non HA scenarios...
141 exclude_virtual_pod = get_config('functest.exclude_virtual')
142 exclude_noha = get_config('functest.exclude_noha')
143 if ((exclude_virtual_pod and "virtual" in r['pod_name']) or
144 (exclude_noha and "noha" in r['scenario'])):
145 print("exclude virtual pod results...")
147 scenario_results[r['scenario']].append(r)
149 return scenario_results
152 def getScenarioStats(scenario_results):
154 for k, v in scenario_results.iteritems():
155 scenario_stats[k] = len(v)
157 return scenario_stats
160 # TODO convergence with above function getScenarios
161 def getScenarioStatus(installer, version):
162 period = get_config('general.period')
163 url_base = get_config('testapi.url')
165 url = ("http://" + url_base + "?case=scenario_status" +
166 "&installer=" + installer +
167 "&version=" + version + "&period=" + str(period))
168 request = Request(url)
171 response = urlopen(request)
174 results = json.loads(k)
175 test_results = results['results']
176 except URLError as e:
177 print('Got an error code:', e)
179 scenario_results = {}
181 if test_results is not None:
182 for r in test_results:
183 if r['stop_date'] != 'None' and r['criteria'] is not None:
184 if not r['scenario'] in scenario_results.keys():
185 scenario_results[r['scenario']] = []
186 scenario_results[r['scenario']].append(r)
188 for k, v in scenario_results.items():
189 # scenario_results[k] = v[:LASTEST_TESTS]
192 if element['criteria'] == 'SUCCESS':
196 result_dict[k] = s_list
198 # return scenario_results
202 def getNbtestOk(results):
205 for k, v in r.iteritems():
210 print("Cannot retrieve test status")
214 def getResult(testCase, installer, scenario, version):
216 # retrieve raw results
217 results = getApiResults(testCase, installer, scenario, version)
218 # let's concentrate on test results only
219 test_results = results['results']
221 # if results found, analyze them
222 if test_results is not None:
223 test_results.reverse()
225 scenario_results = []
227 # print " ---------------- "
229 # print " ---------------- "
230 # print "nb of results:" + str(len(test_results))
232 for r in test_results:
233 # print r["start_date"]
234 # print r["criteria"]
235 scenario_results.append({r["start_date"]: r["criteria"]})
237 scenario_results.sort()
238 # 4 levels for the results
239 # 3: 4+ consecutive runs passing the success criteria
240 # 2: <4 successful consecutive runs but passing the criteria
241 # 1: close to pass the success criteria
242 # 0: 0% success, not passing
243 # -1: no run available
244 test_result_indicator = 0
245 nbTestOk = getNbtestOk(scenario_results)
247 # print "Nb test OK (last 10 days):"+ str(nbTestOk)
248 # check that we have at least 4 runs
249 if len(scenario_results) < 1:
250 # No results available
251 test_result_indicator = -1
253 test_result_indicator = 0
255 test_result_indicator = 1
257 # Test the last 4 run
258 if (len(scenario_results) > 3):
259 last4runResults = scenario_results[-4:]
260 nbTestOkLast4 = getNbtestOk(last4runResults)
261 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
262 if nbTestOkLast4 > 3:
263 test_result_indicator = 3
265 test_result_indicator = 2
267 test_result_indicator = 2
268 return test_result_indicator
271 def getJenkinsUrl(build_tag):
272 # e.g. jenkins-functest-apex-apex-daily-colorado-daily-colorado-246
274 # jenkins-functest-compass-huawei-pod5-daily-master-136
276 # note it is linked to jenkins format
277 # if this format changes...function to be adapted....
278 url_base = get_config('functest.jenkins_url')
280 build_id = [int(s) for s in build_tag.split("-") if s.isdigit()]
281 url_id = (build_tag[8:-(len(str(build_id[0])) + 1)] +
282 "/" + str(build_id[0]))
283 jenkins_url = url_base + url_id + "/console"
285 print('Impossible to get jenkins url:')
287 if "jenkins-" not in build_tag:
293 def getScenarioPercent(scenario_score, scenario_criteria):
296 score = float(scenario_score) / float(scenario_criteria) * 100
298 print('Impossible to calculate the percentage score')
305 def getFunctestConfig(version=""):
306 config_file = get_config('functest.test_conf') + version
307 response = requests.get(config_file)
308 return yaml.safe_load(response.text)
311 def getArchitectures(scenario_results):
312 supported_arch = ['x86']
313 if (len(scenario_results) > 0):
314 for scenario_result in scenario_results.values():
315 for value in scenario_result:
316 if ("armband" in value['build_tag']):
317 supported_arch.append('aarch64')
318 return supported_arch
319 return supported_arch
322 def filterArchitecture(results, architecture):
323 filtered_results = {}
324 for name, results in results.items():
326 for value in results:
327 if (architecture is "x86"):
328 # drop aarch64 results
329 if ("armband" not in value['build_tag']):
330 filtered_values.append(value)
331 elif(architecture is "aarch64"):
333 if ("armband" in value['build_tag']):
334 filtered_values.append(value)
335 if (len(filtered_values) > 0):
336 filtered_results[name] = filtered_values
337 return filtered_results
343 def subfind(given_list, pattern_list):
344 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
345 for i in range(len(given_list)):
346 if given_list[i] == pattern_list[0] and \
347 given_list[i:i + LASTEST_TESTS] == pattern_list:
352 def _get_percent(status):
355 return round(float(status) * 100 / 6, 1)
357 return status * 100 / 6
360 def get_percent(four_list, ten_list):
369 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
370 if four_score == LASTEST_TESTS:
372 elif subfind(ten_list, [1, 1, 1, 1]):
377 status = four_score + 1
379 return _get_percent(status)
383 status = getScenarioStatus("compass", "master")
384 print("status:++++++++++++++++++++++++")
385 print(json.dumps(status, indent=4))
388 # ----------------------------------------------------------
392 # -----------------------------------------------------------
394 def export_csv(scenario_file_name, installer, version):
396 # generate sub files based on scenario_history.txt
397 scenario_installer_file_name = ("./display/" + version +
398 "/functest/scenario_history_" +
400 scenario_installer_file = open(scenario_installer_file_name, "a")
401 with open(scenario_file_name, "r") as f:
402 scenario_installer_file.write("date,scenario,installer,detail,score\n")
404 if installer in line:
405 scenario_installer_file.write(line)
406 scenario_installer_file.close
409 def export_pdf(pdf_path, pdf_doc_name):
411 pdfkit.from_file(pdf_path, pdf_doc_name)
413 print("Error but pdf generated anyway...")
415 print("impossible to generate PDF")