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

   1 import functools
   2
   3 from opnfv_testapi.resources import handlers
   4 import opnfv_testapi.resources.scenario_models as models
   5 from opnfv_testapi.tornado_swagger import swagger
   6
   7
   8 class GenericScenarioHandler(handlers.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 = models.Scenario
  15
  16     def set_query(self, locators):
  17         query = dict()
  18         elem_query = dict()
  19         for k, v in locators.iteritems():
  20             if k == 'scenario':
  21                 query['name'] = v
  22             elif k == 'installer':
  23                 elem_query["installer"] = v
  24             elif k == 'version':
  25                 elem_query["versions.version"] = v
  26             elif k == 'project':
  27                 elem_query["versions.projects.project"] = v
  28             else:
  29                 query[k] = v
  30         if elem_query:
  31             query['installers'] = {'$elemMatch': elem_query}
  32         return query
  33
  34
  35 class ScenariosCLHandler(GenericScenarioHandler):
  36     @swagger.operation(nickname="queryScenarios")
  37     def get(self):
  38         """
  39             @description: Retrieve scenario(s).
  40             @notes: Retrieve scenario(s)
  41                 Available filters for this request are :
  42                  - name : scenario name
  43
  44                 GET /scenarios?name=scenario_1
  45             @param name: scenario name
  46             @type name: L{string}
  47             @in name: query
  48             @required name: False
  49             @param installer: installer type
  50             @type installer: L{string}
  51             @in installer: query
  52             @required installer: False
  53             @param version: version
  54             @type version: L{string}
  55             @in version: query
  56             @required version: False
  57             @param project: project name
  58             @type project: L{string}
  59             @in project: query
  60             @required project: False
  61             @return 200: all scenarios satisfy queries,
  62                          empty list if no scenario is found
  63             @rtype: L{Scenarios}
  64         """
  65
  66         def _set_query():
  67             query = dict()
  68             elem_query = dict()
  69             for k in self.request.query_arguments.keys():
  70                 v = self.get_query_argument(k)
  71                 if k == 'installer':
  72                     elem_query["installer"] = v
  73                 elif k == 'version':
  74                     elem_query["versions.version"] = v
  75                 elif k == 'project':
  76                     elem_query["versions.projects.project"] = v
  77                 else:
  78                     query[k] = v
  79             if elem_query:
  80                 query['installers'] = {'$elemMatch': elem_query}
  81             return query
  82
  83         self._list(query=_set_query())
  84
  85     @swagger.operation(nickname="createScenario")
  86     def post(self):
  87         """
  88             @description: create a new scenario by name
  89             @param body: scenario to be created
  90             @type body: L{ScenarioCreateRequest}
  91             @in body: body
  92             @rtype: L{CreateResponse}
  93             @return 200: scenario is created.
  94             @raise 403: scenario already exists
  95             @raise 400:  body or name not provided
  96         """
  97         def query():
  98             return {'name': self.json_args.get('name')}
  99         miss_fields = ['name']
 100         self._create(miss_fields=miss_fields, query=query)
 101
 102
 103 class ScenarioGURHandler(GenericScenarioHandler):
 104     @swagger.operation(nickname='getScenarioByName')
 105     def get(self, name):
 106         """
 107             @description: get a single scenario by name
 108             @rtype: L{Scenario}
 109             @return 200: scenario exist
 110             @raise 404: scenario not exist
 111         """
 112         self._get_one(query={'name': name})
 113         pass
 114
 115     def put(self, name):
 116         pass
 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         self._delete(query={'name': name})
 126
 127
 128 class ScenarioUpdater(object):
 129     def __init__(self, data, body=None,
 130                  installer=None, version=None, project=None):
 131         self.data = data
 132         self.body = body
 133         self.installer = installer
 134         self.version = version
 135         self.project = project
 136
 137     def update(self, item, action):
 138         updates = {
 139             ('scores', 'post'): self._update_requests_add_score,
 140             ('trust_indicators', 'post'): self._update_requests_add_ti,
 141         }
 142         updates[(item, action)](self.data)
 143
 144         return self.data.format()
 145
 146     def iter_installers(xstep):
 147         @functools.wraps(xstep)
 148         def magic(self, data):
 149             [xstep(self, installer)
 150              for installer in self._filter_installers(data.installers)]
 151         return magic
 152
 153     def iter_versions(xstep):
 154         @functools.wraps(xstep)
 155         def magic(self, installer):
 156             [xstep(self, version)
 157              for version in (self._filter_versions(installer.versions))]
 158         return magic
 159
 160     def iter_projects(xstep):
 161         @functools.wraps(xstep)
 162         def magic(self, version):
 163             [xstep(self, project)
 164              for project in (self._filter_projects(version.projects))]
 165         return magic
 166
 167     @iter_installers
 168     @iter_versions
 169     @iter_projects
 170     def _update_requests_add_score(self, project):
 171         project.scores.append(
 172             models.ScenarioScore.from_dict(self.body))
 173
 174     @iter_installers
 175     @iter_versions
 176     @iter_projects
 177     def _update_requests_add_ti(self, project):
 178         project.trust_indicators.append(
 179             models.ScenarioTI.from_dict(self.body))
 180
 181     def _filter_installers(self, installers):
 182         return self._filter('installer', installers)
 183
 184     def _filter_versions(self, versions):
 185         return self._filter('version', versions)
 186
 187     def _filter_projects(self, projects):
 188         return self._filter('project', projects)
 189
 190     def _filter(self, item, items):
 191         return filter(
 192             lambda f: getattr(f, item) == getattr(self, item),
 193             items)
 194
 195
 196 class GenericScenarioUpdateHandler(GenericScenarioHandler):
 197     def __init__(self, application, request, **kwargs):
 198         super(GenericScenarioUpdateHandler, self).__init__(application,
 199                                                            request,
 200                                                            **kwargs)
 201         self.installer = None
 202         self.version = None
 203         self.project = None
 204         self.item = None
 205         self.action = None
 206
 207     def do_post(self, scenario, item, action, locators):
 208         self.item = item
 209         self.action = action
 210         for k in locators.keys():
 211             v = self.get_query_argument(k)
 212             setattr(self, k, v)
 213             locators[k] = v
 214         db_keys = ['name']
 215         self._update(query=self.set_query(locators=locators), db_keys=db_keys)
 216
 217     def _update_requests(self, data):
 218         return ScenarioUpdater(data,
 219                                self.json_args,
 220                                self.installer,
 221                                self.version,
 222                                self.project).update(self.item, self.action)
 223
 224
 225 class ScenarioScoresHandler(GenericScenarioUpdateHandler):
 226     @swagger.operation(nickname="addScoreRecord")
 227     def post(self, scenario):
 228         """
 229         @description: add a new score record
 230         @notes: add a new score record to a project
 231             POST /api/v1/scenarios/<scenario_name>/scores? \
 232                 installer=<installer_name>& \
 233                 version=<version_name>& \
 234                 project=<project_name>
 235         @param body: score to be added
 236         @type body: L{ScenarioScore}
 237         @in body: body
 238         @param installer: installer type
 239         @type installer: L{string}
 240         @in installer: query
 241         @required installer: True
 242         @param version: version
 243         @type version: L{string}
 244         @in version: query
 245         @required version: True
 246         @param project: project name
 247         @type project: L{string}
 248         @in project: query
 249         @required project: True
 250         @rtype: L{Scenario}
 251         @return 200: score is created.
 252         @raise 404:  scenario/installer/version/project not existed
 253         """
 254         self.do_post(scenario,
 255                      'scores',
 256                      'post',
 257                      locators={'installer': None,
 258                                'version': None,
 259                                'project': None})
 260
 261
 262 class ScenarioTIsHandler(GenericScenarioUpdateHandler):
 263     @swagger.operation(nickname="addTrustIndicatorRecord")
 264     def post(self, scenario):
 265         """
 266         @description: add a new trust indicator record
 267         @notes: add a new trust indicator record to a project
 268             POST /api/v1/scenarios/<scenario_name>/trust_indicators? \
 269                 installer=<installer_name>& \
 270                 version=<version_name>& \
 271                 project=<project_name>
 272         @param body: trust indicator to be added
 273         @type body: L{ScenarioTI}
 274         @in body: body
 275         @param installer: installer type
 276         @type installer: L{string}
 277         @in installer: query
 278         @required installer: True
 279         @param version: version
 280         @type version: L{string}
 281         @in version: query
 282         @required version: True
 283         @param project: project name
 284         @type project: L{string}
 285         @in project: query
 286         @required project: True
 287         @rtype: L{Scenario}
 288         @return 200: trust indicator is added.
 289         @raise 404:  scenario/installer/version/project not existed
 290         """
 291         self.do_post(scenario,
 292                      'trust_indicators',
 293                      'post',
 294                      locators={'installer': None,
 295                                'version': None,
 296                                'project': None})