-import opnfv_testapi.resources.scenario_models as models
+import functools
+
from opnfv_testapi.resources import handlers
+import opnfv_testapi.resources.scenario_models as models
from opnfv_testapi.tornado_swagger import swagger
self.table = self.db_scenarios
self.table_cls = models.Scenario
+ def set_query(self, filters):
+ query = dict()
+ elem_query = dict()
+ for k, v in filters.iteritems():
+ if k == 'scenario':
+ query['name'] = v
+ elif k == 'installer':
+ elem_query["installer"] = v
+ elif k == 'version':
+ elem_query["versions.version"] = v
+ elif k == 'project':
+ elem_query["versions.projects.project"] = v
+ else:
+ query[k] = v
+ if elem_query:
+ query['installers'] = {'$elemMatch': elem_query}
+ return query
+
class ScenariosCLHandler(GenericScenarioHandler):
@swagger.operation(nickname="queryScenarios")
@return 200: delete success
@raise 404: scenario not exist:
"""
-
self._delete(query={'name': name})
+
+
+class ScenarioUpdater(object):
+ def __init__(self, data, body=None,
+ installer=None, version=None, project=None):
+ self.data = data
+ self.body = body
+ self.installer = installer
+ self.version = version
+ self.project = project
+
+ def update(self, item, op):
+ updates = {
+ ('score', 'add'): self._update_requests_add_score,
+ }
+ updates[(item, op)](self.data)
+
+ return self.data.format()
+
+ def iter_installers(xstep):
+ @functools.wraps(xstep)
+ def magic(self, data):
+ [xstep(self, installer)
+ for installer in self._filter_installers(data.installers)]
+ return magic
+
+ def iter_versions(xstep):
+ @functools.wraps(xstep)
+ def magic(self, installer):
+ [xstep(self, version)
+ for version in (self._filter_versions(installer.versions))]
+ return magic
+
+ def iter_projects(xstep):
+ @functools.wraps(xstep)
+ def magic(self, version):
+ [xstep(self, project)
+ for project in (self._filter_projects(version.projects))]
+ return magic
+
+ @iter_installers
+ @iter_versions
+ @iter_projects
+ def _update_requests_add_score(self, project):
+ project.scores.append(
+ models.ScenarioScore.from_dict(self.body))
+
+ def _filter_installers(self, installers):
+ return self._filter('installer', installers)
+
+ def _filter_versions(self, versions):
+ return self._filter('version', versions)
+
+ def _filter_projects(self, projects):
+ return self._filter('project', projects)
+
+ def _filter(self, item, items):
+ return filter(
+ lambda f: getattr(f, item) == getattr(self, item),
+ items)
+
+
+class ScenarioScoresHandler(GenericScenarioHandler):
+ @swagger.operation(nickname="addScoreRecord")
+ def post(self, scenario):
+ """
+ @description: add a new score record
+ @notes: add a new score record to a project
+ POST /api/v1/scenarios/<scenario_name>/scores? \
+ installer=<installer_name>& \
+ version=<version_name>& \
+ project=<project_name>
+ @param body: score to be added
+ @type body: L{ScenarioScore}
+ @in body: body
+ @param installer: installer type
+ @type installer: L{string}
+ @in installer: query
+ @required installer: True
+ @param version: version
+ @type version: L{string}
+ @in version: query
+ @required version: True
+ @param project: project name
+ @type project: L{string}
+ @in project: query
+ @required project: True
+ @rtype: L{Scenario}
+ @return 200: score is created.
+ @raise 404: scenario/installer/version/project not existed
+ """
+ self.installer = self.get_query_argument('installer')
+ self.version = self.get_query_argument('version')
+ self.project = self.get_query_argument('project')
+
+ filters = {'scenario': scenario,
+ 'installer': self.installer,
+ 'version': self.version,
+ 'project': self.project}
+ db_keys = ['name']
+ self._update(query=self.set_query(filters=filters), db_keys=db_keys)
+
+ def _update_requests(self, data):
+ return ScenarioUpdater(data,
+ self.json_args,
+ self.installer,
+ self.version,
+ self.project).update('score', 'add')
+import functools
import httplib
import json
import os
+from copy import deepcopy
+from datetime import datetime
-import opnfv_testapi.resources.scenario_models as models
from opnfv_testapi.common import message
+import opnfv_testapi.resources.scenario_models as models
from opnfv_testapi.tests.unit.resources import test_base as base
+def _none_default(check, default):
+ return check if check else default
+
+
class TestScenarioBase(base.TestBase):
def setUp(self):
super(TestScenarioBase, self).setUp()
self.assertEqual(code, httplib.NOT_FOUND)
-def _none_default(check, default):
- return check if check else default
+class TestScenarioUpdate(TestScenarioBase):
+ def setUp(self):
+ super(TestScenarioUpdate, self).setUp()
+ self.scenario = self.create_return_name(self.req_d)
+ self.scenario_2 = self.create_return_name(self.req_2)
+ self.update_url = ''
+ self.scenario_url = '/api/v1/scenarios/{}'.format(self.scenario)
+ self.installer = self.req_d['installers'][0]['installer']
+ self.version = self.req_d['installers'][0]['versions'][0]['version']
+ self.locate_project = 'installer={}&version={}&project={}'.format(
+ self.installer,
+ self.version,
+ 'functest')
+
+ def update_partial(operate, expected):
+ def _update(set_update):
+ @functools.wraps(set_update)
+ def wrap(self):
+ update, scenario = set_update(self, deepcopy(self.req_d))
+ code, body = getattr(self, operate)(update, self.scenario)
+ getattr(self, expected)(code, scenario)
+ return wrap
+ return _update
+
+ @update_partial('_add', '_success')
+ def test_addScore(self, scenario):
+ add = models.ScenarioScore(date=str(datetime.now()), score='11/12')
+ projects = scenario['installers'][0]['versions'][0]['projects']
+ functest = filter(lambda f: f['project'] == 'functest', projects)[0]
+ functest['scores'].append(add.format())
+ self.update_url = '{}/scores?{}'.format(self.scenario_url,
+ self.locate_project)
+
+ return add, scenario
+
+ def _add(self, update_req, new_scenario):
+ return self.post_direct_url(self.update_url, update_req)
+
+ def _success(self, status, new_scenario):
+ self.assertEqual(status, httplib.OK)
+ self._get_and_assert(new_scenario.get('name'), new_scenario)
+
+ def _forbidden(self, status, new_scenario):
+ self.assertEqual(status, httplib.FORBIDDEN)
+
+ def _bad_request(self, status, new_scenario):
+ self.assertEqual(status, httplib.BAD_REQUEST)
Code Review / releng.git / commitdiff