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

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