1 from opnfv_testapi.common import constants
2 from opnfv_testapi.resources import handlers
3 import opnfv_testapi.resources.scenario_models as models
4 from opnfv_testapi.tornado_swagger import swagger
7 class GenericScenarioHandler(handlers.GenericApiHandler):
8 def __init__(self, application, request, **kwargs):
9 super(GenericScenarioHandler, self).__init__(application,
12 self.table = self.db_scenarios
13 self.table_cls = models.Scenario
16 class ScenariosCLHandler(GenericScenarioHandler):
17 @swagger.operation(nickname="queryScenarios")
20 @description: Retrieve scenario(s).
21 @notes: Retrieve scenario(s)
22 Available filters for this request are :
23 - name : scenario name
25 GET /scenarios?name=scenario_1
26 @param name: scenario name
30 @param installer: installer type
31 @type installer: L{string}
33 @required installer: False
34 @param version: version
35 @type version: L{string}
37 @required version: False
38 @param project: project name
39 @type project: L{string}
41 @required project: False
42 @return 200: all scenarios satisfy queries,
43 empty list if no scenario is found
50 for k in self.request.query_arguments.keys():
51 v = self.get_query_argument(k)
53 elem_query["installer"] = v
55 elem_query["versions.version"] = v
57 elem_query["versions.projects.project"] = v
61 query['installers'] = {'$elemMatch': elem_query}
64 self._list(_set_query())
66 @swagger.operation(nickname="createScenario")
69 @description: create a new scenario by name
70 @param body: scenario to be created
71 @type body: L{ScenarioCreateRequest}
73 @rtype: L{CreateResponse}
74 @return 200: scenario is created.
75 @raise 403: scenario already exists
76 @raise 400: body or name not provided
79 return {'name': data.name}
82 message = '{} already exists as a scenario'.format(data.name)
83 return constants.HTTP_FORBIDDEN, message
85 miss_checks = ['name']
86 db_checks = [(self.table, False, query, error)]
87 self._create(miss_checks=miss_checks, db_checks=db_checks)
90 class ScenarioGURHandler(GenericScenarioHandler):
91 @swagger.operation(nickname='getScenarioByName')
94 @description: get a single scenario by name
96 @return 200: scenario exist
97 @raise 404: scenario not exist
99 self._get_one({'name': name})
102 @swagger.operation(nickname="updateScenarioByName")
105 @description: update a single scenario by name
106 @param body: fields to be updated
107 @type body: L{ScenarioUpdateRequest}
110 @return 200: update success
111 @raise 404: scenario not exist
112 @raise 403: nothing to update
114 query = {'name': name}
116 self._update(query, db_keys)
118 @swagger.operation(nickname="deleteScenarioByName")
119 def delete(self, name):
121 @description: delete a scenario by name
122 @return 200: delete success
123 @raise 404: scenario not exist:
126 query = {'name': name}
129 def _update_query(self, keys, data):
132 if self._is_rename():
133 new = self._term.get('name')
140 def _update_requests(self, data):
142 ('name', 'update'): self._update_requests_rename,
143 ('installer', 'add'): self._update_requests_add_installer,
144 ('installer', 'delete'): self._update_requests_delete_installer,
145 ('version', 'add'): self._update_requests_add_version,
146 ('version', 'delete'): self._update_requests_delete_version,
147 ('owner', 'update'): self._update_requests_change_owner,
148 ('project', 'add'): self._update_requests_add_project,
149 ('project', 'delete'): self._update_requests_delete_project,
150 ('customs', 'add'): self._update_requests_add_customs,
151 ('customs', 'delete'): self._update_requests_delete_customs,
152 ('score', 'add'): self._update_requests_add_score,
153 ('trust_indicator', 'add'): self._update_requests_add_ti,
156 updates[(self._field, self._op)](data)
160 def _iter_installers(xstep):
161 def magic(self, data):
162 [xstep(self, installer)
163 for installer in self._filter_installers(data.installers)]
166 def _iter_versions(xstep):
167 def magic(self, installer):
168 [xstep(self, version)
169 for version in (self._filter_versions(installer.versions))]
172 def _iter_projects(xstep):
173 def magic(self, version):
174 [xstep(self, project)
175 for project in (self._filter_projects(version.projects))]
178 def _update_requests_rename(self, data):
179 data.name = self._term.get('name')
181 def _update_requests_add_installer(self, data):
182 data.installers.append(models.ScenarioInstaller.from_dict(self._term))
184 def _update_requests_delete_installer(self, data):
185 data.installers = self._remove_installers(data.installers)
188 def _update_requests_add_version(self, installer):
189 installer.versions.append(models.ScenarioVersion.from_dict(self._term))
192 def _update_requests_delete_version(self, installer):
193 installer.versions = self._remove_versions(installer.versions)
197 def _update_requests_change_owner(self, version):
198 version.owner = self._term.get('owner')
202 def _update_requests_add_project(self, version):
203 version.projects.append(models.ScenarioProject.from_dict(self._term))
207 def _update_requests_delete_project(self, version):
208 version.projects = self._remove_projects(version.projects)
213 def _update_requests_add_customs(self, project):
214 project.customs = list(set(project.customs + self._term))
219 def _update_requests_delete_customs(self, project):
220 project.customs = filter(
221 lambda f: f not in self._term,
227 def _update_requests_add_score(self, project):
228 project.scores.append(
229 models.ScenarioScore.from_dict(self._term))
234 def _update_requests_add_ti(self, project):
235 project.trust_indicators.append(
236 models.ScenarioTI.from_dict(self._term))
238 def _is_rename(self):
239 return self._field == 'name' and self._op == 'update'
241 def _remove_installers(self, installers):
242 return self._remove('installer', installers)
244 def _filter_installers(self, installers):
245 return self._filter('installer', installers)
247 def _remove_versions(self, versions):
248 return self._remove('version', versions)
250 def _filter_versions(self, versions):
251 return self._filter('version', versions)
253 def _remove_projects(self, projects):
254 return self._remove('project', projects)
256 def _filter_projects(self, projects):
257 return self._filter('project', projects)
259 def _remove(self, field, fields):
261 lambda f: getattr(f, field) != self._locate.get(field),
264 def _filter(self, field, fields):
266 lambda f: getattr(f, field) == self._locate.get(field),
271 return self.json_args.get('field')
275 return self.json_args.get('op')
279 return self.json_args.get('locate')
283 return self.json_args.get('term')