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

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