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

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