utils/test/testapi/opnfv_testapi/tests/unit/resources/test_scenario.py

   1 import functools
   2 import httplib
   3 import json
   4 import os
   5 from copy import deepcopy
   6 from datetime import datetime
   7
   8 from opnfv_testapi.common import message
   9 import opnfv_testapi.resources.scenario_models as models
  10 from opnfv_testapi.tests.unit.resources import test_base as base
  11
  12
  13 def _none_default(check, default):
  14     return check if check else default
  15
  16
  17 class TestScenarioBase(base.TestBase):
  18     def setUp(self):
  19         super(TestScenarioBase, self).setUp()
  20         self.get_res = models.Scenario
  21         self.list_res = models.Scenarios
  22         self.basePath = '/api/v1/scenarios'
  23         self.req_d = self._load_request('scenario-c1.json')
  24         self.req_2 = self._load_request('scenario-c2.json')
  25
  26     def tearDown(self):
  27         pass
  28
  29     def assert_body(self, project, req=None):
  30         pass
  31
  32     @staticmethod
  33     def _load_request(f_req):
  34         abs_file = os.path.join(os.path.dirname(__file__), f_req)
  35         with open(abs_file, 'r') as f:
  36             loader = json.load(f)
  37             f.close()
  38         return loader
  39
  40     def create_return_name(self, req):
  41         _, res = self.create(req)
  42         return res.href.split('/')[-1]
  43
  44     def assert_res(self, code, scenario, req=None):
  45         self.assertEqual(code, httplib.OK)
  46         if req is None:
  47             req = self.req_d
  48         self.assertIsNotNone(scenario._id)
  49         self.assertIsNotNone(scenario.creation_date)
  50
  51         scenario == models.Scenario.from_dict(req)
  52
  53     @staticmethod
  54     def _set_query(*args):
  55         uri = ''
  56         for arg in args:
  57             uri += arg + '&'
  58         return uri[0: -1]
  59
  60     def _get_and_assert(self, name, req=None):
  61         code, body = self.get(name)
  62         self.assert_res(code, body, req)
  63
  64
  65 class TestScenarioCreate(TestScenarioBase):
  66     def test_withoutBody(self):
  67         (code, body) = self.create()
  68         self.assertEqual(code, httplib.BAD_REQUEST)
  69
  70     def test_emptyName(self):
  71         req_empty = models.ScenarioCreateRequest('')
  72         (code, body) = self.create(req_empty)
  73         self.assertEqual(code, httplib.BAD_REQUEST)
  74         self.assertIn(message.missing('name'), body)
  75
  76     def test_noneName(self):
  77         req_none = models.ScenarioCreateRequest(None)
  78         (code, body) = self.create(req_none)
  79         self.assertEqual(code, httplib.BAD_REQUEST)
  80         self.assertIn(message.missing('name'), body)
  81
  82     def test_success(self):
  83         (code, body) = self.create_d()
  84         self.assertEqual(code, httplib.OK)
  85         self.assert_create_body(body)
  86
  87     def test_alreadyExist(self):
  88         self.create_d()
  89         (code, body) = self.create_d()
  90         self.assertEqual(code, httplib.FORBIDDEN)
  91         self.assertIn(message.exist_base, body)
  92
  93
  94 class TestScenarioGet(TestScenarioBase):
  95     def setUp(self):
  96         super(TestScenarioGet, self).setUp()
  97         self.scenario_1 = self.create_return_name(self.req_d)
  98         self.scenario_2 = self.create_return_name(self.req_2)
  99
 100     def test_getByName(self):
 101         self._get_and_assert(self.scenario_1, self.req_d)
 102
 103     def test_getAll(self):
 104         self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
 105
 106     def test_queryName(self):
 107         query = self._set_query('name=nosdn-nofeature-ha')
 108         self._query_and_assert(query, reqs=[self.req_d])
 109
 110     def test_queryInstaller(self):
 111         query = self._set_query('installer=apex')
 112         self._query_and_assert(query, reqs=[self.req_d])
 113
 114     def test_queryVersion(self):
 115         query = self._set_query('version=master')
 116         self._query_and_assert(query, reqs=[self.req_d])
 117
 118     def test_queryProject(self):
 119         query = self._set_query('project=functest')
 120         self._query_and_assert(query, reqs=[self.req_d, self.req_2])
 121
 122     # close due to random fail, open again after solve it in another patch
 123     # def test_queryCombination(self):
 124     #     query = self._set_query('name=nosdn-nofeature-ha',
 125     #                             'installer=apex',
 126     #                             'version=master',
 127     #                             'project=functest')
 128     #
 129     #     self._query_and_assert(query, reqs=[self.req_d])
 130
 131     def _query_and_assert(self, query, found=True, reqs=None):
 132         code, body = self.query(query)
 133         if not found:
 134             self.assertEqual(code, httplib.OK)
 135             self.assertEqual(0, len(body.scenarios))
 136         else:
 137             self.assertEqual(len(reqs), len(body.scenarios))
 138             for req in reqs:
 139                 for scenario in body.scenarios:
 140                     if req['name'] == scenario.name:
 141                         self.assert_res(code, scenario, req)
 142
 143
 144 class TestScenarioDelete(TestScenarioBase):
 145     def test_notFound(self):
 146         code, body = self.delete('notFound')
 147         self.assertEqual(code, httplib.NOT_FOUND)
 148
 149     def test_success(self):
 150         scenario = self.create_return_name(self.req_d)
 151         code, _ = self.delete(scenario)
 152         self.assertEqual(code, httplib.OK)
 153         code, _ = self.get(scenario)
 154         self.assertEqual(code, httplib.NOT_FOUND)
 155
 156
 157 class TestScenarioUpdate(TestScenarioBase):
 158     def setUp(self):
 159         super(TestScenarioUpdate, self).setUp()
 160         self.scenario = self.create_return_name(self.req_d)
 161         self.scenario_2 = self.create_return_name(self.req_2)
 162         self.update_url = ''
 163         self.scenario_url = '/api/v1/scenarios/{}'.format(self.scenario)
 164         self.installer = self.req_d['installers'][0]['installer']
 165         self.version = self.req_d['installers'][0]['versions'][0]['version']
 166         self.locate_project = 'installer={}&version={}&project={}'.format(
 167             self.installer,
 168             self.version,
 169             'functest')
 170
 171     def update_url_fixture(item):
 172         def _update_url_fixture(xstep):
 173             def wrapper(self, *args, **kwargs):
 174                 locator = None
 175                 if item in ['projects', 'owner']:
 176                     locator = 'installer={}&version={}'.format(
 177                         self.installer,
 178                         self.version)
 179                 self.update_url = '{}/{}?{}'.format(self.scenario_url,
 180                                                     item,
 181                                                     locator)
 182                 xstep(self, *args, **kwargs)
 183             return wrapper
 184         return _update_url_fixture
 185
 186     def update_partial(operate, expected):
 187         def _update_partial(set_update):
 188             @functools.wraps(set_update)
 189             def wrapper(self):
 190                 update, scenario = set_update(self, deepcopy(self.req_d))
 191                 code, body = getattr(self, operate)(update, self.scenario)
 192                 getattr(self, expected)(code, scenario)
 193             return wrapper
 194         return _update_partial
 195
 196     @update_partial('_add', '_success')
 197     def test_addScore(self, scenario):
 198         add = models.ScenarioScore(date=str(datetime.now()), score='11/12')
 199         projects = scenario['installers'][0]['versions'][0]['projects']
 200         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 201         functest['scores'].append(add.format())
 202         self.update_url = '{}/scores?{}'.format(self.scenario_url,
 203                                                 self.locate_project)
 204
 205         return add, scenario
 206
 207     @update_partial('_add', '_success')
 208     def test_addTrustIndicator(self, scenario):
 209         add = models.ScenarioTI(date=str(datetime.now()), status='gold')
 210         projects = scenario['installers'][0]['versions'][0]['projects']
 211         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 212         functest['trust_indicators'].append(add.format())
 213         self.update_url = '{}/trust_indicators?{}'.format(self.scenario_url,
 214                                                           self.locate_project)
 215
 216         return add, scenario
 217
 218     @update_partial('_add', '_success')
 219     def test_addCustoms(self, scenario):
 220         add = ['odl', 'parser', 'vping_ssh']
 221         projects = scenario['installers'][0]['versions'][0]['projects']
 222         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 223         functest['customs'] = list(set(functest['customs'] + add))
 224         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 225                                                  self.locate_project)
 226         return add, scenario
 227
 228     @update_partial('_update', '_success')
 229     def test_updateCustoms(self, scenario):
 230         news = ['odl', 'parser', 'vping_ssh']
 231         projects = scenario['installers'][0]['versions'][0]['projects']
 232         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 233         functest['customs'] = news
 234         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 235                                                  self.locate_project)
 236
 237         return news, scenario
 238
 239     @update_partial('_delete', '_success')
 240     def test_deleteCustoms(self, scenario):
 241         obsoletes = ['vping_ssh']
 242         projects = scenario['installers'][0]['versions'][0]['projects']
 243         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 244         functest['customs'] = ['healthcheck']
 245         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 246                                                  self.locate_project)
 247
 248         return obsoletes, scenario
 249
 250     @update_url_fixture('projects')
 251     @update_partial('_add', '_success')
 252     def test_addProjects_succ(self, scenario):
 253         add = models.ScenarioProject(project='qtip').format()
 254         scenario['installers'][0]['versions'][0]['projects'].append(add)
 255         return [add], scenario
 256
 257     @update_url_fixture('projects')
 258     @update_partial('_add', '_conflict')
 259     def test_addProjects_already_exist(self, scenario):
 260         add = models.ScenarioProject(project='functest').format()
 261         scenario['installers'][0]['versions'][0]['projects'].append(add)
 262         return [add], scenario
 263
 264     @update_url_fixture('projects')
 265     @update_partial('_add', '_bad_request')
 266     def test_addProjects_bad_schema(self, scenario):
 267         add = models.ScenarioProject(project='functest').format()
 268         add['score'] = None
 269         scenario['installers'][0]['versions'][0]['projects'].append(add)
 270         return [add], scenario
 271
 272     @update_url_fixture('projects')
 273     @update_partial('_update', '_success')
 274     def test_updateProjects_succ(self, scenario):
 275         update = models.ScenarioProject(project='qtip').format()
 276         scenario['installers'][0]['versions'][0]['projects'] = [update]
 277         return [update], scenario
 278
 279     @update_url_fixture('projects')
 280     @update_partial('_update', '_bad_request')
 281     def test_updateProjects_bad_schema(self, scenario):
 282         update = models.ScenarioProject(project='functest').format()
 283         update['score'] = None
 284         scenario['installers'][0]['versions'][0]['projects'] = [update]
 285         return [update], scenario
 286
 287     @update_url_fixture('projects')
 288     @update_partial('_delete', '_success')
 289     def test_deleteProjects(self, scenario):
 290         deletes = ['functest']
 291         projects = scenario['installers'][0]['versions'][0]['projects']
 292         scenario['installers'][0]['versions'][0]['projects'] = filter(
 293             lambda f: f['project'] != 'functest',
 294             projects)
 295         return deletes, scenario
 296
 297     @update_url_fixture('owner')
 298     @update_partial('_update', '_success')
 299     def test_changeOwner(self, scenario):
 300         new_owner = 'new_owner'
 301         scenario['installers'][0]['versions'][0]['owner'] = 'www'
 302         return new_owner, scenario
 303
 304     def _add(self, update_req, new_scenario):
 305         return self.post_direct_url(self.update_url, update_req)
 306
 307     def _update(self, update_req, new_scenario):
 308         return self.update_direct_url(self.update_url, update_req)
 309
 310     def _delete(self, update_req, new_scenario):
 311         return self.delete_direct_url(self.update_url, update_req)
 312
 313     def _success(self, status, new_scenario):
 314         self.assertEqual(status, httplib.OK)
 315         self._get_and_assert(new_scenario.get('name'), new_scenario)
 316
 317     def _forbidden(self, status, new_scenario):
 318         self.assertEqual(status, httplib.FORBIDDEN)
 319
 320     def _bad_request(self, status, new_scenario):
 321         self.assertEqual(status, httplib.BAD_REQUEST)
 322
 323     def _conflict(self, status, new_scenario):
 324         self.assertEqual(status, httplib.CONFLICT)