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
17 # ----------------------------------------------------------
21 # -----------------------------------------------------------
22 def get_parameter_from_yaml(parameter, file):
24 Returns the value of a given parameter in file.yaml
25 parameter must be given in string format with dots
26 Example: general.openstack.image_name
29 file_yaml = yaml.safe_load(f)
32 for element in parameter.split("."):
33 value = value.get(element)
35 raise ValueError("The parameter %s is not defined in"
36 " reporting.yaml" % parameter)
40 def get_config(parameter):
41 yaml_ = os.environ["CONFIG_REPORTING_YAML"]
42 return get_parameter_from_yaml(parameter, yaml_)
45 # ----------------------------------------------------------
49 # -----------------------------------------------------------
50 def getLogger(module):
51 logFormatter = logging.Formatter("%(asctime)s [" +
53 "] [%(levelname)-5.5s] %(message)s")
54 logger = logging.getLogger()
55 log_file = get_config('general.log.log_file')
56 log_level = get_config('general.log.log_level')
58 fileHandler = logging.FileHandler("{0}/{1}".format('.', log_file))
59 fileHandler.setFormatter(logFormatter)
60 logger.addHandler(fileHandler)
62 consoleHandler = logging.StreamHandler()
63 consoleHandler.setFormatter(logFormatter)
64 logger.addHandler(consoleHandler)
65 logger.setLevel(log_level)
69 # ----------------------------------------------------------
73 # -----------------------------------------------------------
74 def getApiResults(case, installer, scenario, version):
75 results = json.dumps([])
76 # to remove proxy (to be removed at the end for local test only)
77 # proxy_handler = urllib2.ProxyHandler({})
78 # opener = urllib2.build_opener(proxy_handler)
79 # urllib2.install_opener(opener)
80 # url = "http://127.0.0.1:8000/results?case=" + case + \
81 # "&period=30&installer=" + installer
82 period = get_config('general.period')
83 url_base = get_config('testapi.url')
84 nb_tests = get_config('general.nb_iteration_tests_success_criteria')
86 url = ("http://" + url_base + "?case=" + case +
87 "&period=" + str(period) + "&installer=" + installer +
88 "&scenario=" + scenario + "&version=" + version +
89 "&last=" + str(nb_tests))
90 request = Request(url)
93 response = urlopen(request)
95 results = json.loads(k)
97 print('No kittez. Got an error code:', e)
102 def getScenarios(case, installer, version):
104 case = case.getName()
105 period = get_config('general.period')
106 url_base = get_config('testapi.url')
108 url = ("http://" + url_base + "?case=" + case +
109 "&period=" + str(period) + "&installer=" + installer +
110 "&version=" + version)
111 request = Request(url)
114 response = urlopen(request)
116 results = json.loads(k)
117 test_results = results['results']
118 except URLError as e:
119 print('Got an error code:', e)
121 if test_results is not None:
122 test_results.reverse()
124 scenario_results = {}
126 for r in test_results:
127 # Retrieve all the scenarios per installer
128 if not r['scenario'] in scenario_results.keys():
129 scenario_results[r['scenario']] = []
130 # Do we consider results from virtual pods ...
131 # Do we consider results for non HA scenarios...
132 exclude_virtual_pod = get_config('functest.exclude_virtual')
133 exclude_noha = get_config('functest.exclude_noha')
134 if ((exclude_virtual_pod and "virtual" in r['pod_name']) or
135 (exclude_noha and "noha" in r['scenario'])):
136 print("exclude virtual pod results...")
138 scenario_results[r['scenario']].append(r)
140 return scenario_results
143 def getScenarioStats(scenario_results):
145 for k, v in scenario_results.iteritems():
146 scenario_stats[k] = len(v)
148 return scenario_stats
151 # TODO convergence with above function getScenarios
152 def getScenarioStatus(installer, version):
153 period = get_config('general.period')
154 url_base = get_config('testapi.url')
156 url = ("http://" + url_base + "?case=scenario_status" +
157 "&installer=" + installer +
158 "&version=" + version + "&period=" + str(period))
159 request = Request(url)
162 response = urlopen(request)
165 results = json.loads(k)
166 test_results = results['results']
167 except URLError as e:
168 print('Got an error code:', e)
170 scenario_results = {}
172 if test_results is not None:
173 for r in test_results:
174 if r['stop_date'] != 'None' and r['criteria'] is not None:
175 if not r['scenario'] in scenario_results.keys():
176 scenario_results[r['scenario']] = []
177 scenario_results[r['scenario']].append(r)
179 for k, v in scenario_results.items():
180 # scenario_results[k] = v[:LASTEST_TESTS]
183 if element['criteria'] == 'SUCCESS':
187 result_dict[k] = s_list
189 # return scenario_results
193 def getNbtestOk(results):
196 for k, v in r.iteritems():
201 print("Cannot retrieve test status")
205 def getResult(testCase, installer, scenario, version):
207 # retrieve raw results
208 results = getApiResults(testCase, installer, scenario, version)
209 # let's concentrate on test results only
210 test_results = results['results']
212 # if results found, analyze them
213 if test_results is not None:
214 test_results.reverse()
216 scenario_results = []
218 # print " ---------------- "
220 # print " ---------------- "
221 # print "nb of results:" + str(len(test_results))
223 for r in test_results:
224 # print r["start_date"]
225 # print r["criteria"]
226 scenario_results.append({r["start_date"]: r["criteria"]})
228 scenario_results.sort()
229 # 4 levels for the results
230 # 3: 4+ consecutive runs passing the success criteria
231 # 2: <4 successful consecutive runs but passing the criteria
232 # 1: close to pass the success criteria
233 # 0: 0% success, not passing
234 # -1: no run available
235 test_result_indicator = 0
236 nbTestOk = getNbtestOk(scenario_results)
238 # print "Nb test OK (last 10 days):"+ str(nbTestOk)
239 # check that we have at least 4 runs
240 if len(scenario_results) < 1:
241 # No results available
242 test_result_indicator = -1
244 test_result_indicator = 0
246 test_result_indicator = 1
248 # Test the last 4 run
249 if (len(scenario_results) > 3):
250 last4runResults = scenario_results[-4:]
251 nbTestOkLast4 = getNbtestOk(last4runResults)
252 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
253 if nbTestOkLast4 > 3:
254 test_result_indicator = 3
256 test_result_indicator = 2
258 test_result_indicator = 2
259 return test_result_indicator
262 def getJenkinsUrl(build_tag):
263 # e.g. jenkins-functest-apex-apex-daily-colorado-daily-colorado-246
265 # jenkins-functest-compass-huawei-pod5-daily-master-136
267 # note it is linked to jenkins format
268 # if this format changes...function to be adapted....
269 url_base = get_config('functest.jenkins_url')
271 build_id = [int(s) for s in build_tag.split("-") if s.isdigit()]
272 url_id = (build_tag[8:-(len(str(build_id[0])) + 1)] +
273 "/" + str(build_id[0]))
274 jenkins_url = url_base + url_id + "/console"
276 print('Impossible to get jenkins url:')
281 def getScenarioPercent(scenario_score, scenario_criteria):
284 score = float(scenario_score) / float(scenario_criteria) * 100
286 print('Impossible to calculate the percentage score')
293 def subfind(given_list, pattern_list):
294 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
295 for i in range(len(given_list)):
296 if given_list[i] == pattern_list[0] and \
297 given_list[i:i + LASTEST_TESTS] == pattern_list:
302 def _get_percent(status):
305 return round(float(status) * 100 / 6, 1)
307 return status * 100 / 6
310 def get_percent(four_list, ten_list):
319 LASTEST_TESTS = get_config('general.nb_iteration_tests_success_criteria')
320 if four_score == LASTEST_TESTS:
322 elif subfind(ten_list, [1, 1, 1, 1]):
327 status = four_score + 1
329 return _get_percent(status)
333 status = getScenarioStatus("compass", "master")
334 print("status:++++++++++++++++++++++++")
335 print(json.dumps(status, indent=4))
338 # ----------------------------------------------------------
342 # -----------------------------------------------------------
344 def export_csv(scenario_file_name, installer, version):
346 # generate sub files based on scenario_history.txt
347 scenario_installer_file_name = ("./display/" + version +
348 "/functest/scenario_history_" +
350 scenario_installer_file = open(scenario_installer_file_name, "a")
351 with open(scenario_file_name, "r") as f:
352 scenario_installer_file.write("date,scenario,installer,detail,score\n")
354 if installer in line:
355 scenario_installer_file.write(line)
356 scenario_installer_file.close
359 def export_pdf(pdf_path, pdf_doc_name):
361 pdfkit.from_file(pdf_path, pdf_doc_name)
363 print("Error but pdf generated anyway...")
365 print("impossible to generate PDF")