Merge "utils: arm-pod3 test logs push enable"
[releng.git] / utils / test / reporting / functest / reportingUtils.py
1 #!/usr/bin/python
2 #
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
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 from urllib2 import Request, urlopen, URLError
10 import logging
11 import json
12 import reportingConf as conf
13
14
15 def getLogger(module):
16     logFormatter = logging.Formatter("%(asctime)s [" +
17                                      module +
18                                      "] [%(levelname)-5.5s]  %(message)s")
19     logger = logging.getLogger()
20
21     fileHandler = logging.FileHandler("{0}/{1}".format('.', conf.LOG_FILE))
22     fileHandler.setFormatter(logFormatter)
23     logger.addHandler(fileHandler)
24
25     consoleHandler = logging.StreamHandler()
26     consoleHandler.setFormatter(logFormatter)
27     logger.addHandler(consoleHandler)
28     logger.setLevel(conf.LOG_LEVEL)
29     return logger
30
31
32 def getApiResults(case, installer, scenario, version):
33     results = json.dumps([])
34     # to remove proxy (to be removed at the end for local test only)
35     # proxy_handler = urllib2.ProxyHandler({})
36     # opener = urllib2.build_opener(proxy_handler)
37     # urllib2.install_opener(opener)
38     # url = "http://127.0.0.1:8000/results?case=" + case + \
39     #       "&period=30&installer=" + installer
40     url = ("http://" + conf.URL_BASE + "?case=" + case +
41            "&period=" + str(conf.PERIOD) + "&installer=" + installer +
42            "&scenario=" + scenario + "&version=" + version +
43            "&last=" + str(conf.NB_TESTS))
44     request = Request(url)
45
46     try:
47         response = urlopen(request)
48         k = response.read()
49         results = json.loads(k)
50     except URLError, e:
51         print 'No kittez. Got an error code:', e
52
53     return results
54
55
56 def getScenarios(case, installer, version):
57
58     case = case.getName()
59     url = ("http://" + conf.URL_BASE + "?case=" + case +
60            "&period=" + str(conf.PERIOD) + "&installer=" + installer +
61            "&version=" + version)
62     request = Request(url)
63
64     try:
65         response = urlopen(request)
66         k = response.read()
67         results = json.loads(k)
68         test_results = results['results']
69     except URLError, e:
70         print 'Got an error code:', e
71
72     if test_results is not None:
73         test_results.reverse()
74
75         scenario_results = {}
76
77         for r in test_results:
78             # Retrieve all the scenarios per installer
79             if not r['scenario'] in scenario_results.keys():
80                 scenario_results[r['scenario']] = []
81             scenario_results[r['scenario']].append(r)
82
83     return scenario_results
84
85
86 def getScenarioStats(scenario_results):
87     scenario_stats = {}
88     for k, v in scenario_results.iteritems():
89         scenario_stats[k] = len(v)
90
91     return scenario_stats
92
93
94 def getNbtestOk(results):
95     nb_test_ok = 0
96     for r in results:
97         for k, v in r.iteritems():
98             try:
99                 if "PASS" in v:
100                     nb_test_ok += 1
101             except:
102                 print "Cannot retrieve test status"
103     return nb_test_ok
104
105
106 def getResult(testCase, installer, scenario, version):
107
108     # retrieve raw results
109     results = getApiResults(testCase, installer, scenario, version)
110     # let's concentrate on test results only
111     test_results = results['results']
112
113     # if results found, analyze them
114     if test_results is not None:
115         test_results.reverse()
116
117         scenario_results = []
118
119         # print " ---------------- "
120         # print test_results
121         # print " ---------------- "
122         # print "nb of results:" + str(len(test_results))
123
124         for r in test_results:
125             # print r["start_date"]
126             # print r["criteria"]
127             scenario_results.append({r["start_date"]: r["criteria"]})
128         # sort results
129         scenario_results.sort()
130         # 4 levels for the results
131         # 3: 4+ consecutive runs passing the success criteria
132         # 2: <4 successful consecutive runs but passing the criteria
133         # 1: close to pass the success criteria
134         # 0: 0% success, not passing
135         # -1: no run available
136         test_result_indicator = 0
137         nbTestOk = getNbtestOk(scenario_results)
138
139         # print "Nb test OK (last 10 days):"+ str(nbTestOk)
140         # check that we have at least 4 runs
141         if len(scenario_results) < 1:
142             # No results available      
143             test_result_indicator = -1
144         elif nbTestOk < 1:
145             test_result_indicator = 0
146         elif nbTestOk < 2:
147             test_result_indicator = 1
148         else:
149             # Test the last 4 run
150             if (len(scenario_results) > 3):
151                 last4runResults = scenario_results[-4:]
152                 nbTestOkLast4 = getNbtestOk(last4runResults)
153                 # print "Nb test OK (last 4 run):"+ str(nbTestOkLast4)
154                 if nbTestOkLast4 > 3:
155                     test_result_indicator = 3
156                 else:
157                     test_result_indicator = 2
158             else:
159                 test_result_indicator = 2
160     return test_result_indicator