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)
122 request = Request(url)
123 response = urlopen(request)
125 results = json.loads(k)
126 test_results = results['results']
128 page = results['pagination']['total_pages']
131 for i in range(1, page + 1):
132 url_page = url + "&page=" + str(i)
133 request = Request(url_page)
134 response = urlopen(request)
136 results = json.loads(k)
137 test_results += results['results']
138 except URLError as err:
139 print('Got an error code:', err)
141 if test_results is not None:
142 test_results.reverse()
143 scenario_results = {}
145 for r in test_results:
146 # Retrieve all the scenarios per installer
147 if not r['scenario'] in scenario_results.keys():
148 scenario_results[r['scenario']] = []
149 # Do we consider results from virtual pods ...
150 # Do we consider results for non HA scenarios...
151 exclude_virtual_pod = get_config('functest.exclude_virtual')
152 exclude_noha = get_config('functest.exclude_noha')
153 if ((exclude_virtual_pod and "virtual" in r['pod_name']) or
154 (exclude_noha and "noha" in r['scenario'])):
155 print("exclude virtual pod results...")
157 scenario_results[r['scenario']].append(r)
159 return scenario_results
162 def getScenarioStats(scenario_results):
164 for k, v in scenario_results.iteritems():
165 scenario_stats[k] = len(v)
167 return scenario_stats
170 def getScenarioStatus(installer, version):
171 period = get_config('general.period')
172 url_base = get_config('testapi.url')
174 url = ("http://" + url_base + "?case=scenario_status" +
175 "&installer=" + installer +
176 "&version=" + version + "&period=" + str(period))
177 request = Request(url)
180 response = urlopen(request)
183 results = json.loads(k)
184 test_results = results['results']
185 except URLError as e:
186 print('Got an error code:', e)
188 scenario_results = {}
190 if test_results is not None:
191 for r in test_results:
192 if r['stop_date'] != 'None' and r['criteria'] is not None:
193 if not r['scenario'] in scenario_results.keys():
194 scenario_results[r['scenario']] = []
195 scenario_results[r['scenario']].append(r)
197 for k, v in scenario_results.items():
198 # scenario_results[k] = v[:LASTEST_TESTS]
201 if element['criteria'] == 'SUCCESS':
205 result_dict[k] = s_list
207 # return scenario_results
211 def getQtipResults(version, installer):
212 period = get_config('qtip.period')
213 url_base = get_config('testapi.url')
215 url = ("http://" + url_base + "?project=qtip" +
216 "&installer=" + installer +
217 "&version=" + version + "&period=" + str(period))
218 request = Request(url)
221 response = urlopen(request)
224 results = json.loads(k)['results']
225 except URLError as err:
226 print('Got an error code:', err)
231 key = '{}/{}'.format(r['pod_name'], r['scenario'])
232 if key not in result_dict.keys():
233 result_dict[key] = []
234 result_dict[key].append(r['details']['score'])
236 # return scenario_results
240 def getNbtestOk(results):
243 for k, v in r.iteritems():
248 print("Cannot retrieve test status")
252 def getResult(testCase, installer, scenario, version):
254 # retrieve raw results
255 results = getApiResults(testCase, installer, scenario, version)
256 # let's concentrate on test results only
257 test_results = results['results']
259 # if results found, analyze them
260 if test_results is not None:
261 test_results.reverse()
263 scenario_results = []
265 # print " ---------------- "
267 # print " ---------------- "
268 # print "nb of results:" + str(len(test_results))
270 for r in test_results:
271 # print r["start_date"]
272 # print r["criteria"]
273 scenario_results.append({r["start_date"]: r["criteria"]})
275 scenario_results.sort()
276 # 4 levels for the results
277 # 3: 4+ consecutive runs passing the success criteria
278 # 2: <4 successful consecutive runs but passing the criteria
279 # 1: close to pass the success criteria
280 # 0: 0% success, not passing
281 # -1: no run available
282 test_result_indicator = 0
283 nbTestOk = getNbtestOk(scenario_results)
285 # print "Nb test OK (last 10 days):"+ str(nbTestOk)
286 # check that we have at least 4 runs
287 if len(scenario_results) < 1:
288 # No results available
289 test_result_indicator = -1
291 test_result_indicator = 0
293 test_result_indicator = 1
295 # Test the last 4 run
296 if (len(scenario_results) > 3):
297 last4runResults = scenario_results[-4:]
298 nbTestOkLast4 = getNbtestOk(last4runResults)
299 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
300 if nbTestOkLast4 > 3:
301 test_result_indicator = 3
303 test_result_indicator = 2
305 test_result_indicator = 2
306 return test_result_indicator
309 def getJenkinsUrl(build_tag):
310 # e.g. jenkins-functest-apex-apex-daily-colorado-daily-colorado-246
312 # jenkins-functest-compass-huawei-pod5-daily-master-136
314 # note it is linked to jenkins format
315 # if this format changes...function to be adapted....
316 url_base = get_config('functest.jenkins_url')
318 build_id = [int(s) for s in build_tag.split("-") if s.isdigit()]
319 url_id = (build_tag[8:-(len(str(build_id[0])) + 1)] +
320 "/" + str(build_id[0]))
321 jenkins_url = url_base + url_id + "/console"
323 print('Impossible to get jenkins url:')
325 if "jenkins-" not in build_tag:
331 def getScenarioPercent(scenario_score, scenario_criteria):
334 score = float(scenario_score) / float(scenario_criteria) * 100
336 print('Impossible to calculate the percentage score')
343 def getFunctestConfig(version=""):
344 config_file = get_config('functest.test_conf') + version
345 response = requests.get(config_file)
346 return yaml.safe_load(response.text)
349 def getArchitectures(scenario_results):
350 supported_arch = ['x86']
351 if (len(scenario_results) > 0):
352 for scenario_result in scenario_results.values():
353 for value in scenario_result:
354 if ("armband" in value['build_tag']):
355 supported_arch.append('aarch64')
356 return supported_arch
357 return supported_arch
360 def filterArchitecture(results, architecture):
361 filtered_results = {}
362 for name, results in results.items():
364 for value in results:
365 if (architecture is "x86"):
366 # drop aarch64 results
367 if ("armband" not in value['build_tag']):
368 filtered_values.append(value)
369 elif(architecture is "aarch64"):
371 if ("armband" in value['build_tag']):
372 filtered_values.append(value)
373 if (len(filtered_values) > 0):
374 filtered_results[name] = filtered_values
375 return filtered_results
381 def subfind(given_list, pattern_list):
382 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
383 for i in range(len(given_list)):
384 if given_list[i] == pattern_list[0] and \
385 given_list[i:i + LASTEST_TESTS] == pattern_list:
390 def _get_percent(status):
393 return round(float(status) * 100 / 6, 1)
395 return status * 100 / 6
398 def get_percent(four_list, ten_list):
407 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
408 if four_score == LASTEST_TESTS:
410 elif subfind(ten_list, [1, 1, 1, 1]):
415 status = four_score + 1
417 return _get_percent(status)
421 status = getScenarioStatus("compass", "master")
422 print("status:++++++++++++++++++++++++")
423 print(json.dumps(status, indent=4))
426 # ----------------------------------------------------------
430 # -----------------------------------------------------------
432 def export_csv(scenario_file_name, installer, version):
434 # generate sub files based on scenario_history.txt
435 scenario_installer_file_name = ("./display/" + version +
436 "/functest/scenario_history_" +
438 scenario_installer_file = open(scenario_installer_file_name, "a")
439 with open(scenario_file_name, "r") as scenario_file:
440 scenario_installer_file.write("date,scenario,installer,detail,score\n")
441 for line in scenario_file:
442 if installer in line:
443 scenario_installer_file.write(line)
444 scenario_installer_file.close
447 def generate_csv(scenario_file):
450 # generate sub files based on scenario_history.txt
451 csv_file = scenario_file.replace('txt', 'csv')
452 shutil.copy2(scenario_file, csv_file)
455 def export_pdf(pdf_path, pdf_doc_name):
457 pdfkit.from_file(pdf_path, pdf_doc_name)
459 print("Error but pdf generated anyway...")
461 print("impossible to generate PDF")