utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py

   1 import functools
   2
   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
   7
   8
   9 class GenericScenarioHandler(handlers.GenericApiHandler):
  10     def __init__(self, application, request, **kwargs):
  11         super(GenericScenarioHandler, self).__init__(application,
  12                                                      request,
  13                                                      **kwargs)
  14         self.table = self.db_scenarios
  15         self.table_cls = models.Scenario
  16
  17
  18 class ScenariosCLHandler(GenericScenarioHandler):
  19     @swagger.operation(nickname="queryScenarios")
  20     def get(self):
  21         """
  22             @description: Retrieve scenario(s).
  23             @notes: Retrieve scenario(s)
  24                 Available filters for this request are :
  25                  - name : scenario name
  26
  27                 GET /scenarios?name=scenario_1
  28             @param name: scenario name
  29             @type name: L{string}
  30             @in name: query
  31             @required name: False
  32             @param installer: installer type
  33             @type installer: L{string}
  34             @in installer: query
  35             @required installer: False
  36             @param version: version
  37             @type version: L{string}
  38             @in version: query
  39             @required version: False
  40             @param project: project name
  41             @type project: L{string}
  42             @in project: query
  43             @required project: False
  44             @return 200: all scenarios satisfy queries,
  45                          empty list if no scenario is found
  46             @rtype: L{Scenarios}
  47         """
  48
  49         def _set_query():
  50             query = dict()
  51             elem_query = dict()
  52             for k in self.request.query_arguments.keys():
  53                 v = self.get_query_argument(k)
  54                 if k == 'installer':
  55                     elem_query["installer"] = v
  56                 elif k == 'version':
  57                     elem_query["versions.version"] = v
  58                 elif k == 'project':
  59                     elem_query["versions.projects.project"] = v
  60                 else:
  61                     query[k] = v
  62             if elem_query:
  63                 query['installers'] = {'$elemMatch': elem_query}
  64             return query
  65
  66         self._list(_set_query())
  67
  68     @swagger.operation(nickname="createScenario")
  69     def post(self):
  70         """
  71             @description: create a new scenario by name
  72             @param body: scenario to be created
  73             @type body: L{ScenarioCreateRequest}
  74             @in body: body
  75             @rtype: L{CreateResponse}
  76             @return 200: scenario is created.
  77             @raise 403: scenario already exists
  78             @raise 400:  body or name not provided
  79         """
  80         def query(data):
  81             return {'name': data.name}
  82
  83         def error(data):
  84             message = '{} already exists as a scenario'.format(data.name)
  85             return constants.HTTP_FORBIDDEN, message
  86
  87         miss_checks = ['name']
  88         db_checks = [(self.table, False, query, error)]
  89         self._create(miss_checks=miss_checks, db_checks=db_checks)
  90
  91
  92 class ScenarioGURHandler(GenericScenarioHandler):
  93     @swagger.operation(nickname='getScenarioByName')
  94     def get(self, name):
  95         """
  96             @description: get a single scenario by name
  97             @rtype: L{Scenario}
  98             @return 200: scenario exist
  99             @raise 404: scenario not exist
 100         """
 101         self._get_one({'name': name})
 102         pass
 103
 104     @swagger.operation(nickname="updateScenarioByName")
 105     def put(self, name):
 106         """
 107             @description: update a single scenario by name
 108             @param body: fields to be updated
 109             @type body: L{ScenarioUpdateRequest}
 110             @in body: body
 111             @rtype: L{Scenario}
 112             @return 200: update success
 113             @raise 404: scenario not exist
 114             @raise 403: nothing to update
 115         """
 116         query = {'name': name}
 117         db_keys = ['name']
 118         self._update(query, db_keys)
 119
 120     @swagger.operation(nickname="deleteScenarioByName")
 121     def delete(self, name):
 122         """
 123         @description: delete a scenario by name
 124         @return 200: delete success
 125         @raise 404: scenario not exist:
 126         """
 127
 128         query = {'name': name}
 129         self._delete(query)
 130
 131     def _update_query(self, keys, data):
 132         query = dict()
 133         equal = True
 134         if self._is_rename():
 135             new = self._term.get('name')
 136             if data.name != new:
 137                 equal = False
 138                 query['name'] = new
 139
 140         return equal, query
 141
 142     def _update_requests(self, data):
 143         updates = {
 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,
 156         }
 157
 158         updates[(self._field, self._op)](data)
 159
 160         return data.format()
 161
 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)]
 167         return magic
 168
 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))]
 174         return magic
 175
 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))]
 181         return magic
 182
 183     def _update_requests_rename(self, data):
 184         data.name = self._term.get('name')
 185
 186     def _update_requests_add_installer(self, data):
 187         data.installers.append(models.ScenarioInstaller.from_dict(self._term))
 188
 189     def _update_requests_delete_installer(self, data):
 190         data.installers = self._remove_installers(data.installers)
 191
 192     @_iter_installers
 193     def _update_requests_add_version(self, installer):
 194         installer.versions.append(models.ScenarioVersion.from_dict(self._term))
 195
 196     @_iter_installers
 197     def _update_requests_delete_version(self, installer):
 198         installer.versions = self._remove_versions(installer.versions)
 199
 200     @_iter_installers
 201     @_iter_versions
 202     def _update_requests_change_owner(self, version):
 203         version.owner = self._term.get('owner')
 204
 205     @_iter_installers
 206     @_iter_versions
 207     def _update_requests_add_project(self, version):
 208         version.projects.append(models.ScenarioProject.from_dict(self._term))
 209
 210     @_iter_installers
 211     @_iter_versions
 212     def _update_requests_delete_project(self, version):
 213         version.projects = self._remove_projects(version.projects)
 214
 215     @_iter_installers
 216     @_iter_versions
 217     @_iter_projects
 218     def _update_requests_add_customs(self, project):
 219         project.customs = list(set(project.customs + self._term))
 220
 221     @_iter_installers
 222     @_iter_versions
 223     @_iter_projects
 224     def _update_requests_delete_customs(self, project):
 225         project.customs = filter(
 226             lambda f: f not in self._term,
 227             project.customs)
 228
 229     @_iter_installers
 230     @_iter_versions
 231     @_iter_projects
 232     def _update_requests_add_score(self, project):
 233         project.scores.append(
 234             models.ScenarioScore.from_dict(self._term))
 235
 236     @_iter_installers
 237     @_iter_versions
 238     @_iter_projects
 239     def _update_requests_add_ti(self, project):
 240         project.trust_indicators.append(
 241             models.ScenarioTI.from_dict(self._term))
 242
 243     def _is_rename(self):
 244         return self._field == 'name' and self._op == 'update'
 245
 246     def _remove_installers(self, installers):
 247         return self._remove('installer', installers)
 248
 249     def _filter_installers(self, installers):
 250         return self._filter('installer', installers)
 251
 252     def _remove_versions(self, versions):
 253         return self._remove('version', versions)
 254
 255     def _filter_versions(self, versions):
 256         return self._filter('version', versions)
 257
 258     def _remove_projects(self, projects):
 259         return self._remove('project', projects)
 260
 261     def _filter_projects(self, projects):
 262         return self._filter('project', projects)
 263
 264     def _remove(self, field, fields):
 265         return filter(
 266             lambda f: getattr(f, field) != self._locate.get(field),
 267             fields)
 268
 269     def _filter(self, field, fields):
 270         return filter(
 271             lambda f: getattr(f, field) == self._locate.get(field),
 272             fields)
 273
 274     @property
 275     def _field(self):
 276         return self.json_args.get('field')
 277
 278     @property
 279     def _op(self):
 280         return self.json_args.get('op')
 281
 282     @property
 283     def _locate(self):
 284         return self.json_args.get('locate')
 285
 286     @property
 287     def _term(self):
 288         return self.json_args.get('term')