diff options
Diffstat (limited to 'utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py')
| -rw-r--r-- | utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py | 219 |
1 files changed, 85 insertions, 134 deletions
diff --git a/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py b/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py index 5d420a56e..a89e7ee13 100644 --- a/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py +++ b/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py @@ -1,7 +1,5 @@ import functools -from opnfv_testapi.common import message -from opnfv_testapi.common import raises from opnfv_testapi.resources import handlers import opnfv_testapi.resources.scenario_models as models from opnfv_testapi.tornado_swagger import swagger @@ -15,6 +13,24 @@ class GenericScenarioHandler(handlers.GenericApiHandler): self.table = self.db_scenarios self.table_cls = models.Scenario + def set_query(self, filters): + query = dict() + elem_query = dict() + for k, v in filters.iteritems(): + if k == 'scenario': + query['name'] = v + elif k == 'installer': + elem_query["installer"] = v + elif k == 'version': + elem_query["versions.version"] = v + elif k == 'project': + elem_query["versions.projects.project"] = v + else: + query[k] = v + if elem_query: + query['installers'] = {'$elemMatch': elem_query} + return query + class ScenariosCLHandler(GenericScenarioHandler): @swagger.operation(nickname="queryScenarios") @@ -96,21 +112,8 @@ class ScenarioGURHandler(GenericScenarioHandler): self._get_one(query={'name': name}) pass - @swagger.operation(nickname="updateScenarioByName") def put(self, name): - """ - @description: update a single scenario by name - @param body: fields to be updated - @type body: L{ScenarioUpdateRequest} - @in body: body - @rtype: L{Scenario} - @return 200: update success - @raise 404: scenario not exist - @raise 403: nothing to update - """ - query = {'name': name} - db_keys = ['name'] - self._update(query=query, db_keys=db_keys) + pass @swagger.operation(nickname="deleteScenarioByName") def delete(self, name): @@ -119,164 +122,112 @@ class ScenarioGURHandler(GenericScenarioHandler): @return 200: delete success @raise 404: scenario not exist: """ - self._delete(query={'name': name}) - def _update_query(self, keys, data): - query = dict() - if self._is_rename(): - new = self._term.get('name') - if data.get('name') != new: - query['name'] = new - return query +class ScenarioUpdater(object): + def __init__(self, data, body=None, + installer=None, version=None, project=None): + self.data = data + self.body = body + self.installer = installer + self.version = version + self.project = project - def _update_requests(self, data): + def update(self, item, op): updates = { - ('name', 'update'): self._update_requests_rename, - ('installer', 'add'): self._update_requests_add_installer, - ('installer', 'delete'): self._update_requests_delete_installer, - ('version', 'add'): self._update_requests_add_version, - ('version', 'delete'): self._update_requests_delete_version, - ('owner', 'update'): self._update_requests_change_owner, - ('project', 'add'): self._update_requests_add_project, - ('project', 'delete'): self._update_requests_delete_project, - ('customs', 'add'): self._update_requests_add_customs, - ('customs', 'delete'): self._update_requests_delete_customs, ('score', 'add'): self._update_requests_add_score, - ('trust_indicator', 'add'): self._update_requests_add_ti, } + updates[(item, op)](self.data) - updates[(self._field, self._op)](data) - - return data.format() + return self.data.format() - def _iter_installers(xstep): + def iter_installers(xstep): @functools.wraps(xstep) def magic(self, data): [xstep(self, installer) for installer in self._filter_installers(data.installers)] return magic - def _iter_versions(xstep): + def iter_versions(xstep): @functools.wraps(xstep) def magic(self, installer): [xstep(self, version) for version in (self._filter_versions(installer.versions))] return magic - def _iter_projects(xstep): + def iter_projects(xstep): @functools.wraps(xstep) def magic(self, version): [xstep(self, project) for project in (self._filter_projects(version.projects))] return magic - def _update_requests_rename(self, data): - data.name = self._term.get('name') - if not data.name: - raises.BadRequest(message.missing('name')) - - def _update_requests_add_installer(self, data): - data.installers.append(models.ScenarioInstaller.from_dict(self._term)) - - def _update_requests_delete_installer(self, data): - data.installers = self._remove_installers(data.installers) - - @_iter_installers - def _update_requests_add_version(self, installer): - installer.versions.append(models.ScenarioVersion.from_dict(self._term)) - - @_iter_installers - def _update_requests_delete_version(self, installer): - installer.versions = self._remove_versions(installer.versions) - - @_iter_installers - @_iter_versions - def _update_requests_change_owner(self, version): - version.owner = self._term.get('owner') - - @_iter_installers - @_iter_versions - def _update_requests_add_project(self, version): - version.projects.append(models.ScenarioProject.from_dict(self._term)) - - @_iter_installers - @_iter_versions - def _update_requests_delete_project(self, version): - version.projects = self._remove_projects(version.projects) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_add_customs(self, project): - project.customs = list(set(project.customs + self._term)) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_delete_customs(self, project): - project.customs = filter( - lambda f: f not in self._term, - project.customs) - - @_iter_installers - @_iter_versions - @_iter_projects + @iter_installers + @iter_versions + @iter_projects def _update_requests_add_score(self, project): project.scores.append( - models.ScenarioScore.from_dict(self._term)) - - @_iter_installers - @_iter_versions - @_iter_projects - def _update_requests_add_ti(self, project): - project.trust_indicators.append( - models.ScenarioTI.from_dict(self._term)) - - def _is_rename(self): - return self._field == 'name' and self._op == 'update' - - def _remove_installers(self, installers): - return self._remove('installer', installers) + models.ScenarioScore.from_dict(self.body)) def _filter_installers(self, installers): return self._filter('installer', installers) - def _remove_versions(self, versions): - return self._remove('version', versions) - def _filter_versions(self, versions): return self._filter('version', versions) - def _remove_projects(self, projects): - return self._remove('project', projects) - def _filter_projects(self, projects): return self._filter('project', projects) - def _remove(self, field, fields): + def _filter(self, item, items): return filter( - lambda f: getattr(f, field) != self._locate.get(field), - fields) - - def _filter(self, field, fields): - return filter( - lambda f: getattr(f, field) == self._locate.get(field), - fields) - - @property - def _field(self): - return self.json_args.get('field') + lambda f: getattr(f, item) == getattr(self, item), + items) - @property - def _op(self): - return self.json_args.get('op') - @property - def _locate(self): - return self.json_args.get('locate') +class ScenarioScoresHandler(GenericScenarioHandler): + @swagger.operation(nickname="addScoreRecord") + def post(self, scenario): + """ + @description: add a new score record + @notes: add a new score record to a project + POST /api/v1/scenarios/<scenario_name>/scores? \ + installer=<installer_name>& \ + version=<version_name>& \ + project=<project_name> + @param body: score to be added + @type body: L{ScenarioScore} + @in body: body + @param installer: installer type + @type installer: L{string} + @in installer: query + @required installer: True + @param version: version + @type version: L{string} + @in version: query + @required version: True + @param project: project name + @type project: L{string} + @in project: query + @required project: True + @rtype: L{Scenario} + @return 200: score is created. + @raise 404: scenario/installer/version/project not existed + """ + self.installer = self.get_query_argument('installer') + self.version = self.get_query_argument('version') + self.project = self.get_query_argument('project') + + filters = {'scenario': scenario, + 'installer': self.installer, + 'version': self.version, + 'project': self.project} + db_keys = ['name'] + self._update(query=self.set_query(filters=filters), db_keys=db_keys) - @property - def _term(self): - return self.json_args.get('term') + def _update_requests(self, data): + return ScenarioUpdater(data, + self.json_args, + self.installer, + self.version, + self.project).update('score', 'add') |