3 # All rights reserved. This program and the accompanying materials
4 # are made available under the terms of the Apache License, Version 2.0
5 # which accompanies this distribution, and is available at
6 # http://www.apache.org/licenses/LICENSE-2.0
12 from tackerclient.v1_0 import client as tackerclient
14 from functest.utils import openstack_tacker
15 from functest.tests.unit import test_utils
18 class OSTackerTesting(unittest.TestCase):
21 self.tacker_client = mock.Mock()
22 self.getresponse = {'vnfds': [{'id': 'test_id'}],
23 'vnfs': [{'id': 'test_id'}],
24 'sfcs': [{'id': 'test_id'}]}
25 self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]}
26 self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]}
27 self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]}
28 self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'},
29 {'id': 'test_sfc_cl2'}]}
31 self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}}
32 self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}}
33 self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}}
34 self.createsfc_clf = {"sfc_classifier": {"attributes":
35 {"sfc_clf": 'sfc_clf_body'}}}
37 self.resource_type = 'vnfd'
38 self.resource_name = 'resource_name'
39 self.tosca_file = 'test_tosca_file'
40 self.vnfd = 'test_vnfd'
43 self.sfc_clf = 'test_sfc_clf'
47 'OS_USERNAME': 'username',
48 'OS_PASSWORD': 'password',
49 'OS_AUTH_URL': 'auth_url',
50 'OS_TENANT_NAME': 'tenant_name',
51 'OS_USER_DOMAIN_NAME': 'user_domain_name',
52 'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
53 'OS_PROJECT_NAME': 'project_name',
54 'OS_ENDPOINT_TYPE': 'endpoint_type',
55 'OS_REGION_NAME': 'region_name'
59 def test_get_tacker_client(self):
60 with mock.patch('functest.utils.openstack_tacker.'
61 'os_utils.get_session'):
62 tackerclient.Client = mock.Mock
63 ret = openstack_tacker.get_tacker_client()
64 self.assertTrue(isinstance(ret, mock.Mock))
66 def test_get_id_from_name(self):
67 with mock.patch.object(self.tacker_client, 'get',
68 return_value=self.getresponse):
69 resp = openstack_tacker.get_id_from_name(self.tacker_client,
72 self.assertEqual(resp, 'test_id')
74 @mock.patch('functest.utils.openstack_tacker.logger.error')
75 def test_get_id_from_name_exception(self, mock_logger_error):
76 with mock.patch.object(self.tacker_client, 'get',
77 side_effect=Exception):
78 resp = openstack_tacker.get_id_from_name(self.tacker_client,
81 self.assertIsNone(resp)
82 mock_logger_error.assert_called_once_with(test_utils.
83 SubstrMatch("Error [get"
92 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
93 def test_get_vnfd_id(self, mock_get_id_from_name):
94 openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name)
95 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
99 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
100 def test_get_vnf_id(self, mock_get_id_from_name):
101 openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name)
102 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
106 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
107 def test_get_sfc_id(self, mock_get_id_from_name):
108 openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name)
109 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
113 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
114 def test_get_sfc_classifier_id(self, mock_get_id_from_name):
115 openstack_tacker.get_sfc_classifier_id(self.tacker_client,
117 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
121 def test_list_vnfds(self):
122 with mock.patch.object(self.tacker_client, 'list_vnfds',
123 return_value=self.vnfdlist):
124 resp = openstack_tacker.list_vnfds(self.tacker_client,
126 self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2'])
128 def test_list_vnfds_verbose(self):
129 with mock.patch.object(self.tacker_client, 'list_vnfds',
130 return_value=self.vnfdlist):
131 resp = openstack_tacker.list_vnfds(self.tacker_client,
133 self.assertEqual(resp, self.vnfdlist)
135 @mock.patch('functest.utils.openstack_tacker.logger.error')
136 def test_list_vnfds_exception(self, mock_logger_error):
137 with mock.patch.object(self.tacker_client, 'list_vnfds',
138 side_effect=Exception):
139 resp = openstack_tacker.list_vnfds(self.tacker_client,
141 mock_logger_error.assert_called_once_with(test_utils.
147 self.assertIsNone(resp)
149 def test_create_vnfd_missing_file(self):
150 with mock.patch.object(self.tacker_client, 'create_vnfd',
151 return_value=self.createvnfd):
152 resp = openstack_tacker.create_vnfd(self.tacker_client,
154 self.assertEqual(resp, self.createvnfd)
156 def test_create_vnfd_default(self):
157 with mock.patch.object(self.tacker_client, 'create_vnfd',
158 return_value=self.createvnfd), \
159 mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
161 resp = openstack_tacker.create_vnfd(self.tacker_client,
162 tosca_file=self.tosca_file)
163 m.assert_called_once_with(self.tosca_file)
164 self.assertEqual(resp, self.createvnfd)
166 @mock.patch('functest.utils.openstack_tacker.logger.error')
167 def test_create_vnfd_exception(self, mock_logger_error):
168 with mock.patch.object(self.tacker_client, 'create_vnfd',
169 side_effect=Exception):
170 resp = openstack_tacker.create_vnfd(self.tacker_client,
171 tosca_file=self.tosca_file)
172 mock_logger_error.assert_called_once_with(test_utils.
181 self.assertIsNone(resp)
183 def test_delete_vnfd(self):
184 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
185 return_value=self.vnfd), \
186 mock.patch.object(self.tacker_client, 'delete_vnfd',
187 return_value=self.vnfd):
188 resp = openstack_tacker.delete_vnfd(self.tacker_client,
191 self.assertEqual(resp, self.vnfd)
193 def test_delete_vnfd_missing_vnfd_name(self):
194 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
195 return_value=self.vnfd), \
196 self.assertRaises(Exception) as context:
197 resp = openstack_tacker.delete_vnfd(self.tacker_client,
200 self.assertIsNone(resp)
201 msg = 'You need to provide VNFD id or VNFD name'
202 self.assertTrue(msg in context)
204 @mock.patch('functest.utils.openstack_tacker.logger.error')
205 def test_delete_vnfd_exception(self, mock_logger_error):
206 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
207 return_value=self.vnfd), \
208 mock.patch.object(self.tacker_client, 'delete_vnfd',
209 side_effect=Exception):
210 resp = openstack_tacker.delete_vnfd(self.tacker_client,
213 self.assertIsNone(resp)
214 self.assertTrue(mock_logger_error.called)
216 def test_list_vnfs(self):
217 with mock.patch.object(self.tacker_client, 'list_vnfs',
218 return_value=self.vnflist):
219 resp = openstack_tacker.list_vnfs(self.tacker_client,
221 self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
223 def test_list_vnfs_verbose(self):
224 with mock.patch.object(self.tacker_client, 'list_vnfs',
225 return_value=self.vnflist):
226 resp = openstack_tacker.list_vnfs(self.tacker_client,
228 self.assertEqual(resp, self.vnflist)
230 @mock.patch('functest.utils.openstack_tacker.logger.error')
231 def test_list_vnfs_exception(self, mock_logger_error):
232 with mock.patch.object(self.tacker_client, 'list_vnfs',
233 side_effect=Exception):
234 resp = openstack_tacker.list_vnfs(self.tacker_client,
236 mock_logger_error.assert_called_once_with(test_utils.
242 self.assertIsNone(resp)
244 def test_create_vnf_default(self):
245 with mock.patch.object(self.tacker_client, 'create_vnf',
246 return_value=self.createvnf), \
247 mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
248 return_value=self.vnf):
249 resp = openstack_tacker.create_vnf(self.tacker_client,
253 self.assertEqual(resp, self.createvnf)
255 @mock.patch('functest.utils.openstack_tacker.logger.error')
256 def test_create_vnf_exception(self, mock_logger_error):
257 with mock.patch.object(self.tacker_client, 'create_vnf',
258 side_effect=Exception):
259 resp = openstack_tacker.create_vnf(self.tacker_client,
263 mock_logger_error.assert_called_once_with(test_utils.
269 self.assertIsNone(resp)
271 def test_wait_for_vnf_vnf_retrieval_failed(self):
272 with mock.patch('functest.utils.openstack_tacker.get_vnf',
273 return_value=None), \
274 self.assertRaises(Exception) as context:
275 openstack_tacker.wait_for_vnf(self.tacker_client,
278 msg = ("Could not retrieve VNF - id='vnf_id', "
280 self.assertTrue(msg in context)
281 with mock.patch('functest.utils.openstack_tacker.get_vnf',
282 side_effect=Exception):
283 ret = openstack_tacker.wait_for_vnf(self.tacker_client,
286 self.assertEqual(ret, None)
288 def test_wait_for_vnf_vnf_status_error(self):
289 vnf = {'id': 'vnf_id',
291 with mock.patch('functest.utils.openstack_tacker.get_vnf',
293 self.assertRaises(Exception) as context:
294 openstack_tacker.wait_for_vnf(self.tacker_client,
297 msg = ('Error when booting vnf vnf_id')
298 self.assertTrue(msg in context)
300 def test_wait_for_vnf_vnf_timeout(self):
301 vnf = {'id': 'vnf_id',
302 'status': 'PENDING_CREATE'}
303 with mock.patch('functest.utils.openstack_tacker.get_vnf',
305 self.assertRaises(Exception) as context:
306 openstack_tacker.wait_for_vnf(self.tacker_client,
310 msg = ('Timeout when booting vnf vnf_id')
311 self.assertTrue(msg in context)
313 def test_delete_vnf(self):
314 with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
315 return_value=self.vnf), \
316 mock.patch.object(self.tacker_client, 'delete_vnf',
317 return_value=self.vnf):
318 resp = openstack_tacker.delete_vnf(self.tacker_client,
321 self.assertEqual(resp, self.vnf)
323 def test_delete_vnf_missing_vnf_name(self):
324 with self.assertRaises(Exception) as context:
325 openstack_tacker.delete_vnf(self.tacker_client,
328 msg = 'You need to provide a VNF id or name'
329 self.assertTrue(msg in context)
331 @mock.patch('functest.utils.openstack_tacker.logger.error')
332 def test_delete_vnf_exception(self, mock_logger_error):
333 with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
334 return_value=self.vnf), \
335 mock.patch.object(self.tacker_client, 'delete_vnf',
336 side_effect=Exception):
337 resp = openstack_tacker.delete_vnf(self.tacker_client,
340 self.assertIsNone(resp)
341 self.assertTrue(mock_logger_error.called)
343 def test_list_sfcs(self):
344 with mock.patch.object(self.tacker_client, 'list_sfcs',
345 return_value=self.sfclist):
346 resp = openstack_tacker.list_sfcs(self.tacker_client,
348 self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
350 def test_list_sfcs_verbose(self):
351 with mock.patch.object(self.tacker_client, 'list_sfcs',
352 return_value=self.sfclist):
353 resp = openstack_tacker.list_sfcs(self.tacker_client,
355 self.assertEqual(resp, self.sfclist)
357 @mock.patch('functest.utils.openstack_tacker.logger.error')
358 def test_list_sfcs_exception(self, mock_logger_error):
359 with mock.patch.object(self.tacker_client, 'list_sfcs',
360 side_effect=Exception):
361 resp = openstack_tacker.list_sfcs(self.tacker_client,
363 mock_logger_error.assert_called_once_with(test_utils.
369 self.assertIsNone(resp)
371 def test_create_sfc_default(self):
372 with mock.patch.object(self.tacker_client, 'create_sfc',
373 return_value=self.createsfc), \
374 mock.patch('functest.utils.openstack_tacker.get_vnf_id',
375 return_value=self.vnf):
376 resp = openstack_tacker.create_sfc(self.tacker_client,
378 chain_vnf_ids=['chain_vnf_id'],
379 chain_vnf_names=[self.vnf])
380 self.assertEqual(resp, self.createsfc)
382 @mock.patch('functest.utils.openstack_tacker.logger.error')
383 def test_create_sfc_exception(self, mock_logger_error):
384 with mock.patch.object(self.tacker_client, 'create_sfc',
385 side_effect=Exception):
386 resp = openstack_tacker.create_sfc(self.tacker_client,
388 chain_vnf_ids=['chain_vnf_id'],
389 chain_vnf_names=[self.vnf])
390 mock_logger_error.assert_called_once_with(test_utils.
396 self.assertIsNone(resp)
398 def test_delete_sfc(self):
399 with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
400 return_value=self.sfc), \
401 mock.patch.object(self.tacker_client, 'delete_sfc',
402 return_value=self.sfc):
403 resp = openstack_tacker.delete_sfc(self.tacker_client,
406 self.assertEqual(resp, self.sfc)
408 def test_delete_sfc_missing_sfc_name(self):
409 with self.assertRaises(Exception) as context:
410 openstack_tacker.delete_sfc(self.tacker_client,
413 msg = 'You need to provide an SFC id or name'
414 self.assertTrue(msg in context)
416 @mock.patch('functest.utils.openstack_tacker.logger.error')
417 def test_delete_sfc_exception(self, mock_logger_error):
418 with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
419 return_value=self.sfc), \
420 mock.patch.object(self.tacker_client, 'delete_sfc',
421 side_effect=Exception):
422 resp = openstack_tacker.delete_sfc(self.tacker_client,
425 self.assertIsNone(resp)
426 self.assertTrue(mock_logger_error.called)
428 def test_list_sfc_classifiers(self):
429 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
430 return_value=self.sfc_classifierlist):
431 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
433 self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
435 def test_list_sfc_classifiers_verbose(self):
436 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
437 return_value=self.sfc_classifierlist):
438 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
440 self.assertEqual(resp, self.sfc_classifierlist)
442 @mock.patch('functest.utils.openstack_tacker.logger.error')
443 def test_list_sfc_classifiers_exception(self, mock_logger_error):
444 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
445 side_effect=Exception):
446 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
448 mock_logger_error.assert_called_once_with(test_utils.
455 self.assertIsNone(resp)
457 def test_create_sfc_classifier_default(self):
458 with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
459 return_value=self.createsfc_clf), \
460 mock.patch('functest.utils.openstack_tacker.get_sfc_id',
461 return_value=self.sfc):
463 resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
467 self.assertEqual(resp, self.createsfc_clf)
469 @mock.patch('functest.utils.openstack_tacker.logger.error')
470 def test_sfc_classifier_exception(self, mock_logger_error):
471 with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
472 side_effect=Exception):
474 resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
478 mock_logger_error.assert_called_once_with(test_utils.
485 self.assertIsNone(resp)
487 def test_delete_sfc_classifier(self):
488 with mock.patch('functest.utils.openstack_tacker.get_sfc_'
490 return_value=self.sfc_clf), \
491 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
492 return_value=self.sfc_clf):
494 resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
497 self.assertEqual(resp, cl)
499 def test_delete_sfc_classifier_missing_sfc_name(self):
500 with self.assertRaises(Exception) as context:
501 openstack_tacker.delete_vnf(self.tacker_client,
504 msg = 'You need to provide an SFCclassifier id or name'
505 self.assertTrue(msg in context)
507 @mock.patch('functest.utils.openstack_tacker.logger.error')
508 def test_delete_sfc_classifier_exception(self, mock_logger_error):
509 with mock.patch('functest.utils.openstack_tacker.get_sfc_'
511 return_value=self.sfc_clf), \
512 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
513 side_effect=Exception):
515 resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
518 self.assertIsNone(resp)
519 self.assertTrue(mock_logger_error.called)
522 if __name__ == "__main__":
523 logging.disable(logging.CRITICAL)
524 unittest.main(verbosity=2)