5 from copy import deepcopy
6 from datetime import datetime
8 import opnfv_testapi.resources.scenario_models as models
9 from opnfv_testapi.common import message
10 from opnfv_testapi.tests.unit.resources import test_base as base
13 class TestScenarioBase(base.TestBase):
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')
25 def assert_body(self, project, req=None):
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:
36 def create_return_name(self, req):
37 _, res = self.create(req)
38 return res.href.split('/')[-1]
40 def assert_res(self, code, scenario, req=None):
41 self.assertEqual(code, httplib.OK)
44 self.assertIsNotNone(scenario._id)
45 self.assertIsNotNone(scenario.creation_date)
47 scenario == models.Scenario.from_dict(req)
50 def _set_query(*args):
56 def _get_and_assert(self, name, req=None):
57 code, body = self.get(name)
58 self.assert_res(code, body, req)
61 class TestScenarioCreate(TestScenarioBase):
62 def test_withoutBody(self):
63 (code, body) = self.create()
64 self.assertEqual(code, httplib.BAD_REQUEST)
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)
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)
78 def test_success(self):
79 (code, body) = self.create_d()
80 self.assertEqual(code, httplib.OK)
81 self.assert_create_body(body)
83 def test_alreadyExist(self):
85 (code, body) = self.create_d()
86 self.assertEqual(code, httplib.FORBIDDEN)
87 self.assertIn(message.exist_base, body)
90 class TestScenarioGet(TestScenarioBase):
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)
96 def test_getByName(self):
97 self._get_and_assert(self.scenario_1, self.req_d)
99 def test_getAll(self):
100 self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
102 def test_queryName(self):
103 query = self._set_query('name=nosdn-nofeature-ha')
104 self._query_and_assert(query, reqs=[self.req_d])
106 def test_queryInstaller(self):
107 query = self._set_query('installer=apex')
108 self._query_and_assert(query, reqs=[self.req_d])
110 def test_queryVersion(self):
111 query = self._set_query('version=master')
112 self._query_and_assert(query, reqs=[self.req_d])
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])
118 def test_queryCombination(self):
119 query = self._set_query('name=nosdn-nofeature-ha',
124 self._query_and_assert(query, reqs=[self.req_d])
126 def _query_and_assert(self, query, found=True, reqs=None):
127 code, body = self.query(query)
129 self.assertEqual(code, httplib.OK)
130 self.assertEqual(0, len(body.scenarios))
132 self.assertEqual(len(reqs), len(body.scenarios))
134 for scenario in body.scenarios:
135 if req['name'] == scenario.name:
136 self.assert_res(code, scenario, req)
139 class TestScenarioUpdate(TestScenarioBase):
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)
145 def _execute(set_update):
146 @functools.wraps(set_update)
148 update, scenario = set_update(self, deepcopy(self.req_d))
149 self._update_and_assert(update, scenario)
152 def _update(expected):
153 def _update(set_update):
154 @functools.wraps(set_update)
156 update, scenario = set_update(self, deepcopy(self.req_d))
157 code, body = self.update(update, self.scenario)
158 getattr(self, expected)(code, scenario)
163 def test_renameScenario(self, scenario):
164 new_name = 'nosdn-nofeature-noha'
165 scenario['name'] = new_name
166 update_req = models.ScenarioUpdateRequest(field='name',
169 term={'name': new_name})
170 return update_req, scenario
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',
179 term={'name': new_name})
180 return update_req, scenario
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',
190 return update_req, scenario
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',
200 return update, scenario
203 def test_deleteInstaller(self, scenario):
204 scenario['installers'] = filter(lambda f: f['installer'] != 'apex',
205 scenario['installers'])
207 update = models.ScenarioUpdateRequest(field='installer',
209 locate={'installer': 'apex'})
210 return update, scenario
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',
218 locate={'installer': 'apex'},
220 return update, scenario
223 def test_deleteVersion(self, scenario):
224 scenario['installers'][0]['versions'] = filter(
225 lambda f: f['version'] != 'master',
226 scenario['installers'][0]['versions'])
228 update = models.ScenarioUpdateRequest(field='version',
230 locate={'installer': 'apex',
231 'version': 'master'})
232 return update, scenario
235 def test_changeOwner(self, scenario):
236 scenario['installers'][0]['versions'][0]['owner'] = 'lucy'
238 update = models.ScenarioUpdateRequest(field='owner',
240 locate={'installer': 'apex',
241 'version': 'master'},
242 term={'owner': 'lucy'})
243 return update, scenario
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',
251 locate={'installer': 'apex',
252 'version': 'master'},
254 return update, scenario
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'])
262 update = models.ScenarioUpdateRequest(field='project',
267 'project': 'functest'})
268 return update, scenario
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',
281 'project': 'functest'},
283 return update, scenario
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',
295 'project': 'functest'},
297 return update, scenario
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',
310 'project': 'functest'},
312 return update, scenario
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',
325 'project': 'functest'},
327 return update, scenario
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),
335 def _success(self, status, new_scenario):
336 self.assertEqual(status, httplib.OK)
337 self._get_and_assert(new_scenario.get('name'), new_scenario)
339 def _forbidden(self, status, new_scenario):
340 self.assertEqual(status, httplib.FORBIDDEN)
342 def _bad_request(self, status, new_scenario):
343 self.assertEqual(status, httplib.BAD_REQUEST)
346 class TestScenarioDelete(TestScenarioBase):
347 def test_notFound(self):
348 code, body = self.delete('notFound')
349 self.assertEqual(code, httplib.NOT_FOUND)
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)
359 def _none_default(check, default):
360 return check if check else default