import functools from opnfv_testapi.resources import handlers import opnfv_testapi.resources.scenario_models as models from opnfv_testapi.tornado_swagger import swagger class GenericScenarioHandler(handlers.GenericApiHandler): def __init__(self, application, request, **kwargs): super(GenericScenarioHandler, self).__init__(application, request, **kwargs) self.table = self.db_scenarios self.table_cls = models.Scenario def set_query(self, locators): query = dict() elem_query = dict() for k, v in locators.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") def get(self): """ @description: Retrieve scenario(s). @notes: Retrieve scenario(s) Available filters for this request are : - name : scenario name GET /scenarios?name=scenario_1 @param name: scenario name @type name: L{string} @in name: query @required name: False @param installer: installer type @type installer: L{string} @in installer: query @required installer: False @param version: version @type version: L{string} @in version: query @required version: False @param project: project name @type project: L{string} @in project: query @required project: False @return 200: all scenarios satisfy queries, empty list if no scenario is found @rtype: L{Scenarios} """ def _set_query(): query = dict() elem_query = dict() for k in self.request.query_arguments.keys(): v = self.get_query_argument(k) if 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 self._list(query=_set_query()) @swagger.operation(nickname="createScenario") def post(self): """ @description: create a new scenario by name @param body: scenario to be created @type body: L{ScenarioCreateRequest} @in body: body @rtype: L{CreateResponse} @return 200: scenario is created. @raise 403: scenario already exists @raise 400: body or name not provided """ def query(): return {'name': self.json_args.get('name')} miss_fields = ['name'] self._create(miss_fields=miss_fields, query=query) class ScenarioGURHandler(GenericScenarioHandler): @swagger.operation(nickname='getScenarioByName') def get(self, name): """ @description: get a single scenario by name @rtype: L{Scenario} @return 200: scenario exist @raise 404: scenario not exist """ self._get_one(query={'name': name}) pass def put(self, name): pass @swagger.operation(nickname="deleteScenarioByName") def delete(self, name): """ @description: delete a scenario by name @return 200: delete success @raise 404: scenario not exist: """ self._delete(query={'name': name}) 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(self, item, action): updates = { ('scores', 'post'): self._update_requests_add_score, ('trust_indicators', 'post'): self._update_requests_add_ti, ('customs', 'post'): self._update_requests_add_customs, ('customs', 'put'): self._update_requests_update_customs, ('customs', 'delete'): self._update_requests_delete_customs, } updates[(item, action)](self.data) return self.data.format() 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): @functools.wraps(xstep) def magic(self, installer): [xstep(self, version) for version in (self._filter_versions(installer.versions))] return magic def iter_projects(xstep): @functools.wraps(xstep) def magic(self, version): [xstep(self, project) for project in (self._filter_projects(version.projects))] return magic @iter_installers @iter_versions @iter_projects def _update_requests_add_score(self, project): project.scores.append( models.ScenarioScore.from_dict(self.body)) @iter_installers @iter_versions @iter_projects def _update_requests_add_ti(self, project): project.trust_indicators.append( models.ScenarioTI.from_dict(self.body)) @iter_installers @iter_versions @iter_projects def _update_requests_add_customs(self, project): project.customs = list(set(project.customs + self.body)) @iter_installers @iter_versions @iter_projects def _update_requests_update_customs(self, project): project.customs = list(set(self.body)) @iter_installers @iter_versions @iter_projects def _update_requests_delete_customs(self, project): project.customs = filter( lambda f: f not in self.body, project.customs) def _filter_installers(self, installers): return self._filter('installer', installers) def _filter_versions(self, versions): return self._filter('version', versions) def _filter_projects(self, projects): return self._filter('project', projects) def _filter(self, item, items): return filter( lambda f: getattr(f, item) == getattr(self, item), items) class GenericScenarioUpdateHandler(GenericScenarioHandler): def __init__(self, application, request, **kwargs): super(GenericScenarioUpdateHandler, self).__init__(application, request, **kwargs) self.installer = None self.version = None self.project = None self.item = None self.action = None def do_update(self, item, action, locators): self.item = item self.action = action for k, v in locators.iteritems(): if not v: v = self.get_query_argument(k) setattr(self, k, v) locators[k] = v self.pure_update(query=self.set_query(locators=locators)) self.finish_request() def _update_requests(self, data): return ScenarioUpdater(data, self.json_args, self.installer, self.version, self.project).update(self.item, self.action) class ScenarioScoresHandler(GenericScenarioUpdateHandler): @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//scores? \ installer=& \ version=& \ project= @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 @return 200: score is created. @raise 404: scenario/installer/version/project not existed """ self.do_update('scores', 'post', locators={'scenario': scenario, 'installer': None, 'version': None, 'project': None}) class ScenarioTIsHandler(GenericScenarioUpdateHandler): @swagger.operation(nickname="addTrustIndicatorRecord") def post(self, scenario): """ @description: add a new trust indicator record @notes: add a new trust indicator record to a project POST /api/v1/scenarios//trust_indicators? \ installer=& \ version=& \ project= @param body: trust indicator to be added @type body: L{ScenarioTI} @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 @return 200: trust indicator is added. @raise 404: scenario/installer/version/project not existed """ self.do_update('trust_indicators', 'post', locators={'scenario': scenario, 'installer': None, 'version': None, 'project': None}) class ScenarioCustomsHandler(GenericScenarioUpdateHandler): @swagger.operation(nickname="addCustomizedTestCases") def post(self, scenario): """ @description: add customized test cases @notes: add several test cases to a project POST /api/v1/scenarios//customs? \ installer=& \ version=& \ project= @param body: test cases to be added @type body: C{list} of L{string} @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 @return 200: test cases are added. @raise 404: scenario/installer/version/project not existed """ self.do_update('customs', 'post', locators={'scenario': scenario, 'installer': None, 'version': None, 'project': None}) @swagger.operation(nickname="updateCustomizedTestCases") def put(self, scenario): """ @description: update customized test cases @notes: substitute all the customized test cases PUT /api/v1/scenarios//customs? \ installer=& \ version=& \ project= @param body: new supported test cases @type body: C{list} of L{string} @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 @return 200: substitute test cases success. @raise 404: scenario/installer/version/project not existed """ self.do_update('customs', 'put', locators={'scenario': scenario, 'installer': None, 'version': None, 'project': None}) @swagger.operation(nickname="deleteCustomizedTestCases") def delete(self, scenario): """ @description: delete one or several customized test cases @notes: delete one or some customized test cases DELETE /api/v1/scenarios//customs? \ installer=& \ version=& \ project= @param body: test case(s) to be deleted @type body: C{list} of L{string} @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 @return 200: delete test case(s) success. @raise 404: scenario/installer/version/project not existed """ self.do_update('customs', 'delete', locators={'scenario': scenario, 'installer': None, 'version': None, 'project': None})