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:'.format(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']
139 print ('No pagination detected')
140 except URLError as err:
141 print 'Got an error code: {}'.format(err)
143 if test_results is not None:
144 test_results.reverse()
145 scenario_results = {}
147 for r in test_results:
148 # Retrieve all the scenarios per installer
149 if not r['scenario'] in scenario_results.keys():
150 scenario_results[r['scenario']] = []
151 # Do we consider results from virtual pods ...
152 # Do we consider results for non HA scenarios...
153 exclude_virtual_pod = get_config('functest.exclude_virtual')
154 exclude_noha = get_config('functest.exclude_noha')
155 if ((exclude_virtual_pod and "virtual" in r['pod_name']) or
156 (exclude_noha and "noha" in r['scenario'])):
157 print "exclude virtual pod results..."
159 scenario_results[r['scenario']].append(r)
161 return scenario_results
164 def getScenarioStats(scenario_results):
166 for k, v in scenario_results.iteritems():
167 scenario_stats[k] = len(v)
169 return scenario_stats
172 def getScenarioStatus(installer, version):
173 period = get_config('general.period')
174 url_base = get_config('testapi.url')
176 url = ("http://" + url_base + "?case=scenario_status" +
177 "&installer=" + installer +
178 "&version=" + version + "&period=" + str(period))
179 request = Request(url)
182 response = urlopen(request)
185 results = json.loads(k)
186 test_results = results['results']
187 except URLError as e:
188 print 'Got an error code: {}'.format(e)
190 scenario_results = {}
192 if test_results is not None:
193 for r in test_results:
194 if r['stop_date'] != 'None' and r['criteria'] is not None:
195 if not r['scenario'] in scenario_results.keys():
196 scenario_results[r['scenario']] = []
197 scenario_results[r['scenario']].append(r)
199 for k, v in scenario_results.items():
200 # scenario_results[k] = v[:LASTEST_TESTS]
203 if element['criteria'] == 'SUCCESS':
207 result_dict[k] = s_list
209 # return scenario_results
213 def getQtipResults(version, installer):
214 period = get_config('qtip.period')
215 url_base = get_config('testapi.url')
217 url = ("http://" + url_base + "?project=qtip" +
218 "&installer=" + installer +
219 "&version=" + version + "&period=" + str(period))
220 request = Request(url)
223 response = urlopen(request)
226 results = json.loads(k)['results']
227 except URLError as err:
228 print 'Got an error code: {}'.format(err)
233 key = '{}/{}'.format(r['pod_name'], r['scenario'])
234 if key not in result_dict.keys():
235 result_dict[key] = []
236 result_dict[key].append(r['details']['score'])
238 # return scenario_results
242 def getNbtestOk(results):
245 for k, v in r.iteritems():
250 print "Cannot retrieve test status"
254 def getResult(testCase, installer, scenario, version):
256 # retrieve raw results
257 results = getApiResults(testCase, installer, scenario, version)
258 # let's concentrate on test results only
259 test_results = results['results']
261 # if results found, analyze them
262 if test_results is not None:
263 test_results.reverse()
265 scenario_results = []
267 # print " ---------------- "
269 # print " ---------------- "
270 # print "nb of results:" + str(len(test_results))
272 for r in test_results:
273 # print r["start_date"]
274 # print r["criteria"]
275 scenario_results.append({r["start_date"]: r["criteria"]})
277 scenario_results.sort()
278 # 4 levels for the results
279 # 3: 4+ consecutive runs passing the success criteria
280 # 2: <4 successful consecutive runs but passing the criteria
281 # 1: close to pass the success criteria
282 # 0: 0% success, not passing
283 # -1: no run available
284 test_result_indicator = 0
285 nbTestOk = getNbtestOk(scenario_results)
287 # print "Nb test OK (last 10 days):"+ str(nbTestOk)
288 # check that we have at least 4 runs
289 if len(scenario_results) < 1:
290 # No results available
291 test_result_indicator = -1
293 test_result_indicator = 0
295 test_result_indicator = 1
297 # Test the last 4 run
298 if (len(scenario_results) > 3):
299 last4runResults = scenario_results[-4:]
300 nbTestOkLast4 = getNbtestOk(last4runResults)
301 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
302 if nbTestOkLast4 > 3:
303 test_result_indicator = 3
305 test_result_indicator = 2
307 test_result_indicator = 2
308 return test_result_indicator
311 def getJenkinsUrl(build_tag):
312 # e.g. jenkins-functest-apex-apex-daily-colorado-daily-colorado-246
314 # jenkins-functest-compass-huawei-pod5-daily-master-136
316 # note it is linked to jenkins format
317 # if this format changes...function to be adapted....
318 url_base = get_config('functest.jenkins_url')
320 build_id = [int(s) for s in build_tag.split("-") if s.isdigit()]
321 url_id = (build_tag[8:-(len(str(build_id[0])) + 1)] +
322 "/" + str(build_id[0]))
323 jenkins_url = url_base + url_id + "/console"
325 print 'Impossible to get jenkins url:'
327 if "jenkins-" not in build_tag:
333 def getScenarioPercent(scenario_score, scenario_criteria):
336 score = float(scenario_score) / float(scenario_criteria) * 100
338 print 'Impossible to calculate the percentage score'
345 def getFunctestConfig(version=""):
346 config_file = get_config('functest.test_conf') + version
347 response = requests.get(config_file)
348 return yaml.safe_load(response.text)
351 def getArchitectures(scenario_results):
352 supported_arch = ['x86']
353 if (len(scenario_results) > 0):
354 for scenario_result in scenario_results.values():
355 for value in scenario_result:
356 if ("armband" in value['build_tag']):
357 supported_arch.append('aarch64')
358 return supported_arch
359 return supported_arch
362 def filterArchitecture(results, architecture):
363 filtered_results = {}
364 for name, results in results.items():
366 for value in results:
367 if (architecture is "x86"):
368 # drop aarch64 results
369 if ("armband" not in value['build_tag']):
370 filtered_values.append(value)
371 elif(architecture is "aarch64"):
373 if ("armband" in value['build_tag']):
374 filtered_values.append(value)
375 if (len(filtered_values) > 0):
376 filtered_results[name] = filtered_values
377 return filtered_results
383 def subfind(given_list, pattern_list):
384 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
385 for i in range(len(given_list)):
386 if given_list[i] == pattern_list[0] and \
387 given_list[i:i + LASTEST_TESTS] == pattern_list:
392 def _get_percent(status):
395 return round(float(status) * 100 / 6, 1)
397 return status * 100 / 6
400 def get_percent(four_list, ten_list):
409 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
410 if four_score == LASTEST_TESTS:
412 elif subfind(ten_list, [1, 1, 1, 1]):
417 status = four_score + 1
419 return _get_percent(status)
423 status = getScenarioStatus("compass", "master")
424 print "status:++++++++++++++++++++++++"
425 print(json.dumps(status, indent=4))
428 # ----------------------------------------------------------
432 # -----------------------------------------------------------
434 def export_csv(scenario_file_name, installer, version):
436 # generate sub files based on scenario_history.txt
437 scenario_installer_file_name = ("./display/" + version +
438 "/functest/scenario_history_" +
440 scenario_installer_file = open(scenario_installer_file_name, "a")
441 with open(scenario_file_name, "r") as scenario_file:
442 scenario_installer_file.write("date,scenario,installer,detail,score\n")
443 for line in scenario_file:
444 if installer in line:
445 scenario_installer_file.write(line)
446 scenario_installer_file.close
449 def generate_csv(scenario_file):
452 # generate sub files based on scenario_history.txt
453 csv_file = scenario_file.replace('txt', 'csv')
454 shutil.copy2(scenario_file, csv_file)
457 def export_pdf(pdf_path, pdf_doc_name):
459 pdfkit.from_file(pdf_path, pdf_doc_name)
461 print "Error but pdf generated anyway..."
463 print "impossible to generate PDF"