5 from copy import deepcopy
6 from datetime import datetime
8 from opnfv_testapi.common import message
9 import opnfv_testapi.resources.scenario_models as models
10 from opnfv_testapi.tests.unit.resources 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)
51 scenario == models.Scenario.from_dict(req)
54 def _set_query(*args):
60 def _get_and_assert(self, name, req=None):
61 code, body = self.get(name)
62 self.assert_res(code, body, req)
65 class TestScenarioCreate(TestScenarioBase):
66 def test_withoutBody(self):
67 (code, body) = self.create()
68 self.assertEqual(code, httplib.BAD_REQUEST)
70 def test_emptyName(self):
71 req_empty = models.ScenarioCreateRequest('')
72 (code, body) = self.create(req_empty)
73 self.assertEqual(code, httplib.BAD_REQUEST)
74 self.assertIn(message.missing('name'), body)
76 def test_noneName(self):
77 req_none = models.ScenarioCreateRequest(None)
78 (code, body) = self.create(req_none)
79 self.assertEqual(code, httplib.BAD_REQUEST)
80 self.assertIn(message.missing('name'), body)
82 def test_success(self):
83 (code, body) = self.create_d()
84 self.assertEqual(code, httplib.OK)
85 self.assert_create_body(body)
87 def test_alreadyExist(self):
89 (code, body) = self.create_d()
90 self.assertEqual(code, httplib.FORBIDDEN)
91 self.assertIn(message.exist_base, body)
94 class TestScenarioGet(TestScenarioBase):
96 super(TestScenarioGet, self).setUp()
97 self.scenario_1 = self.create_return_name(self.req_d)
98 self.scenario_2 = self.create_return_name(self.req_2)
100 def test_getByName(self):
101 self._get_and_assert(self.scenario_1, self.req_d)
103 def test_getAll(self):
104 self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
106 def test_queryName(self):
107 query = self._set_query('name=nosdn-nofeature-ha')
108 self._query_and_assert(query, reqs=[self.req_d])
110 def test_queryInstaller(self):
111 query = self._set_query('installer=apex')
112 self._query_and_assert(query, reqs=[self.req_d])
114 def test_queryVersion(self):
115 query = self._set_query('version=master')
116 self._query_and_assert(query, reqs=[self.req_d])
118 def test_queryProject(self):
119 query = self._set_query('project=functest')
120 self._query_and_assert(query, reqs=[self.req_d, self.req_2])
122 # close due to random fail, open again after solve it in another patch
123 # def test_queryCombination(self):
124 # query = self._set_query('name=nosdn-nofeature-ha',
127 # 'project=functest')
129 # self._query_and_assert(query, reqs=[self.req_d])
131 def _query_and_assert(self, query, found=True, reqs=None):
132 code, body = self.query(query)
134 self.assertEqual(code, httplib.OK)
135 self.assertEqual(0, len(body.scenarios))
137 self.assertEqual(len(reqs), len(body.scenarios))
139 for scenario in body.scenarios:
140 if req['name'] == scenario.name:
141 self.assert_res(code, scenario, req)
144 class TestScenarioDelete(TestScenarioBase):
145 def test_notFound(self):
146 code, body = self.delete('notFound')
147 self.assertEqual(code, httplib.NOT_FOUND)
149 def test_success(self):
150 scenario = self.create_return_name(self.req_d)
151 code, _ = self.delete(scenario)
152 self.assertEqual(code, httplib.OK)
153 code, _ = self.get(scenario)
154 self.assertEqual(code, httplib.NOT_FOUND)
157 class TestScenarioUpdate(TestScenarioBase):
159 super(TestScenarioUpdate, self).setUp()
160 self.scenario = self.create_return_name(self.req_d)
161 self.scenario_2 = self.create_return_name(self.req_2)
163 self.scenario_url = '/api/v1/scenarios/{}'.format(self.scenario)
164 self.installer = self.req_d['installers'][0]['installer']
165 self.version = self.req_d['installers'][0]['versions'][0]['version']
166 self.locate_project = 'installer={}&version={}&project={}'.format(
171 def update_url_fixture(item):
172 def _update_url_fixture(xstep):
173 def wrapper(self, *args, **kwargs):
175 if item in ['projects', 'owner']:
176 locator = 'installer={}&version={}'.format(
179 self.update_url = '{}/{}?{}'.format(self.scenario_url,
182 xstep(self, *args, **kwargs)
184 return _update_url_fixture
186 def update_partial(operate, expected):
187 def _update_partial(set_update):
188 @functools.wraps(set_update)
190 update, scenario = set_update(self, deepcopy(self.req_d))
191 code, body = getattr(self, operate)(update, self.scenario)
192 getattr(self, expected)(code, scenario)
194 return _update_partial
196 @update_partial('_add', '_success')
197 def test_addScore(self, scenario):
198 add = models.ScenarioScore(date=str(datetime.now()), score='11/12')
199 projects = scenario['installers'][0]['versions'][0]['projects']
200 functest = filter(lambda f: f['project'] == 'functest', projects)[0]
201 functest['scores'].append(add.format())
202 self.update_url = '{}/scores?{}'.format(self.scenario_url,
207 @update_partial('_add', '_success')
208 def test_addTrustIndicator(self, scenario):
209 add = models.ScenarioTI(date=str(datetime.now()), status='gold')
210 projects = scenario['installers'][0]['versions'][0]['projects']
211 functest = filter(lambda f: f['project'] == 'functest', projects)[0]
212 functest['trust_indicators'].append(add.format())
213 self.update_url = '{}/trust_indicators?{}'.format(self.scenario_url,
218 @update_partial('_add', '_success')
219 def test_addCustoms(self, scenario):
220 add = ['odl', 'parser', 'vping_ssh']
221 projects = scenario['installers'][0]['versions'][0]['projects']
222 functest = filter(lambda f: f['project'] == 'functest', projects)[0]
223 functest['customs'] = list(set(functest['customs'] + add))
224 self.update_url = '{}/customs?{}'.format(self.scenario_url,
228 @update_partial('_update', '_success')
229 def test_updateCustoms(self, scenario):
230 news = ['odl', 'parser', 'vping_ssh']
231 projects = scenario['installers'][0]['versions'][0]['projects']
232 functest = filter(lambda f: f['project'] == 'functest', projects)[0]
233 functest['customs'] = news
234 self.update_url = '{}/customs?{}'.format(self.scenario_url,
237 return news, scenario
239 @update_partial('_delete', '_success')
240 def test_deleteCustoms(self, scenario):
241 obsoletes = ['vping_ssh']
242 projects = scenario['installers'][0]['versions'][0]['projects']
243 functest = filter(lambda f: f['project'] == 'functest', projects)[0]
244 functest['customs'] = ['healthcheck']
245 self.update_url = '{}/customs?{}'.format(self.scenario_url,
248 return obsoletes, scenario
250 @update_url_fixture('projects')
251 @update_partial('_add', '_success')
252 def test_addProjects_succ(self, scenario):
253 add = models.ScenarioProject(project='qtip').format()
254 scenario['installers'][0]['versions'][0]['projects'].append(add)
255 return [add], scenario
257 @update_url_fixture('projects')
258 @update_partial('_add', '_conflict')
259 def test_addProjects_already_exist(self, scenario):
260 add = models.ScenarioProject(project='functest').format()
261 scenario['installers'][0]['versions'][0]['projects'].append(add)
262 return [add], scenario
264 @update_url_fixture('projects')
265 @update_partial('_add', '_bad_request')
266 def test_addProjects_bad_schema(self, scenario):
267 add = models.ScenarioProject(project='functest').format()
269 scenario['installers'][0]['versions'][0]['projects'].append(add)
270 return [add], scenario
272 @update_url_fixture('projects')
273 @update_partial('_update', '_success')
274 def test_updateProjects_succ(self, scenario):
275 update = models.ScenarioProject(project='qtip').format()
276 scenario['installers'][0]['versions'][0]['projects'] = [update]
277 return [update], scenario
279 @update_url_fixture('projects')
280 @update_partial('_update', '_bad_request')
281 def test_updateProjects_bad_schema(self, scenario):
282 update = models.ScenarioProject(project='functest').format()
283 update['score'] = None
284 scenario['installers'][0]['versions'][0]['projects'] = [update]
285 return [update], scenario
287 @update_url_fixture('projects')
288 @update_partial('_delete', '_success')
289 def test_deleteProjects(self, scenario):
290 deletes = ['functest']
291 projects = scenario['installers'][0]['versions'][0]['projects']
292 scenario['installers'][0]['versions'][0]['projects'] = filter(
293 lambda f: f['project'] != 'functest',
295 return deletes, scenario
297 @update_url_fixture('owner')
298 @update_partial('_update', '_success')
299 def test_changeOwner(self, scenario):
300 new_owner = 'new_owner'
301 scenario['installers'][0]['versions'][0]['owner'] = 'www'
302 return new_owner, scenario
304 def _add(self, update_req, new_scenario):
305 return self.post_direct_url(self.update_url, update_req)
307 def _update(self, update_req, new_scenario):
308 return self.update_direct_url(self.update_url, update_req)
310 def _delete(self, update_req, new_scenario):
311 return self.delete_direct_url(self.update_url, update_req)
313 def _success(self, status, new_scenario):
314 self.assertEqual(status, httplib.OK)
315 self._get_and_assert(new_scenario.get('name'), new_scenario)
317 def _forbidden(self, status, new_scenario):
318 self.assertEqual(status, httplib.FORBIDDEN)
320 def _bad_request(self, status, new_scenario):
321 self.assertEqual(status, httplib.BAD_REQUEST)
323 def _conflict(self, status, new_scenario):
324 self.assertEqual(status, httplib.CONFLICT)