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

   1 import functools
   2 import httplib
   3 import json
   4 import os
   5
   6 from datetime import datetime
   7
   8 from opnfv_testapi.common import message
   9 import opnfv_testapi.models.scenario_models as models
  10 from opnfv_testapi.tests.unit.handlers 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         self.assertEqual(scenario, models.Scenario.from_dict(req))
  51
  52     @staticmethod
  53     def set_query(*args):
  54         uri = ''
  55         for arg in args:
  56             uri += arg + '&'
  57         return uri[0: -1]
  58
  59     def get_and_assert(self, name):
  60         code, body = self.get(name)
  61         self.assert_res(code, body, self.req_d)
  62
  63
  64 class TestScenarioCreate(TestScenarioBase):
  65     def test_withoutBody(self):
  66         (code, body) = self.create()
  67         self.assertEqual(code, httplib.BAD_REQUEST)
  68
  69     def test_emptyName(self):
  70         req_empty = models.ScenarioCreateRequest('')
  71         (code, body) = self.create(req_empty)
  72         self.assertEqual(code, httplib.BAD_REQUEST)
  73         self.assertIn(message.missing('name'), body)
  74
  75     def test_noneName(self):
  76         req_none = models.ScenarioCreateRequest(None)
  77         (code, body) = self.create(req_none)
  78         self.assertEqual(code, httplib.BAD_REQUEST)
  79         self.assertIn(message.missing('name'), body)
  80
  81     def test_success(self):
  82         (code, body) = self.create_d()
  83         self.assertEqual(code, httplib.OK)
  84         self.assert_create_body(body)
  85
  86     def test_alreadyExist(self):
  87         self.create_d()
  88         (code, body) = self.create_d()
  89         self.assertEqual(code, httplib.FORBIDDEN)
  90         self.assertIn(message.exist_base, body)
  91
  92
  93 class TestScenarioGet(TestScenarioBase):
  94     def setUp(self):
  95         super(TestScenarioGet, self).setUp()
  96         self.scenario_1 = self.create_return_name(self.req_d)
  97         self.scenario_2 = self.create_return_name(self.req_2)
  98
  99     def test_getByName(self):
 100         self.get_and_assert(self.scenario_1)
 101
 102     def test_getAll(self):
 103         self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
 104
 105     def test_queryName(self):
 106         query = self.set_query('name=nosdn-nofeature-ha')
 107         self._query_and_assert(query, reqs=[self.req_d])
 108
 109     def test_queryInstaller(self):
 110         query = self.set_query('installer=apex')
 111         self._query_and_assert(query, reqs=[self.req_d])
 112
 113     def test_queryVersion(self):
 114         query = self.set_query('version=master')
 115         self._query_and_assert(query, reqs=[self.req_d])
 116
 117     def test_queryProject(self):
 118         query = self.set_query('project=functest')
 119         self._query_and_assert(query, reqs=[self.req_d, self.req_2])
 120
 121     # close due to random fail, open again after solve it in another patch
 122     # def test_queryCombination(self):
 123     #     query = self._set_query('name=nosdn-nofeature-ha',
 124     #                             'installer=apex',
 125     #                             'version=master',
 126     #                             'project=functest')
 127     #
 128     #     self._query_and_assert(query, reqs=[self.req_d])
 129
 130     def _query_and_assert(self, query, found=True, reqs=None):
 131         code, body = self.query(query)
 132         if not found:
 133             self.assertEqual(code, httplib.OK)
 134             self.assertEqual(0, len(body.scenarios))
 135         else:
 136             self.assertEqual(len(reqs), len(body.scenarios))
 137             for req in reqs:
 138                 for scenario in body.scenarios:
 139                     if req['name'] == scenario.name:
 140                         self.assert_res(code, scenario, req)
 141
 142
 143 class TestScenarioDelete(TestScenarioBase):
 144     def test_notFound(self):
 145         code, body = self.delete('notFound')
 146         self.assertEqual(code, httplib.NOT_FOUND)
 147
 148     def test_success(self):
 149         scenario = self.create_return_name(self.req_d)
 150         code, _ = self.delete(scenario)
 151         self.assertEqual(code, httplib.OK)
 152         code, _ = self.get(scenario)
 153         self.assertEqual(code, httplib.NOT_FOUND)
 154
 155
 156 class TestScenarioUpdate(TestScenarioBase):
 157     def setUp(self):
 158         super(TestScenarioUpdate, self).setUp()
 159         self.scenario = self.create_return_name(self.req_d)
 160         self.scenario_2 = self.create_return_name(self.req_2)
 161         self.update_url = ''
 162         self.scenario_url = '/api/v1/scenarios/{}'.format(self.scenario)
 163         self.installer = self.req_d['installers'][0]['installer']
 164         self.version = self.req_d['installers'][0]['versions'][0]['version']
 165         self.locate_project = 'installer={}&version={}&project={}'.format(
 166             self.installer,
 167             self.version,
 168             'functest')
 169
 170     def update_url_fixture(item):
 171         def _update_url_fixture(xstep):
 172             def wrapper(self, *args, **kwargs):
 173                 self.update_url = '{}/{}'.format(self.scenario_url, item)
 174                 locator = None
 175                 if item in ['projects', 'owner']:
 176                     locator = 'installer={}&version={}'.format(
 177                         self.installer,
 178                         self.version)
 179                 elif item in ['versions']:
 180                     locator = 'installer={}'.format(
 181                         self.installer)
 182                 elif item in ['rename']:
 183                     self.update_url = self.scenario_url
 184
 185                 if locator:
 186                     self.update_url = '{}?{}'.format(self.update_url, locator)
 187
 188                 xstep(self, *args, **kwargs)
 189             return wrapper
 190         return _update_url_fixture
 191
 192     def update_partial(operate, expected):
 193         def _update_partial(set_update):
 194             @functools.wraps(set_update)
 195             def wrapper(self):
 196                 update = set_update(self)
 197                 code, body = getattr(self, operate)(update)
 198                 getattr(self, expected)(code)
 199             return wrapper
 200         return _update_partial
 201
 202     @update_partial('_add', '_success')
 203     def test_addScore(self):
 204         add = models.ScenarioScore(date=str(datetime.now()), score='11/12')
 205         projects = self.req_d['installers'][0]['versions'][0]['projects']
 206         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 207         functest['scores'].append(add.format())
 208         self.update_url = '{}/scores?{}'.format(self.scenario_url,
 209                                                 self.locate_project)
 210
 211         return add
 212
 213     @update_partial('_add', '_success')
 214     def test_addTrustIndicator(self):
 215         add = models.ScenarioTI(date=str(datetime.now()), status='gold')
 216         projects = self.req_d['installers'][0]['versions'][0]['projects']
 217         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 218         functest['trust_indicators'].append(add.format())
 219         self.update_url = '{}/trust_indicators?{}'.format(self.scenario_url,
 220                                                           self.locate_project)
 221
 222         return add
 223
 224     @update_partial('_add', '_success')
 225     def test_addCustoms(self):
 226         adds = ['odl', 'parser', 'vping_ssh']
 227         projects = self.req_d['installers'][0]['versions'][0]['projects']
 228         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 229         functest['customs'] = list(set(functest['customs'] + adds))
 230         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 231                                                  self.locate_project)
 232         return adds
 233
 234     @update_partial('_update', '_success')
 235     def test_updateCustoms(self):
 236         updates = ['odl', 'parser', 'vping_ssh']
 237         projects = self.req_d['installers'][0]['versions'][0]['projects']
 238         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 239         functest['customs'] = updates
 240         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 241                                                  self.locate_project)
 242
 243         return updates
 244
 245     @update_partial('_delete', '_success')
 246     def test_deleteCustoms(self):
 247         deletes = ['vping_ssh']
 248         projects = self.req_d['installers'][0]['versions'][0]['projects']
 249         functest = filter(lambda f: f['project'] == 'functest', projects)[0]
 250         functest['customs'] = ['healthcheck']
 251         self.update_url = '{}/customs?{}'.format(self.scenario_url,
 252                                                  self.locate_project)
 253
 254         return deletes
 255
 256     @update_url_fixture('projects')
 257     @update_partial('_add', '_success')
 258     def test_addProjects_succ(self):
 259         add = models.ScenarioProject(project='qtip').format()
 260         self.req_d['installers'][0]['versions'][0]['projects'].append(add)
 261         return [add]
 262
 263     @update_url_fixture('projects')
 264     @update_partial('_add', '_conflict')
 265     def test_addProjects_already_exist(self):
 266         add = models.ScenarioProject(project='functest').format()
 267         return [add]
 268
 269     @update_url_fixture('projects')
 270     @update_partial('_add', '_bad_request')
 271     def test_addProjects_bad_schema(self):
 272         add = models.ScenarioProject(project='functest').format()
 273         add['score'] = None
 274         return [add]
 275
 276     @update_url_fixture('projects')
 277     @update_partial('_update', '_success')
 278     def test_updateProjects_succ(self):
 279         update = models.ScenarioProject(project='qtip').format()
 280         self.req_d['installers'][0]['versions'][0]['projects'] = [update]
 281         return [update]
 282
 283     @update_url_fixture('projects')
 284     @update_partial('_update', '_conflict')
 285     def test_updateProjects_duplicated(self):
 286         update = models.ScenarioProject(project='qtip').format()
 287         return [update, update]
 288
 289     @update_url_fixture('projects')
 290     @update_partial('_update', '_bad_request')
 291     def test_updateProjects_bad_schema(self):
 292         update = models.ScenarioProject(project='functest').format()
 293         update['score'] = None
 294         return [update]
 295
 296     @update_url_fixture('projects')
 297     @update_partial('_delete', '_success')
 298     def test_deleteProjects(self):
 299         deletes = ['functest']
 300         projects = self.req_d['installers'][0]['versions'][0]['projects']
 301         self.req_d['installers'][0]['versions'][0]['projects'] = filter(
 302             lambda f: f['project'] != 'functest',
 303             projects)
 304         return deletes
 305
 306     @update_url_fixture('owner')
 307     @update_partial('_update', '_success')
 308     def test_changeOwner(self):
 309         new_owner = 'new_owner'
 310         update = models.ScenarioChangeOwnerRequest(new_owner).format()
 311         self.req_d['installers'][0]['versions'][0]['owner'] = new_owner
 312         return update
 313
 314     @update_url_fixture('versions')
 315     @update_partial('_add', '_success')
 316     def test_addVersions_succ(self):
 317         add = models.ScenarioVersion(version='Euphrates').format()
 318         self.req_d['installers'][0]['versions'].append(add)
 319         return [add]
 320
 321     @update_url_fixture('versions')
 322     @update_partial('_add', '_conflict')
 323     def test_addVersions_already_exist(self):
 324         add = models.ScenarioVersion(version='master').format()
 325         return [add]
 326
 327     @update_url_fixture('versions')
 328     @update_partial('_add', '_bad_request')
 329     def test_addVersions_bad_schema(self):
 330         add = models.ScenarioVersion(version='euphrates').format()
 331         add['notexist'] = None
 332         return [add]
 333
 334     @update_url_fixture('versions')
 335     @update_partial('_update', '_success')
 336     def test_updateVersions_succ(self):
 337         update = models.ScenarioVersion(version='euphrates').format()
 338         self.req_d['installers'][0]['versions'] = [update]
 339         return [update]
 340
 341     @update_url_fixture('versions')
 342     @update_partial('_update', '_conflict')
 343     def test_updateVersions_duplicated(self):
 344         update = models.ScenarioVersion(version='euphrates').format()
 345         return [update, update]
 346
 347     @update_url_fixture('versions')
 348     @update_partial('_update', '_bad_request')
 349     def test_updateVersions_bad_schema(self):
 350         update = models.ScenarioVersion(version='euphrates').format()
 351         update['not_owner'] = 'Iam'
 352         return [update]
 353
 354     @update_url_fixture('versions')
 355     @update_partial('_delete', '_success')
 356     def test_deleteVersions(self):
 357         deletes = ['master']
 358         versions = self.req_d['installers'][0]['versions']
 359         self.req_d['installers'][0]['versions'] = filter(
 360             lambda f: f['version'] != 'master',
 361             versions)
 362         return deletes
 363
 364     @update_url_fixture('installers')
 365     @update_partial('_add', '_success')
 366     def test_addInstallers_succ(self):
 367         add = models.ScenarioInstaller(installer='daisy').format()
 368         self.req_d['installers'].append(add)
 369         return [add]
 370
 371     @update_url_fixture('installers')
 372     @update_partial('_add', '_conflict')
 373     def test_addInstallers_already_exist(self):
 374         add = models.ScenarioInstaller(installer='apex').format()
 375         return [add]
 376
 377     @update_url_fixture('installers')
 378     @update_partial('_add', '_bad_request')
 379     def test_addInstallers_bad_schema(self):
 380         add = models.ScenarioInstaller(installer='daisy').format()
 381         add['not_exist'] = 'not_exist'
 382         return [add]
 383
 384     @update_url_fixture('installers')
 385     @update_partial('_update', '_success')
 386     def test_updateInstallers_succ(self):
 387         update = models.ScenarioInstaller(installer='daisy').format()
 388         self.req_d['installers'] = [update]
 389         return [update]
 390
 391     @update_url_fixture('installers')
 392     @update_partial('_update', '_conflict')
 393     def test_updateInstallers_duplicated(self):
 394         update = models.ScenarioInstaller(installer='daisy').format()
 395         return [update, update]
 396
 397     @update_url_fixture('installers')
 398     @update_partial('_update', '_bad_request')
 399     def test_updateInstallers_bad_schema(self):
 400         update = models.ScenarioInstaller(installer='daisy').format()
 401         update['not_exist'] = 'not_exist'
 402         return [update]
 403
 404     @update_url_fixture('installers')
 405     @update_partial('_delete', '_success')
 406     def test_deleteInstallers(self):
 407         deletes = ['apex']
 408         installers = self.req_d['installers']
 409         self.req_d['installers'] = filter(
 410             lambda f: f['installer'] != 'apex',
 411             installers)
 412         return deletes
 413
 414     @update_url_fixture('rename')
 415     @update_partial('_update', '_success')
 416     def test_renameScenario(self):
 417         new_name = 'new_scenario_name'
 418         update = models.ScenarioUpdateRequest(name=new_name)
 419         self.req_d['name'] = new_name
 420         return update
 421
 422     @update_url_fixture('rename')
 423     @update_partial('_update', '_forbidden')
 424     def test_renameScenario_exist(self):
 425         new_name = self.req_d['name']
 426         update = models.ScenarioUpdateRequest(name=new_name)
 427         return update
 428
 429     def _add(self, update_req):
 430         return self.post_direct_url(self.update_url, update_req)
 431
 432     def _update(self, update_req):
 433         return self.update_direct_url(self.update_url, update_req)
 434
 435     def _delete(self, update_req):
 436         return self.delete_direct_url(self.update_url, update_req)
 437
 438     def _success(self, status):
 439         self.assertEqual(status, httplib.OK)
 440         self.get_and_assert(self.req_d['name'])
 441
 442     def _forbidden(self, status):
 443         self.assertEqual(status, httplib.FORBIDDEN)
 444
 445     def _bad_request(self, status):
 446         self.assertEqual(status, httplib.BAD_REQUEST)
 447
 448     def _conflict(self, status):
 449         self.assertEqual(status, httplib.CONFLICT)