diff options
Diffstat (limited to 'utils/test/testapi/opnfv_testapi/tests/unit')
| -rw-r--r-- | utils/test/testapi/opnfv_testapi/tests/unit/resources/test_base.py | 33 | ||||
| -rw-r--r-- | utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py | 118 |
2 files changed, 139 insertions, 12 deletions
diff --git a/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_base.py b/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_base.py index aa6b83544..77a8d18c1 100644 --- a/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_base.py +++ b/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_base.py @@ -92,21 +92,35 @@ class TestBase(testing.AsyncHTTPTestCase): headers=self.headers) return self._get_return(res, self.list_res) - def update(self, new=None, *args): - if new: + def update_direct_url(self, url, new=None): + if new and hasattr(new, 'format'): new = new.format() - res = self.fetch(self._get_uri(*args), + res = self.fetch(url, method='PUT', body=json.dumps(new), headers=self.headers) return self._get_return(res, self.update_res) - def delete(self, *args): - res = self.fetch(self._get_uri(*args), - method='DELETE', - headers=self.headers) + def update(self, new=None, *args): + return self.update_direct_url(self._get_uri(*args), new) + + def delete_direct_url(self, url, body): + if body: + res = self.fetch(url, + method='DELETE', + body=json.dumps(body), + headers=self.headers, + allow_nonstandard_methods=True) + else: + res = self.fetch(url, + method='DELETE', + headers=self.headers) + return res.code, res.body + def delete(self, *args): + return self.delete_direct_url(self._get_uri(*args), None) + @staticmethod def _get_valid_args(*args): new_args = tuple(['%s' % arg for arg in args if arg is not None]) @@ -132,7 +146,10 @@ class TestBase(testing.AsyncHTTPTestCase): def _get_return(self, res, cls): code = res.code body = res.body - return code, self._get_return_body(code, body, cls) + if body: + return code, self._get_return_body(code, body, cls) + else: + return code, None @staticmethod def _get_return_body(code, body, cls): diff --git a/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py b/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py index 0558ea30d..466caaf13 100644 --- a/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py +++ b/utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py @@ -168,15 +168,30 @@ class TestScenarioUpdate(TestScenarioBase): self.version, 'functest') + def update_url_fixture(item): + def _update_url_fixture(xstep): + def wrapper(self, *args, **kwargs): + locator = None + if item in ['projects', 'owner']: + locator = 'installer={}&version={}'.format( + self.installer, + self.version) + self.update_url = '{}/{}?{}'.format(self.scenario_url, + item, + locator) + xstep(self, *args, **kwargs) + return wrapper + return _update_url_fixture + def update_partial(operate, expected): - def _update(set_update): + def _update_partial(set_update): @functools.wraps(set_update) - def wrap(self): + def wrapper(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 + return wrapper + return _update_partial @update_partial('_add', '_success') def test_addScore(self, scenario): @@ -200,9 +215,101 @@ class TestScenarioUpdate(TestScenarioBase): return add, scenario + @update_partial('_add', '_success') + 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'] = list(set(functest['customs'] + add)) + self.update_url = '{}/customs?{}'.format(self.scenario_url, + self.locate_project) + return add, scenario + + @update_partial('_update', '_success') + def test_updateCustoms(self, scenario): + news = ['odl', 'parser', 'vping_ssh'] + projects = scenario['installers'][0]['versions'][0]['projects'] + functest = filter(lambda f: f['project'] == 'functest', projects)[0] + functest['customs'] = news + self.update_url = '{}/customs?{}'.format(self.scenario_url, + self.locate_project) + + return news, scenario + + @update_partial('_delete', '_success') + def test_deleteCustoms(self, scenario): + obsoletes = ['vping_ssh'] + projects = scenario['installers'][0]['versions'][0]['projects'] + functest = filter(lambda f: f['project'] == 'functest', projects)[0] + functest['customs'] = ['healthcheck'] + self.update_url = '{}/customs?{}'.format(self.scenario_url, + self.locate_project) + + return obsoletes, scenario + + @update_url_fixture('projects') + @update_partial('_add', '_success') + def test_addProjects_succ(self, scenario): + add = models.ScenarioProject(project='qtip').format() + scenario['installers'][0]['versions'][0]['projects'].append(add) + return [add], scenario + + @update_url_fixture('projects') + @update_partial('_add', '_conflict') + def test_addProjects_already_exist(self, scenario): + add = models.ScenarioProject(project='functest').format() + scenario['installers'][0]['versions'][0]['projects'].append(add) + return [add], scenario + + @update_url_fixture('projects') + @update_partial('_add', '_bad_request') + def test_addProjects_bad_schema(self, scenario): + add = models.ScenarioProject(project='functest').format() + add['score'] = None + scenario['installers'][0]['versions'][0]['projects'].append(add) + return [add], scenario + + @update_url_fixture('projects') + @update_partial('_update', '_success') + def test_updateProjects_succ(self, scenario): + update = models.ScenarioProject(project='qtip').format() + scenario['installers'][0]['versions'][0]['projects'] = [update] + return [update], scenario + + @update_url_fixture('projects') + @update_partial('_update', '_bad_request') + def test_updateProjects_bad_schema(self, scenario): + update = models.ScenarioProject(project='functest').format() + update['score'] = None + scenario['installers'][0]['versions'][0]['projects'] = [update] + return [update], scenario + + @update_url_fixture('projects') + @update_partial('_delete', '_success') + def test_deleteProjects(self, scenario): + deletes = ['functest'] + projects = scenario['installers'][0]['versions'][0]['projects'] + scenario['installers'][0]['versions'][0]['projects'] = filter( + lambda f: f['project'] != 'functest', + projects) + return deletes, scenario + + @update_url_fixture('owner') + @update_partial('_update', '_success') + def test_changeOwner(self, scenario): + new_owner = 'new_owner' + scenario['installers'][0]['versions'][0]['owner'] = 'www' + return new_owner, scenario + def _add(self, update_req, new_scenario): return self.post_direct_url(self.update_url, update_req) + def _update(self, update_req, new_scenario): + return self.update_direct_url(self.update_url, update_req) + + def _delete(self, update_req, new_scenario): + return self.delete_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) @@ -212,3 +319,6 @@ class TestScenarioUpdate(TestScenarioBase): def _bad_request(self, status, new_scenario): self.assertEqual(status, httplib.BAD_REQUEST) + + def _conflict(self, status, new_scenario): + self.assertEqual(status, httplib.CONFLICT) |