1 from copy import deepcopy
2 from datetime import datetime
6 from opnfv_testapi.common import constants
7 import opnfv_testapi.resources.scenario_models as models
8 import test_base as base
11 class TestScenarioBase(base.TestBase):
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')
23 def assert_body(self, project, req=None):
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:
34 def create_return_name(self, req):
35 _, res = self.create(req)
36 return res.href.split('/')[-1]
38 def assert_res(self, code, scenario, req=None):
39 self.assertEqual(code, constants.HTTP_OK)
42 self.assertIsNotNone(scenario._id)
43 self.assertIsNotNone(scenario.creation_date)
45 scenario == models.Scenario.from_dict(req)
48 def _set_query(*args):
54 def _get_and_assert(self, name, req=None):
55 code, body = self.get(name)
56 self.assert_res(code, body, req)
59 class TestScenarioCreate(TestScenarioBase):
60 def test_withoutBody(self):
61 (code, body) = self.create()
62 self.assertEqual(code, constants.HTTP_BAD_REQUEST)
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)
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)
76 def test_success(self):
77 (code, body) = self.create_d()
78 self.assertEqual(code, constants.HTTP_OK)
79 self.assert_create_body(body)
81 def test_alreadyExist(self):
83 (code, body) = self.create_d()
84 self.assertEqual(code, constants.HTTP_FORBIDDEN)
85 self.assertIn('already exists', body)
88 class TestScenarioGet(TestScenarioBase):
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)
94 def test_getByName(self):
95 self._get_and_assert(self.scenario_1, self.req_d)
97 def test_getAll(self):
98 self._query_and_assert(query=None, reqs=[self.req_d, self.req_2])
100 def test_queryName(self):
101 query = self._set_query('name=nosdn-nofeature-ha')
102 self._query_and_assert(query, reqs=[self.req_d])
104 def test_queryInstaller(self):
105 query = self._set_query('installer=apex')
106 self._query_and_assert(query, reqs=[self.req_d])
108 def test_queryVersion(self):
109 query = self._set_query('version=master')
110 self._query_and_assert(query, reqs=[self.req_d])
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])
116 def test_queryCombination(self):
117 query = self._set_query('name=nosdn-nofeature-ha',
122 self._query_and_assert(query, reqs=[self.req_d])
124 def _query_and_assert(self, query, found=True, reqs=None):
125 code, body = self.query(query)
127 self.assertEqual(code, constants.HTTP_OK)
128 self.assertEqual(0, len(body.scenarios))
130 self.assertEqual(len(reqs), len(body.scenarios))
132 for scenario in body.scenarios:
133 if req['name'] == scenario.name:
134 self.assert_res(code, scenario, req)
137 class TestScenarioUpdate(TestScenarioBase):
139 super(TestScenarioUpdate, self).setUp()
140 self.scenario = self.create_return_name(self.req_d)
142 def _execute(set_update):
144 update, scenario = set_update(self, deepcopy(self.req_d))
145 self._update_and_assert(update, scenario)
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',
155 term={'name': new_name})
156 self._update_and_assert(update_req, new_scenario, new_name)
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',
166 return update, scenario
169 def test_deleteInstaller(self, scenario):
170 scenario['installers'] = filter(lambda f: f['installer'] != 'apex',
171 scenario['installers'])
173 update = models.ScenarioUpdateRequest(field='installer',
175 locate={'installer': 'apex'})
176 return update, scenario
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',
184 locate={'installer': 'apex'},
186 return update, scenario
189 def test_deleteVersion(self, scenario):
190 scenario['installers'][0]['versions'] = filter(
191 lambda f: f['version'] != 'master',
192 scenario['installers'][0]['versions'])
194 update = models.ScenarioUpdateRequest(field='version',
196 locate={'installer': 'apex',
197 'version': 'master'})
198 return update, scenario
201 def test_changeOwner(self, scenario):
202 scenario['installers'][0]['versions'][0]['owner'] = 'lucy'
204 update = models.ScenarioUpdateRequest(field='owner',
206 locate={'installer': 'apex',
207 'version': 'master'},
208 term={'owner': 'lucy'})
209 return update, scenario
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',
217 locate={'installer': 'apex',
218 'version': 'master'},
220 return update, scenario
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'])
228 update = models.ScenarioUpdateRequest(field='project',
233 'project': 'functest'})
234 return update, scenario
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',
247 'project': 'functest'},
249 return update, scenario
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',
261 'project': 'functest'},
263 return update, scenario
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',
276 'project': 'functest'},
278 return update, scenario
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',
291 'project': 'functest'},
293 return update, scenario
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),
302 def _none_default(check, default):
303 return check if check else default
306 class TestScenarioDelete(TestScenarioBase):
307 def test_notFound(self):
308 code, body = self.delete('notFound')
309 self.assertEqual(code, constants.HTTP_NOT_FOUND)
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)