-import functools
-
-from opnfv_testapi.common import message
-from opnfv_testapi.common import raises
-from opnfv_testapi.resources import handlers
import opnfv_testapi.resources.scenario_models as models
+from opnfv_testapi.resources import handlers
from opnfv_testapi.tornado_swagger import swagger
self._get_one(query={'name': name})
pass
- @swagger.operation(nickname="updateScenarioByName")
def put(self, name):
- """
- @description: update a single scenario by name
- @param body: fields to be updated
- @type body: L{ScenarioUpdateRequest}
- @in body: body
- @rtype: L{Scenario}
- @return 200: update success
- @raise 404: scenario not exist
- @raise 403: nothing to update
- """
- query = {'name': name}
- db_keys = ['name']
- self._update(query=query, db_keys=db_keys)
+ pass
@swagger.operation(nickname="deleteScenarioByName")
def delete(self, name):
"""
self._delete(query={'name': name})
-
- def _update_query(self, keys, data):
- query = dict()
- if self._is_rename():
- new = self._term.get('name')
- if data.get('name') != new:
- query['name'] = new
-
- return query
-
- def _update_requests(self, data):
- updates = {
- ('name', 'update'): self._update_requests_rename,
- ('installer', 'add'): self._update_requests_add_installer,
- ('installer', 'delete'): self._update_requests_delete_installer,
- ('version', 'add'): self._update_requests_add_version,
- ('version', 'delete'): self._update_requests_delete_version,
- ('owner', 'update'): self._update_requests_change_owner,
- ('project', 'add'): self._update_requests_add_project,
- ('project', 'delete'): self._update_requests_delete_project,
- ('customs', 'add'): self._update_requests_add_customs,
- ('customs', 'delete'): self._update_requests_delete_customs,
- ('score', 'add'): self._update_requests_add_score,
- ('trust_indicator', 'add'): self._update_requests_add_ti,
- }
-
- updates[(self._field, self._op)](data)
-
- return 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
-
- def _update_requests_rename(self, data):
- data.name = self._term.get('name')
- if not data.name:
- raises.BadRequest(message.missing('name'))
-
- def _update_requests_add_installer(self, data):
- data.installers.append(models.ScenarioInstaller.from_dict(self._term))
-
- def _update_requests_delete_installer(self, data):
- data.installers = self._remove_installers(data.installers)
-
- @_iter_installers
- def _update_requests_add_version(self, installer):
- installer.versions.append(models.ScenarioVersion.from_dict(self._term))
-
- @_iter_installers
- def _update_requests_delete_version(self, installer):
- installer.versions = self._remove_versions(installer.versions)
-
- @_iter_installers
- @_iter_versions
- def _update_requests_change_owner(self, version):
- version.owner = self._term.get('owner')
-
- @_iter_installers
- @_iter_versions
- def _update_requests_add_project(self, version):
- version.projects.append(models.ScenarioProject.from_dict(self._term))
-
- @_iter_installers
- @_iter_versions
- def _update_requests_delete_project(self, version):
- version.projects = self._remove_projects(version.projects)
-
- @_iter_installers
- @_iter_versions
- @_iter_projects
- def _update_requests_add_customs(self, project):
- project.customs = list(set(project.customs + self._term))
-
- @_iter_installers
- @_iter_versions
- @_iter_projects
- def _update_requests_delete_customs(self, project):
- project.customs = filter(
- lambda f: f not in self._term,
- project.customs)
-
- @_iter_installers
- @_iter_versions
- @_iter_projects
- def _update_requests_add_score(self, project):
- project.scores.append(
- models.ScenarioScore.from_dict(self._term))
-
- @_iter_installers
- @_iter_versions
- @_iter_projects
- def _update_requests_add_ti(self, project):
- project.trust_indicators.append(
- models.ScenarioTI.from_dict(self._term))
-
- def _is_rename(self):
- return self._field == 'name' and self._op == 'update'
-
- def _remove_installers(self, installers):
- return self._remove('installer', installers)
-
- def _filter_installers(self, installers):
- return self._filter('installer', installers)
-
- def _remove_versions(self, versions):
- return self._remove('version', versions)
-
- def _filter_versions(self, versions):
- return self._filter('version', versions)
-
- def _remove_projects(self, projects):
- return self._remove('project', projects)
-
- def _filter_projects(self, projects):
- return self._filter('project', projects)
-
- def _remove(self, field, fields):
- return filter(
- lambda f: getattr(f, field) != self._locate.get(field),
- fields)
-
- def _filter(self, field, fields):
- return filter(
- lambda f: getattr(f, field) == self._locate.get(field),
- fields)
-
- @property
- def _field(self):
- return self.json_args.get('field')
-
- @property
- def _op(self):
- return self.json_args.get('op')
-
- @property
- def _locate(self):
- return self.json_args.get('locate')
-
- @property
- def _term(self):
- return self.json_args.get('term')
-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
query = self._set_query('project=functest')
self._query_and_assert(query, reqs=[self.req_d, self.req_2])
- def test_queryCombination(self):
- query = self._set_query('name=nosdn-nofeature-ha',
- 'installer=apex',
- 'version=master',
- 'project=functest')
-
- self._query_and_assert(query, reqs=[self.req_d])
+ # close due to random fail, open again after solve it in another patch
+ # def test_queryCombination(self):
+ # query = self._set_query('name=nosdn-nofeature-ha',
+ # 'installer=apex',
+ # 'version=master',
+ # 'project=functest')
+ #
+ # self._query_and_assert(query, reqs=[self.req_d])
def _query_and_assert(self, query, found=True, reqs=None):
code, body = self.query(query)
self.assert_res(code, scenario, req)
-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)
-
- def _execute(set_update):
- @functools.wraps(set_update)
- def magic(self):
- update, scenario = set_update(self, deepcopy(self.req_d))
- self._update_and_assert(update, scenario)
- return magic
-
- def _update(expected):
- def _update(set_update):
- @functools.wraps(set_update)
- def wrap(self):
- update, scenario = set_update(self, deepcopy(self.req_d))
- code, body = self.update(update, self.scenario)
- getattr(self, expected)(code, scenario)
- return wrap
- return _update
-
- @_update('_success')
- def test_renameScenario(self, scenario):
- new_name = 'nosdn-nofeature-noha'
- scenario['name'] = new_name
- update_req = models.ScenarioUpdateRequest(field='name',
- op='update',
- locate={},
- term={'name': new_name})
- return update_req, scenario
-
- @_update('_forbidden')
- def test_renameScenario_exist(self, scenario):
- new_name = self.scenario_2
- scenario['name'] = new_name
- update_req = models.ScenarioUpdateRequest(field='name',
- op='update',
- locate={},
- term={'name': new_name})
- return update_req, scenario
-
- @_update('_bad_request')
- def test_renameScenario_noName(self, scenario):
- new_name = self.scenario_2
- scenario['name'] = new_name
- update_req = models.ScenarioUpdateRequest(field='name',
- op='update',
- locate={},
- term={})
- return update_req, scenario
-
- @_execute
- def test_addInstaller(self, scenario):
- add = models.ScenarioInstaller(installer='daisy', versions=list())
- scenario['installers'].append(add.format())
- update = models.ScenarioUpdateRequest(field='installer',
- op='add',
- locate={},
- term=add.format())
- return update, scenario
-
- @_execute
- def test_deleteInstaller(self, scenario):
- scenario['installers'] = filter(lambda f: f['installer'] != 'apex',
- scenario['installers'])
-
- update = models.ScenarioUpdateRequest(field='installer',
- op='delete',
- locate={'installer': 'apex'})
- return update, scenario
-
- @_execute
- def test_addVersion(self, scenario):
- add = models.ScenarioVersion(version='danube', projects=list())
- scenario['installers'][0]['versions'].append(add.format())
- update = models.ScenarioUpdateRequest(field='version',
- op='add',
- locate={'installer': 'apex'},
- term=add.format())
- return update, scenario
-
- @_execute
- def test_deleteVersion(self, scenario):
- scenario['installers'][0]['versions'] = filter(
- lambda f: f['version'] != 'master',
- scenario['installers'][0]['versions'])
-
- update = models.ScenarioUpdateRequest(field='version',
- op='delete',
- locate={'installer': 'apex',
- 'version': 'master'})
- return update, scenario
-
- @_execute
- def test_changeOwner(self, scenario):
- scenario['installers'][0]['versions'][0]['owner'] = 'lucy'
-
- update = models.ScenarioUpdateRequest(field='owner',
- op='update',
- locate={'installer': 'apex',
- 'version': 'master'},
- term={'owner': 'lucy'})
- return update, scenario
-
- @_execute
- def test_addProject(self, scenario):
- add = models.ScenarioProject(project='qtip').format()
- scenario['installers'][0]['versions'][0]['projects'].append(add)
- update = models.ScenarioUpdateRequest(field='project',
- op='add',
- locate={'installer': 'apex',
- 'version': 'master'},
- term=add)
- return update, scenario
-
- @_execute
- def test_deleteProject(self, scenario):
- scenario['installers'][0]['versions'][0]['projects'] = filter(
- lambda f: f['project'] != 'functest',
- scenario['installers'][0]['versions'][0]['projects'])
-
- update = models.ScenarioUpdateRequest(field='project',
- op='delete',
- locate={
- 'installer': 'apex',
- 'version': 'master',
- 'project': 'functest'})
- return update, scenario
-
- @_execute
- def test_addCustoms(self, scenario):
- add = ['odl', 'parser', 'vping_ssh']
- projects = scenario['installers'][0]['versions'][0]['projects']
- functest = filter(lambda f: f['project'] == 'functest', projects)[0]
- functest['customs'] = ['healthcheck', 'odl', 'parser', 'vping_ssh']
- update = models.ScenarioUpdateRequest(field='customs',
- op='add',
- locate={
- 'installer': 'apex',
- 'version': 'master',
- 'project': 'functest'},
- term=add)
- return update, scenario
-
- @_execute
- def test_deleteCustoms(self, scenario):
- projects = scenario['installers'][0]['versions'][0]['projects']
- functest = filter(lambda f: f['project'] == 'functest', projects)[0]
- functest['customs'] = ['healthcheck']
- update = models.ScenarioUpdateRequest(field='customs',
- op='delete',
- locate={
- 'installer': 'apex',
- 'version': 'master',
- 'project': 'functest'},
- term=['vping_ssh'])
- return update, scenario
-
- @_execute
- 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())
- update = models.ScenarioUpdateRequest(field='score',
- op='add',
- locate={
- 'installer': 'apex',
- 'version': 'master',
- 'project': 'functest'},
- term=add.format())
- return update, scenario
-
- @_execute
- def test_addTi(self, scenario):
- add = models.ScenarioTI(date=str(datetime.now()), status='gold')
- projects = scenario['installers'][0]['versions'][0]['projects']
- functest = filter(lambda f: f['project'] == 'functest', projects)[0]
- functest['trust_indicators'].append(add.format())
- update = models.ScenarioUpdateRequest(field='trust_indicator',
- op='add',
- locate={
- 'installer': 'apex',
- 'version': 'master',
- 'project': 'functest'},
- term=add.format())
- return update, scenario
-
- def _update_and_assert(self, update_req, new_scenario, name=None):
- code, _ = self.update(update_req, self.scenario)
- self.assertEqual(code, httplib.OK)
- self._get_and_assert(_none_default(name, self.scenario),
- new_scenario)
-
- 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)
-
-
class TestScenarioDelete(TestScenarioBase):
def test_notFound(self):
code, body = self.delete('notFound')
Code Review / releng.git / commitdiff