path: root/testapi/opnfv_testapi/tests/unit/handlers/test_scenario.py

from datetime import datetime
import functools
import httplib

from opnfv_testapi.common import message
import opnfv_testapi.models.scenario_models as sm
from opnfv_testapi.tests.unit.handlers 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.get_res = sm.Scenario
        self.list_res = sm.Scenarios
        self.basePath = '/api/v1/scenarios'
        self.req_d = self.load_json('scenario-c1')
        self.req_2 = self.load_json('scenario-c2')

    def create_return_name(self, req):
        _, res = self.create(req)
        return res.href.split('/')[-1]

    def assert_res(self, code, scenario, req=None):
        self.assertEqual(code, httplib.OK)
        if req is None:
            req = self.req_d
        self.assertIsNotNone(scenario._id)
        self.assertIsNotNone(scenario.creation_date)
        self.assertEqual(scenario, sm.Scenario.from_dict(req))

    @staticmethod
    def set_query(*args):
        uri = ''
        for arg in args:
            uri += arg + '&'
        return uri[0: -1]

    def get_and_assert(self, name):
        code, body = self.get(name)
        self.assert_res(code, body, self.req_d)


class TestScenarioCreate(TestScenarioBase):
    def test_withoutBody(self):
        (code, body) = self.create()
        self.assertEqual(code, httplib.BAD_REQUEST)

    def test_emptyName(self):
        req_empty = sm.ScenarioCreateRequest('')
        (code, body) = self.create(req_empty)
        self.assertEqual(code, httplib.BAD_REQUEST)
        self.assertIn(message.missing('name'), body)

    def test_noneName(self):
        req_none = sm.ScenarioCreateRequest(None)
        (code, body) = self.create(req_none)
        self.assertEqual(code, httplib.BAD_REQUEST)
        self.assertIn(message.missing('name'), body)

    def test_success(self):
        (code, body) = self.create_d()
        self.assertEqual(code, httplib.OK)
        self.assert_create_body(body)

    def test_alreadyExist(self):
        self.create_d()
        (code, body) = self.create_d()
        self.assertEqual(code, httplib.FORBIDDEN)
        self.assertIn(message.exist_base, body)


class TestScenarioGet(TestScenarioBase):
    def setUp(self):
        super(TestScenarioGet, self).setUp()
        self.scenario_1 = self.create_return_name(self.req_d)
        self.scenario_2 = self.create_return_name(self.req_2)

    def test_getByName(self):
        self.get_and_assert(self.scenario_1)

    def test_getAll(self):
        self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])

    def test_queryName(self):
        query = self.set_query('name=nosdn-nofeature-ha')
        self._query_and_assert(query, reqs=[self.req_d])

    def test_queryInstaller(self):
        query = self.set_query('installer=apex')
        self._query_and_assert(query, reqs=[self.req_d])

    def test_queryVersion(self):
        query = self.set_query('version=master')
        self._query_and_assert(query, reqs=[self.req_d])

    def test_queryProject(self):
        query = self.set_query('project=functest')
        self._query_and_assert(query, reqs=[self.req_d, self.req_2])

    # 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)
        if not found:
            self.assertEqual(code, httplib.OK)
            self.assertEqual(0, len(body.scenarios))
        else:
            self.assertEqual(len(reqs), len(body.scenarios))
            for req in reqs:
                for scenario in body.scenarios:
                    if req['name'] == scenario.name:
                        self.assert_res(code, scenario, req)


class TestScenarioDelete(TestScenarioBase):
    def test_notFound(self):
        code, body = self.delete('notFound')
        self.assertEqual(code, httplib.NOT_FOUND)

    def test_success(self):
        scenario = self.create_return_name(self.req_d)
        code, _ = self.delete(scenario)
        self.assertEqual(code, httplib.OK)
        code, _ = self.get(scenario)
        self.assertEqual(code, httplib.NOT_FOUND)


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_url_fixture(item):
        def _update_url_fixture(xstep):
            def wrapper(self, *args, **kwargs):
                self.update_url = '{}/{}'.format(self.scenario_url, item)
                locator = None
                if item in ['projects', 'owner']:
                    locator = 'installer={}&version={}'.format(
                        self.installer,
                        self.version)
                elif item in ['versions']:
                    locator = 'installer={}'.format(
                        self.installer)
                elif item in ['rename']:
                    self.update_url = self.scenario_url

                if locator:
                    self.update_url = '{}?{}'.format(self.update_url, locator)

                xstep(self, *args, **kwargs)
            return wrapper
        return _update_url_fixture

    def update_partial(operate, expected):
        def _update_partial(set_update):
            @functools.wraps(set_update)
            def wrapper(self):
                update = set_update(self)
                code, body = getattr(self, operate)(update)
                getattr(self, expected)(code)
            return wrapper
        return _update_partial

    @update_partial('_add', '_success')
    def test_addScore(self):
        add = sm.ScenarioScore(date=str(datetime.now()), score='11/12')
        projects = self.req_d['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

    @update_partial('_add', '_success')
    def test_addTrustIndicator(self):
        add = sm.ScenarioTI(date=str(datetime.now()), status='gold')
        projects = self.req_d['installers'][0]['versions'][0]['projects']
        functest = filter(lambda f: f['project'] == 'functest', projects)[0]
        functest['trust_indicators'].append(add.format())
        self.update_url = '{}/trust_indicators?{}'.format(self.scenario_url,
                                                          self.locate_project)

        return add

    @update_partial('_add', '_success')
    def test_addCustoms(self):
        adds = ['odl', 'parser', 'vping_ssh']
        projects = self.req_d['installers'][0]['versions'][0]['projects']
        functest = filter(lambda f: f['project'] == 'functest', projects)[0]
        functest['customs'] = list(set(functest['customs'] + adds))
        self.update_url = '{}/customs?{}'.format(self.scenario_url,
                                                 self.locate_project)
        return adds

    @update_partial('_update', '_success')
    def test_updateCustoms(self):
        updates = ['odl', 'parser', 'vping_ssh']
        projects = self.req_d['installers'][0]['versions'][0]['projects']
        functest = filter(lambda f: f['project'] == 'functest', projects)[0]
        functest['customs'] = updates
        self.update_url = '{}/customs?{}'.format(self.scenario_url,
                                                 self.locate_project)

        return updates

    @update_partial('_delete', '_success')
    def test_deleteCustoms(self):
        deletes = ['vping_ssh']
        projects = self.req_d['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 deletes

    @update_url_fixture('projects')
    @update_partial('_add', '_success')
    def test_addProjects_succ(self):
        add = sm.ScenarioProject(project='qtip').format()
        self.req_d['installers'][0]['versions'][0]['projects'].append(add)
        return [add]

    @update_url_fixture('projects')
    @update_partial('_add', '_conflict')
    def test_addProjects_already_exist(self):
        add = sm.ScenarioProject(project='functest').format()
        return [add]

    @update_url_fixture('projects')
    @update_partial('_add', '_bad_request')
    def test_addProjects_bad_schema(self):
        add = sm.ScenarioProject(project='functest').format()
        add['score'] = None
        return [add]

    @update_url_fixture('projects')
    @update_partial('_update', '_success')
    def test_updateProjects_succ(self):
        update = sm.ScenarioProject(project='qtip').format()
        self.req_d['installers'][0]['versions'][0]['projects'] = [update]
        return [update]

    @update_url_fixture('projects')
    @update_partial('_update', '_conflict')
    def test_updateProjects_duplicated(self):
        update = sm.ScenarioProject(project='qtip').format()
        return [update, update]

    @update_url_fixture('projects')
    @update_partial('_update', '_bad_request')
    def test_updateProjects_bad_schema(self):
        update = sm.ScenarioProject(project='functest').format()
        update['score'] = None
        return [update]

    @update_url_fixture('projects')
    @update_partial('_delete', '_success')
    def test_deleteProjects(self):
        deletes = ['functest']
        projects = self.req_d['installers'][0]['versions'][0]['projects']
        self.req_d['installers'][0]['versions'][0]['projects'] = filter(
            lambda f: f['project'] != 'functest',
            projects)
        return deletes

    @update_url_fixture('owner')
    @update_partial('_update', '_success')
    def test_changeOwner(self):
        new_owner = 'new_owner'
        update = sm.ScenarioChangeOwnerRequest(new_owner).format()
        self.req_d['installers'][0]['versions'][0]['owner'] = new_owner
        return update

    @update_url_fixture('versions')
    @update_partial('_add', '_success')
    def test_addVersions_succ(self):
        add = sm.ScenarioVersion(version='Euphrates').format()
        self.req_d['installers'][0]['versions'].append(add)
        return [add]

    @update_url_fixture('versions')
    @update_partial('_add', '_conflict')
    def test_addVersions_already_exist(self):
        add = sm.ScenarioVersion(version='master').format()
        return [add]

    @update_url_fixture('versions')
    @update_partial('_add', '_bad_request')
    def test_addVersions_bad_schema(self):
        add = sm.ScenarioVersion(version='euphrates').format()
        add['notexist'] = None
        return [add]

    @update_url_fixture('versions')
    @update_partial('_update', '_success')
    def test_updateVersions_succ(self):
        update = sm.ScenarioVersion(version='euphrates').format()
        self.req_d['installers'][0]['versions'] = [update]
        return [update]

    @update_url_fixture('versions')
    @update_partial('_update', '_conflict')
    def test_updateVersions_duplicated(self):
        update = sm.ScenarioVersion(version='euphrates').format()
        return [update, update]

    @update_url_fixture('versions')
    @update_partial('_update', '_bad_request')
    def test_updateVersions_bad_schema(self):
        update = sm.ScenarioVersion(version='euphrates').format()
        update['not_owner'] = 'Iam'
        return [update]

    @update_url_fixture('versions')
    @update_partial('_delete', '_success')
    def test_deleteVersions(self):
        deletes = ['master']
        versions = self.req_d['installers'][0]['versions']
        self.req_d['installers'][0]['versions'] = filter(
            lambda f: f['version'] != 'master',
            versions)
        return deletes

    @update_url_fixture('installers')
    @update_partial('_add', '_success')
    def test_addInstallers_succ(self):
        add = sm.ScenarioInstaller(installer='daisy').format()
        self.req_d['installers'].append(add)
        return [add]

    @update_url_fixture('installers')
    @update_partial('_add', '_conflict')
    def test_addInstallers_already_exist(self):
        add = sm.ScenarioInstaller(installer='apex').format()
        return [add]

    @update_url_fixture('installers')
    @update_partial('_add', '_bad_request')
    def test_addInstallers_bad_schema(self):
        add = sm.ScenarioInstaller(installer='daisy').format()
        add['not_exist'] = 'not_exist'
        return [add]

    @update_url_fixture('installers')
    @update_partial('_update', '_success')
    def test_updateInstallers_succ(self):
        update = sm.ScenarioInstaller(installer='daisy').format()
        self.req_d['installers'] = [update]
        return [update]

    @update_url_fixture('installers')
    @update_partial('_update', '_conflict')
    def test_updateInstallers_duplicated(self):
        update = sm.ScenarioInstaller(installer='daisy').format()
        return [update, update]

    @update_url_fixture('installers')
    @update_partial('_update', '_bad_request')
    def test_updateInstallers_bad_schema(self):
        update = sm.ScenarioInstaller(installer='daisy').format()
        update['not_exist'] = 'not_exist'
        return [update]

    @update_url_fixture('installers')
    @update_partial('_delete', '_success')
    def test_deleteInstallers(self):
        deletes = ['apex']
        installers = self.req_d['installers']
        self.req_d['installers'] = filter(
            lambda f: f['installer'] != 'apex',
            installers)
        return deletes

    @update_url_fixture('rename')
    @update_partial('_update', '_success')
    def test_renameScenario(self):
        new_name = 'new_scenario_name'
        update = sm.ScenarioUpdateRequest(name=new_name)
        self.req_d['name'] = new_name
        return update

    @update_url_fixture('rename')
    @update_partial('_update', '_forbidden')
    def test_renameScenario_exist(self):
        new_name = self.req_d['name']
        update = sm.ScenarioUpdateRequest(name=new_name)
        return update

    def _add(self, update_req):
        return self.post_direct_url(self.update_url, update_req)

    def _update(self, update_req):
        return self.update_direct_url(self.update_url, update_req)

    def _delete(self, update_req):
        return self.delete_direct_url(self.update_url, update_req)

    def _success(self, status):
        self.assertEqual(status, httplib.OK)
        self.get_and_assert(self.req_d['name'])

    def _forbidden(self, status):
        self.assertEqual(status, httplib.FORBIDDEN)

    def _bad_request(self, status):
        self.assertEqual(status, httplib.BAD_REQUEST)

    def _conflict(self, status):
        self.assertEqual(status, httplib.CONFLICT)