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

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