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

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