6 from datetime import datetime
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
13 def _none_default(check, default):
14 return check if check else default
17 class TestScenarioBase(base.TestBase):
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')
29 def assert_body(self, project, req=None):
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:
40 def create_return_name(self, req):
41 _, res = self.create(req)
42 return res.href.split('/')[-1]
44 def assert_res(self, code, scenario, req=None):
45 self.assertEqual(code, httplib.OK)
48 self.assertIsNotNone(scenario._id)
49 self.assertIsNotNone(scenario.creation_date)
50 self.assertEqual(scenario, models.Scenario.from_dict(req))
59 def get_and_assert(self, name):
60 code, body = self.get(name)
61 self.assert_res(code, body, self.req_d)
64 class TestScenarioCreate(TestScenarioBase):
65 def test_withoutBody(self):
66 (code, body) = self.create()
67 self.assertEqual(code, httplib.BAD_REQUEST)
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)
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)
81 def test_success(self):
82 (code, body) = self.create_d()
83 self.assertEqual(code, httplib.OK)
84 self.assert_create_body(body)
86 def test_alreadyExist(self):
88 (code, body) = self.create_d()
89 self.assertEqual(code, httplib.FORBIDDEN)
90 self.assertIn(message.exist_base, body)
93 class TestScenarioGet(TestScenarioBase):
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)
99 def test_getByName(self):
100 self.get_and_assert(self.scenario_1)
102 def test_getAll(self):
103 self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
105 def test_queryName(self):
106 query = self.set_query('name=nosdn-nofeature-ha')
107 self._query_and_assert(query, reqs=[self.req_d])
109 def test_queryInstaller(self):
110 query = self.set_query('installer=apex')
111 self._query_and_assert(query, reqs=[self.req_d])
113 def test_queryVersion(self):
114 query = self.set_query('version=master')
115 self._query_and_assert(query, reqs=[self.req_d])
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])
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',
126 # 'project=functest')
128 # self._query_and_assert(query, reqs=[self.req_d])
130 def _query_and_assert(self, query, found=True, reqs=None):
131 code, body = self.query(query)
133 self.assertEqual(code, httplib.OK)
134 self.assertEqual(0, len(body.scenarios))
136 self.assertEqual(len(reqs), len(body.scenarios))
138 for scenario in body.scenarios:
139 if req['name'] == scenario.name:
140 self.assert_res(code, scenario, req)
143 class TestScenarioDelete(TestScenarioBase):
144 def test_notFound(self):
145 code, body = self.delete('notFound')
146 self.assertEqual(code, httplib.NOT_FOUND)
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)
156 class TestScenarioUpdate(TestScenarioBase):
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)
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(
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)
175 if item in ['projects', 'owner']:
176 locator = 'installer={}&version={}'.format(
179 elif item in ['versions']:
180 locator = 'installer={}'.format(
182 elif item in ['rename']:
183 self.update_url = self.scenario_url
186 self.update_url = '{}?{}'.format(self.update_url, locator)
188 xstep(self, *args, **kwargs)
190 return _update_url_fixture
192 def update_partial(operate, expected):
193 def _update_partial(set_update):
194 @functools.wraps(set_update)
196 update = set_update(self)
197 code, body = getattr(self, operate)(update)
198 getattr(self, expected)(code)
200 return _update_partial
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,
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,
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,
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,
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,
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)
263 @update_url_fixture('projects')
264 @update_partial('_add', '_conflict')
265 def test_addProjects_already_exist(self):
266 add = models.ScenarioProject(project='functest').format()
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()
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]
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]
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
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',
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
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)
321 @update_url_fixture('versions')
322 @update_partial('_add', '_conflict')
323 def test_addVersions_already_exist(self):
324 add = models.ScenarioVersion(version='master').format()
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
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]
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]
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'
354 @update_url_fixture('versions')
355 @update_partial('_delete', '_success')
356 def test_deleteVersions(self):
358 versions = self.req_d['installers'][0]['versions']
359 self.req_d['installers'][0]['versions'] = filter(
360 lambda f: f['version'] != 'master',
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)
371 @update_url_fixture('installers')
372 @update_partial('_add', '_conflict')
373 def test_addInstallers_already_exist(self):
374 add = models.ScenarioInstaller(installer='apex').format()
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'
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]
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]
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'
404 @update_url_fixture('installers')
405 @update_partial('_delete', '_success')
406 def test_deleteInstallers(self):
408 installers = self.req_d['installers']
409 self.req_d['installers'] = filter(
410 lambda f: f['installer'] != 'apex',
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
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)
429 def _add(self, update_req):
430 return self.post_direct_url(self.update_url, update_req)
432 def _update(self, update_req):
433 return self.update_direct_url(self.update_url, update_req)
435 def _delete(self, update_req):
436 return self.delete_direct_url(self.update_url, update_req)
438 def _success(self, status):
439 self.assertEqual(status, httplib.OK)
440 self.get_and_assert(self.req_d['name'])
442 def _forbidden(self, status):
443 self.assertEqual(status, httplib.FORBIDDEN)
445 def _bad_request(self, status):
446 self.assertEqual(status, httplib.BAD_REQUEST)
448 def _conflict(self, status):
449 self.assertEqual(status, httplib.CONFLICT)