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, config_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
29 with open(config_file) as my_file:
30 file_yaml = yaml.safe_load(my_file)
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):
43 Get configuration parameter from yaml configuration file
45 yaml_ = os.environ["CONFIG_REPORTING_YAML"]
46 return get_parameter_from_yaml(parameter, yaml_)
49 # ----------------------------------------------------------
53 # -----------------------------------------------------------
54 def getLogger(module):
58 log_formatter = logging.Formatter("%(asctime)s [" +
60 "] [%(levelname)-5.5s] %(message)s")
61 logger = logging.getLogger()
62 log_file = get_config('general.log.log_file')
63 log_level = get_config('general.log.log_level')
65 file_handler = logging.FileHandler("{0}/{1}".format('.', log_file))
66 file_handler.setFormatter(log_formatter)
67 logger.addHandler(file_handler)
69 console_handler = logging.StreamHandler()
70 console_handler.setFormatter(log_formatter)
71 logger.addHandler(console_handler)
72 logger.setLevel(log_level)
76 # ----------------------------------------------------------
80 # -----------------------------------------------------------
81 def getApiResults(case, installer, scenario, version):
83 Get Results by calling the API
85 results = json.dumps([])
86 # to remove proxy (to be removed at the end for local test only)
87 # proxy_handler = urllib2.ProxyHandler({})
88 # opener = urllib2.build_opener(proxy_handler)
89 # urllib2.install_opener(opener)
90 # url = "http://127.0.0.1:8000/results?case=" + case + \
91 # "&period=30&installer=" + installer
92 period = get_config('general.period')
93 url_base = get_config('testapi.url')
94 nb_tests = get_config('general.nb_iteration_tests_success_criteria')
96 url = ("http://" + url_base + "?case=" + case +
97 "&period=" + str(period) + "&installer=" + installer +
98 "&scenario=" + scenario + "&version=" + version +
99 "&last=" + str(nb_tests))
100 request = Request(url)
103 response = urlopen(request)
105 results = json.loads(k)
107 print "Error when retrieving results form API"
112 def getScenarios(project, case, installer, version):
114 Get the list of Scenarios
117 period = get_config('general.period')
118 url_base = get_config('testapi.url')
120 url = ("http://" + url_base +
121 "?installer=" + installer +
122 "&period=" + str(period))
124 if version is not None:
125 url += "&version=" + version
127 if project is not None:
128 url += "&project=" + project
131 url += "&case=" + case
134 request = Request(url)
135 response = urlopen(request)
137 results = json.loads(k)
138 test_results = results['results']
140 page = results['pagination']['total_pages']
143 for i in range(1, page + 1):
144 url_page = url + "&page=" + str(i)
145 request = Request(url_page)
146 response = urlopen(request)
148 results = json.loads(k)
149 test_results += results['results']
151 print "No pagination detected"
152 except URLError as err:
153 print 'Got an error code: {}'.format(err)
155 if test_results is not None:
156 test_results.reverse()
157 scenario_results = {}
159 for my_result in test_results:
160 # Retrieve all the scenarios per installer
161 if not my_result['scenario'] in scenario_results.keys():
162 scenario_results[my_result['scenario']] = []
163 # Do we consider results from virtual pods ...
164 # Do we consider results for non HA scenarios...
165 exclude_virtual_pod = get_config('functest.exclude_virtual')
166 exclude_noha = get_config('functest.exclude_noha')
167 if ((exclude_virtual_pod and "virtual" in my_result['pod_name']) or
168 (exclude_noha and "noha" in my_result['scenario'])):
169 print "exclude virtual pod results..."
171 scenario_results[my_result['scenario']].append(my_result)
173 return scenario_results
176 def getScenarioStats(scenario_results):
178 Get the number of occurence of scenarios over the defined PERIOD
181 for res_k, res_v in scenario_results.iteritems():
182 scenario_stats[res_k] = len(res_v)
183 return scenario_stats
186 def getScenarioStatus(installer, version):
188 Get the status of a scenariofor Yardstick
190 period = get_config('general.period')
191 url_base = get_config('testapi.url')
193 url = ("http://" + url_base + "?case=scenario_status" +
194 "&installer=" + installer +
195 "&version=" + version + "&period=" + str(period))
196 request = Request(url)
199 response = urlopen(request)
202 results = json.loads(k)
203 test_results = results['results']
205 print "GetScenarioStatus: error when calling the API"
209 scenario_results = {x86: {}, aarch64: {}}
210 result_dict = {x86: {}, aarch64: {}}
211 if test_results is not None:
212 for test_r in test_results:
213 if (test_r['stop_date'] != 'None' and
214 test_r['criteria'] is not None):
215 scenario_name = test_r['scenario']
216 if 'arm' in test_r['pod_name']:
217 if not test_r['scenario'] in scenario_results[aarch64]:
218 scenario_results[aarch64][scenario_name] = []
219 scenario_results[aarch64][scenario_name].append(test_r)
221 if not test_r['scenario'] in scenario_results[x86]:
222 scenario_results[x86][scenario_name] = []
223 scenario_results[x86][scenario_name].append(test_r)
225 for key in scenario_results:
226 for scen_k, scen_v in scenario_results[key].items():
227 # scenario_results[k] = v[:LASTEST_TESTS]
229 for element in scen_v:
230 if element['criteria'] == 'PASS':
234 result_dict[key][scen_k] = s_list
236 # return scenario_results
240 def getQtipResults(version, installer):
244 period = get_config('qtip.period')
245 url_base = get_config('testapi.url')
247 url = ("http://" + url_base + "?project=qtip" +
248 "&installer=" + installer +
249 "&version=" + version + "&period=" + str(period))
250 request = Request(url)
253 response = urlopen(request)
256 results = json.loads(k)['results']
257 except URLError as err:
258 print 'Got an error code: {}'.format(err)
263 key = '{}/{}'.format(r['pod_name'], r['scenario'])
264 if key not in result_dict.keys():
265 result_dict[key] = []
266 result_dict[key].append(r['details']['score'])
268 # return scenario_results
272 def getNbtestOk(results):
274 based on default value (PASS) count the number of test OK
277 for my_result in results:
278 for res_k, res_v in my_result.iteritems():
283 print "Cannot retrieve test status"
287 def getResult(testCase, installer, scenario, version):
289 Get Result for a given Functest Testcase
291 # retrieve raw results
292 results = getApiResults(testCase, installer, scenario, version)
293 # let's concentrate on test results only
294 test_results = results['results']
296 # if results found, analyze them
297 if test_results is not None:
298 test_results.reverse()
300 scenario_results = []
302 # print " ---------------- "
304 # print " ---------------- "
305 # print "nb of results:" + str(len(test_results))
307 for res_r in test_results:
308 # print r["start_date"]
309 # print r["criteria"]
310 scenario_results.append({res_r["start_date"]: res_r["criteria"]})
312 scenario_results.sort()
313 # 4 levels for the results
314 # 3: 4+ consecutive runs passing the success criteria
315 # 2: <4 successful consecutive runs but passing the criteria
316 # 1: close to pass the success criteria
317 # 0: 0% success, not passing
318 # -1: no run available
319 test_result_indicator = 0
320 nbTestOk = getNbtestOk(scenario_results)
322 # print "Nb test OK (last 10 days):"+ str(nbTestOk)
323 # check that we have at least 4 runs
324 if len(scenario_results) < 1:
325 # No results available
326 test_result_indicator = -1
328 test_result_indicator = 0
330 test_result_indicator = 1
332 # Test the last 4 run
333 if len(scenario_results) > 3:
334 last4runResults = scenario_results[-4:]
335 nbTestOkLast4 = getNbtestOk(last4runResults)
336 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
337 if nbTestOkLast4 > 3:
338 test_result_indicator = 3
340 test_result_indicator = 2
342 test_result_indicator = 2
343 return test_result_indicator
346 def getJenkinsUrl(build_tag):
348 Get Jenkins url_base corespoding to the last test CI run
349 e.g. jenkins-functest-apex-apex-daily-colorado-daily-colorado-246
351 jenkins-functest-compass-huawei-pod5-daily-master-136
353 note it is linked to jenkins format
354 if this format changes...function to be adapted....
356 url_base = get_config('functest.jenkins_url')
358 build_id = [int(s) for s in build_tag.split("-") if s.isdigit()]
359 url_id = (build_tag[8:-(len(str(build_id[0])) + 1)] +
360 "/" + str(build_id[0]))
361 jenkins_url = url_base + url_id + "/console"
363 print 'Impossible to get jenkins url:'
365 if "jenkins-" not in build_tag:
371 def getScenarioPercent(scenario_score, scenario_criteria):
373 Get success rate of the scenario (in %)
377 score = float(scenario_score) / float(scenario_criteria) * 100
379 print 'Impossible to calculate the percentage score'
386 def getFunctestConfig(version=""):
388 Get Functest configuration
390 config_file = get_config('functest.test_conf') + version
391 response = requests.get(config_file)
392 return yaml.safe_load(response.text)
395 def getArchitectures(scenario_results):
397 Get software architecture (x86 or Aarch64)
399 supported_arch = ['x86']
400 if len(scenario_results) > 0:
401 for scenario_result in scenario_results.values():
402 for value in scenario_result:
403 if "armband" in value['build_tag']:
404 supported_arch.append('aarch64')
405 return supported_arch
406 return supported_arch
409 def filterArchitecture(results, architecture):
411 Restrict the list of results based on given architecture
413 filtered_results = {}
414 for name, res in results.items():
417 if architecture is "x86":
418 # drop aarch64 results
419 if ("armband" not in value['build_tag']):
420 filtered_values.append(value)
421 elif architecture is "aarch64":
423 if ("armband" in value['build_tag']):
424 filtered_values.append(value)
425 if (len(filtered_values) > 0):
426 filtered_results[name] = filtered_values
427 return filtered_results
433 def subfind(given_list, pattern_list):
435 Yardstick util function
437 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
438 for i in range(len(given_list)):
439 if given_list[i] == pattern_list[0] and \
440 given_list[i:i + LASTEST_TESTS] == pattern_list:
445 def _get_percent(status):
447 Yardstick util function to calculate success rate
450 return round(float(status) * 100 / 6, 1)
452 return status * 100 / 6
455 def get_percent(four_list, ten_list):
457 Yardstick util function to calculate success rate
462 for res_v in four_list:
464 for res_v in ten_list:
467 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
468 if four_score == LASTEST_TESTS:
470 elif subfind(ten_list, [1, 1, 1, 1]):
475 status = four_score + 1
477 return _get_percent(status)
482 Yardstick util function (test)
484 status = getScenarioStatus("compass", "master")
485 print "status:++++++++++++++++++++++++"
486 print json.dumps(status, indent=4)
489 # ----------------------------------------------------------
493 # -----------------------------------------------------------
495 def export_csv(scenario_file_name, installer, version):
497 Generate sub files based on scenario_history.txt
499 scenario_installer_file_name = ("./display/" + version +
500 "/functest/scenario_history_" +
502 scenario_installer_file = open(scenario_installer_file_name, "a")
503 with open(scenario_file_name, "r") as scenario_file:
504 scenario_installer_file.write("date,scenario,installer,detail,score\n")
505 for line in scenario_file:
506 if installer in line:
507 scenario_installer_file.write(line)
508 scenario_installer_file.close
511 def generate_csv(scenario_file):
513 Generate sub files based on scenario_history.txt
516 csv_file = scenario_file.replace('txt', 'csv')
517 shutil.copy2(scenario_file, csv_file)
520 def export_pdf(pdf_path, pdf_doc_name):
522 Export results to pdf
525 pdfkit.from_file(pdf_path, pdf_doc_name)
527 print "Error but pdf generated anyway..."
529 print "impossible to generate PDF"