1 ##############################################################################
2 # Copyright (c) 2016 Huawei Technologies Co.,Ltd and others.
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
11 from tornado.web import RequestHandler
12 from tornado.escape import json_encode
13 from tornado.escape import json_decode
16 class BaseHandler(RequestHandler):
17 def _set_header(self):
18 self.set_header('Access-Control-Allow-Origin', '*')
19 self.set_header('Access-Control-Allow-Headers',
20 'Content-Type, Content-Length, Authorization, \
21 Accept, X-Requested-With , PRIVATE-TOKEN')
22 self.set_header('Access-Control-Allow-Methods',
23 'PUT, POST, GET, DELETE, OPTIONS')
26 class FiltersHandler(BaseHandler):
32 'status': ['success', 'warning', 'danger'],
33 'projects': ['functest', 'yardstick'],
34 'installers': ['apex', 'compass', 'fuel', 'joid'],
35 'version': ['colorado', 'master'],
36 'loops': ['daily', 'weekly', 'monthly'],
37 'time': ['10 days', '30 days']
40 return self.write(json_encode(filters))
43 class ScenariosHandler(BaseHandler):
47 body = json_decode(self.request.body)
48 args = self._get_args(body)
50 scenarios = self._get_result_data(self._get_scenarios(), args)
52 return self.write(json_encode(dict(scenarios=scenarios)))
54 def _get_result_data(self, data, args):
55 data = self._filter_status(data, args)
56 return {s: self._get_scenario_result(s, data[s], args) for s in data}
58 def _filter_status(self, data, args):
59 return {k: v for k, v in data.items() if v['status'] in args['status']}
61 def _get_scenario_result(self, scenario, data, args):
63 'status': data.get('status'),
64 'installers': self._get_installers_result(data['installers'], args)
68 def _get_installers_result(self, data, args):
69 func = self._get_installer_result
70 return {k: func(k, data.get(k, {}), args) for k in args['installers']}
72 def _get_installer_result(self, installer, data, args):
73 projects = data.get(args['version'], [])
74 return [self._get_project_data(projects, p) for p in args['projects']]
76 def _get_project_data(self, projects, project):
83 if p['project'] == project:
87 def _get_scenarios(self):
88 url = 'http://testresults.opnfv.org/test/api/v1/scenarios'
89 resp = requests.get(url).json()
90 data = self._change_to_utf8(resp).get('scenarios', {})
91 return {a.get('name'): self._get_scenario(a.get('installers', [])
94 def _get_scenario(self, data):
95 installers = {a.get('installer'): self._get_installer(a.get('versions',
99 'status': self._get_status(),
100 'installers': installers
104 def _get_status(self):
107 def _get_installer(self, data):
108 return {a.get('version'): self._get_version(a) for a in data}
110 def _get_version(self, data):
112 scores = data.get('score', {}).get('projects')[0]
113 trusts = data.get('trust_indicator', {}).get('projects')[0]
114 except (TypeError, IndexError):
117 scores = {key: [dict(date=a.get('date')[:10],
119 ) for a in scores[key]] for key in scores}
120 trusts = {key: [dict(date=a.get('date')[:10],
121 status=a.get('status')
122 ) for a in trusts[key]] for key in trusts}
123 atom = self._get_atom(scores, trusts)
124 return [dict(project=k,
125 score=sorted(atom[k], reverse=True)[0].get('score'),
126 status=sorted(atom[k], reverse=True)[0].get('status')
127 ) for k in atom if atom[k]]
129 def _get_atom(self, scores, trusts):
130 s = {k: {a['date']: a['score'] for a in scores[k]} for k in scores}
131 t = {k: {a['date']: a['status'] for a in trusts[k]} for k in trusts}
132 return {k: [dict(score=s[k][a], status=t[k][a], data=a
133 ) for a in s[k] if a in t[k]] for k in s}
135 def _change_to_utf8(self, obj):
136 if isinstance(obj, dict):
137 return {str(k): self._change_to_utf8(v) for k, v in obj.items()}
138 elif isinstance(obj, list):
139 return [self._change_to_utf8(ele) for ele in obj]
143 if isinstance(new, int):
145 return self._change_to_utf8(new)
146 except (NameError, TypeError, SyntaxError):
149 def _get_args(self, body):
150 status = self._get_status_args(body)
151 projects = self._get_projects_args(body)
152 installers = self._get_installers_args(body)
156 'projects': projects,
157 'installers': installers,
158 'version': body.get('version', 'master').lower(),
159 'loops': body.get('loops', 'daily').lower(),
160 'time': body.get('times', '10 days')[:2].lower()
164 def _get_status_args(self, body):
165 status_all = ['success', 'warning', 'danger']
166 status = [a.lower() for a in body.get('status', ['all'])]
167 return status_all if 'all' in status else status
169 def _get_projects_args(self, body):
170 project_all = ['functest', 'yardstick']
171 projects = [a.lower() for a in body.get('projects', ['all'])]
172 return project_all if 'all' in projects else projects
174 def _get_installers_args(self, body):
175 installer_all = ['apex', 'compass', 'fuel', 'joid']
176 installers = [a.lower() for a in body.get('installers', ['all'])]
177 return installer_all if 'all' in installers else installers