3 from opnfv_testapi.common import constants
4 from opnfv_testapi.resources import handlers
5 import opnfv_testapi.resources.scenario_models as models
6 from opnfv_testapi.tornado_swagger import swagger
9 class GenericScenarioHandler(handlers.GenericApiHandler):
10 def __init__(self, application, request, **kwargs):
11 super(GenericScenarioHandler, self).__init__(application,
14 self.table = self.db_scenarios
15 self.table_cls = models.Scenario
18 class ScenariosCLHandler(GenericScenarioHandler):
19 @swagger.operation(nickname="queryScenarios")
22 @description: Retrieve scenario(s).
23 @notes: Retrieve scenario(s)
24 Available filters for this request are :
25 - name : scenario name
27 GET /scenarios?name=scenario_1
28 @param name: scenario name
32 @param installer: installer type
33 @type installer: L{string}
35 @required installer: False
36 @param version: version
37 @type version: L{string}
39 @required version: False
40 @param project: project name
41 @type project: L{string}
43 @required project: False
44 @return 200: all scenarios satisfy queries,
45 empty list if no scenario is found
52 for k in self.request.query_arguments.keys():
53 v = self.get_query_argument(k)
55 elem_query["installer"] = v
57 elem_query["versions.version"] = v
59 elem_query["versions.projects.project"] = v
63 query['installers'] = {'$elemMatch': elem_query}
66 self._list(_set_query())
68 @swagger.operation(nickname="createScenario")
71 @description: create a new scenario by name
72 @param body: scenario to be created
73 @type body: L{ScenarioCreateRequest}
75 @rtype: L{CreateResponse}
76 @return 200: scenario is created.
77 @raise 403: scenario already exists
78 @raise 400: body or name not provided
81 return {'name': data.name}
84 message = '{} already exists as a scenario'.format(data.name)
85 return constants.HTTP_FORBIDDEN, message
87 miss_checks = ['name']
88 db_checks = [(self.table, False, query, error)]
89 self._create(miss_checks=miss_checks, db_checks=db_checks)
92 class ScenarioGURHandler(GenericScenarioHandler):
93 @swagger.operation(nickname='getScenarioByName')
96 @description: get a single scenario by name
98 @return 200: scenario exist
99 @raise 404: scenario not exist
101 self._get_one({'name': name})
104 @swagger.operation(nickname="updateScenarioByName")
107 @description: update a single scenario by name
108 @param body: fields to be updated
109 @type body: L{ScenarioUpdateRequest}
112 @return 200: update success
113 @raise 404: scenario not exist
114 @raise 403: nothing to update
116 query = {'name': name}
118 self._update(query, db_keys)
120 @swagger.operation(nickname="deleteScenarioByName")
121 def delete(self, name):
123 @description: delete a scenario by name
124 @return 200: delete success
125 @raise 404: scenario not exist:
128 query = {'name': name}
131 def _update_query(self, keys, data):
134 if self._is_rename():
135 new = self._term.get('name')
142 def _update_requests(self, data):
144 ('name', 'update'): self._update_requests_rename,
145 ('installer', 'add'): self._update_requests_add_installer,
146 ('installer', 'delete'): self._update_requests_delete_installer,
147 ('version', 'add'): self._update_requests_add_version,
148 ('version', 'delete'): self._update_requests_delete_version,
149 ('owner', 'update'): self._update_requests_change_owner,
150 ('project', 'add'): self._update_requests_add_project,
151 ('project', 'delete'): self._update_requests_delete_project,
152 ('customs', 'add'): self._update_requests_add_customs,
153 ('customs', 'delete'): self._update_requests_delete_customs,
154 ('score', 'add'): self._update_requests_add_score,
155 ('trust_indicator', 'add'): self._update_requests_add_ti,
158 updates[(self._field, self._op)](data)
162 def _iter_installers(xstep):
163 @functools.wraps(xstep)
164 def magic(self, data):
165 [xstep(self, installer)
166 for installer in self._filter_installers(data.installers)]
169 def _iter_versions(xstep):
170 @functools.wraps(xstep)
171 def magic(self, installer):
172 [xstep(self, version)
173 for version in (self._filter_versions(installer.versions))]
176 def _iter_projects(xstep):
177 @functools.wraps(xstep)
178 def magic(self, version):
179 [xstep(self, project)
180 for project in (self._filter_projects(version.projects))]
183 def _update_requests_rename(self, data):
184 data.name = self._term.get('name')
186 def _update_requests_add_installer(self, data):
187 data.installers.append(models.ScenarioInstaller.from_dict(self._term))
189 def _update_requests_delete_installer(self, data):
190 data.installers = self._remove_installers(data.installers)
193 def _update_requests_add_version(self, installer):
194 installer.versions.append(models.ScenarioVersion.from_dict(self._term))
197 def _update_requests_delete_version(self, installer):
198 installer.versions = self._remove_versions(installer.versions)
202 def _update_requests_change_owner(self, version):
203 version.owner = self._term.get('owner')
207 def _update_requests_add_project(self, version):
208 version.projects.append(models.ScenarioProject.from_dict(self._term))
212 def _update_requests_delete_project(self, version):
213 version.projects = self._remove_projects(version.projects)
218 def _update_requests_add_customs(self, project):
219 project.customs = list(set(project.customs + self._term))
224 def _update_requests_delete_customs(self, project):
225 project.customs = filter(
226 lambda f: f not in self._term,
232 def _update_requests_add_score(self, project):
233 project.scores.append(
234 models.ScenarioScore.from_dict(self._term))
239 def _update_requests_add_ti(self, project):
240 project.trust_indicators.append(
241 models.ScenarioTI.from_dict(self._term))
243 def _is_rename(self):
244 return self._field == 'name' and self._op == 'update'
246 def _remove_installers(self, installers):
247 return self._remove('installer', installers)
249 def _filter_installers(self, installers):
250 return self._filter('installer', installers)
252 def _remove_versions(self, versions):
253 return self._remove('version', versions)
255 def _filter_versions(self, versions):
256 return self._filter('version', versions)
258 def _remove_projects(self, projects):
259 return self._remove('project', projects)
261 def _filter_projects(self, projects):
262 return self._filter('project', projects)
264 def _remove(self, field, fields):
266 lambda f: getattr(f, field) != self._locate.get(field),
269 def _filter(self, field, fields):
271 lambda f: getattr(f, field) == self._locate.get(field),
276 return self.json_args.get('field')
280 return self.json_args.get('op')
284 return self.json_args.get('locate')
288 return self.json_args.get('term')