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):
20 logging.disable(logging.CRITICAL)
23 self.tacker_client = mock.Mock()
24 self.getresponse = {'vnfds': [{'id': 'test_id'}],
25 'vnfs': [{'id': 'test_id'}],
26 'sfcs': [{'id': 'test_id'}]}
27 self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]}
28 self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]}
29 self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]}
30 self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'},
31 {'id': 'test_sfc_cl2'}]}
33 self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}}
34 self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}}
35 self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}}
36 self.createsfc_clf = {"sfc_classifier": {"attributes":
37 {"sfc_clf": 'sfc_clf_body'}}}
39 self.resource_type = 'vnfd'
40 self.resource_name = 'resource_name'
41 self.tosca_file = 'test_tosca_file'
42 self.vnfd = 'test_vnfd'
45 self.sfc_clf = 'test_sfc_clf'
49 'OS_USERNAME': 'username',
50 'OS_PASSWORD': 'password',
51 'OS_AUTH_URL': 'auth_url',
52 'OS_TENANT_NAME': 'tenant_name',
53 'OS_USER_DOMAIN_NAME': 'user_domain_name',
54 'OS_PROJECT_DOMAIN_NAME': 'project_domain_name',
55 'OS_PROJECT_NAME': 'project_name',
56 'OS_ENDPOINT_TYPE': 'endpoint_type',
57 'OS_REGION_NAME': 'region_name'
61 def test_get_tacker_client(self):
62 with mock.patch('functest.utils.openstack_tacker.'
63 'os_utils.get_session'):
64 tackerclient.Client = mock.Mock
65 ret = openstack_tacker.get_tacker_client()
66 self.assertTrue(isinstance(ret, mock.Mock))
68 def test_get_id_from_name(self):
69 with mock.patch.object(self.tacker_client, 'get',
70 return_value=self.getresponse):
71 resp = openstack_tacker.get_id_from_name(self.tacker_client,
74 self.assertEqual(resp, 'test_id')
76 @mock.patch('functest.utils.openstack_tacker.logger.error')
77 def test_get_id_from_name_exception(self, mock_logger_error):
78 with mock.patch.object(self.tacker_client, 'get',
79 side_effect=Exception):
80 resp = openstack_tacker.get_id_from_name(self.tacker_client,
83 self.assertIsNone(resp)
84 mock_logger_error.assert_called_once_with(test_utils.
85 SubstrMatch("Error [get"
94 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
95 def test_get_vnfd_id(self, mock_get_id_from_name):
96 openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name)
97 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
101 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
102 def test_get_vnf_id(self, mock_get_id_from_name):
103 openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name)
104 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
108 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
109 def test_get_sfc_id(self, mock_get_id_from_name):
110 openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name)
111 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
115 @mock.patch('functest.utils.openstack_tacker.get_id_from_name')
116 def test_get_sfc_classifier_id(self, mock_get_id_from_name):
117 openstack_tacker.get_sfc_classifier_id(self.tacker_client,
119 mock_get_id_from_name.assert_called_once_with(self.tacker_client,
123 def test_list_vnfds(self):
124 with mock.patch.object(self.tacker_client, 'list_vnfds',
125 return_value=self.vnfdlist):
126 resp = openstack_tacker.list_vnfds(self.tacker_client,
128 self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2'])
130 def test_list_vnfds_verbose(self):
131 with mock.patch.object(self.tacker_client, 'list_vnfds',
132 return_value=self.vnfdlist):
133 resp = openstack_tacker.list_vnfds(self.tacker_client,
135 self.assertEqual(resp, self.vnfdlist)
137 @mock.patch('functest.utils.openstack_tacker.logger.error')
138 def test_list_vnfds_exception(self, mock_logger_error):
139 with mock.patch.object(self.tacker_client, 'list_vnfds',
140 side_effect=Exception):
141 resp = openstack_tacker.list_vnfds(self.tacker_client,
143 mock_logger_error.assert_called_once_with(test_utils.
149 self.assertIsNone(resp)
151 def test_create_vnfd_missing_file(self):
152 with mock.patch.object(self.tacker_client, 'create_vnfd',
153 return_value=self.createvnfd):
154 resp = openstack_tacker.create_vnfd(self.tacker_client,
156 self.assertEqual(resp, self.createvnfd)
158 def test_create_vnfd_default(self):
159 with mock.patch.object(self.tacker_client, 'create_vnfd',
160 return_value=self.createvnfd), \
161 mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \
163 resp = openstack_tacker.create_vnfd(self.tacker_client,
164 tosca_file=self.tosca_file)
165 m.assert_called_once_with(self.tosca_file)
166 self.assertEqual(resp, self.createvnfd)
168 @mock.patch('functest.utils.openstack_tacker.logger.error')
169 def test_create_vnfd_exception(self, mock_logger_error):
170 with mock.patch.object(self.tacker_client, 'create_vnfd',
171 side_effect=Exception):
172 resp = openstack_tacker.create_vnfd(self.tacker_client,
173 tosca_file=self.tosca_file)
174 mock_logger_error.assert_called_once_with(test_utils.
183 self.assertIsNone(resp)
185 def test_delete_vnfd(self):
186 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
187 return_value=self.vnfd), \
188 mock.patch.object(self.tacker_client, 'delete_vnfd',
189 return_value=self.vnfd):
190 resp = openstack_tacker.delete_vnfd(self.tacker_client,
193 self.assertEqual(resp, self.vnfd)
195 def test_delete_vnfd_missing_vnfd_name(self):
196 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
197 return_value=self.vnfd), \
198 self.assertRaises(Exception) as context:
199 resp = openstack_tacker.delete_vnfd(self.tacker_client,
202 self.assertIsNone(resp)
203 msg = 'You need to provide VNFD id or VNFD name'
204 self.assertTrue(msg in context)
206 @mock.patch('functest.utils.openstack_tacker.logger.error')
207 def test_delete_vnfd_exception(self, mock_logger_error):
208 with mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
209 return_value=self.vnfd), \
210 mock.patch.object(self.tacker_client, 'delete_vnfd',
211 side_effect=Exception):
212 resp = openstack_tacker.delete_vnfd(self.tacker_client,
215 self.assertIsNone(resp)
216 self.assertTrue(mock_logger_error.called)
218 def test_list_vnfs(self):
219 with mock.patch.object(self.tacker_client, 'list_vnfs',
220 return_value=self.vnflist):
221 resp = openstack_tacker.list_vnfs(self.tacker_client,
223 self.assertEqual(resp, ['test_vnf1', 'test_vnf2'])
225 def test_list_vnfs_verbose(self):
226 with mock.patch.object(self.tacker_client, 'list_vnfs',
227 return_value=self.vnflist):
228 resp = openstack_tacker.list_vnfs(self.tacker_client,
230 self.assertEqual(resp, self.vnflist)
232 @mock.patch('functest.utils.openstack_tacker.logger.error')
233 def test_list_vnfs_exception(self, mock_logger_error):
234 with mock.patch.object(self.tacker_client, 'list_vnfs',
235 side_effect=Exception):
236 resp = openstack_tacker.list_vnfs(self.tacker_client,
238 mock_logger_error.assert_called_once_with(test_utils.
244 self.assertIsNone(resp)
246 def test_create_vnf_default(self):
247 with mock.patch.object(self.tacker_client, 'create_vnf',
248 return_value=self.createvnf), \
249 mock.patch('functest.utils.openstack_tacker.get_vnfd_id',
250 return_value=self.vnf):
251 resp = openstack_tacker.create_vnf(self.tacker_client,
255 self.assertEqual(resp, self.createvnf)
257 @mock.patch('functest.utils.openstack_tacker.logger.error')
258 def test_create_vnf_exception(self, mock_logger_error):
259 with mock.patch.object(self.tacker_client, 'create_vnf',
260 side_effect=Exception):
261 resp = openstack_tacker.create_vnf(self.tacker_client,
265 mock_logger_error.assert_called_once_with(test_utils.
271 self.assertIsNone(resp)
273 def test_wait_for_vnf_vnf_retrieval_failed(self):
274 with mock.patch('functest.utils.openstack_tacker.get_vnf',
275 return_value=None), \
276 self.assertRaises(Exception) as context:
277 openstack_tacker.wait_for_vnf(self.tacker_client,
280 msg = ("Could not retrieve VNF - id='vnf_id', "
282 self.assertTrue(msg in context)
283 with mock.patch('functest.utils.openstack_tacker.get_vnf',
284 side_effect=Exception):
285 ret = openstack_tacker.wait_for_vnf(self.tacker_client,
288 self.assertEqual(ret, None)
290 def test_wait_for_vnf_vnf_status_error(self):
291 vnf = {'id': 'vnf_id',
293 with mock.patch('functest.utils.openstack_tacker.get_vnf',
295 self.assertRaises(Exception) as context:
296 openstack_tacker.wait_for_vnf(self.tacker_client,
299 msg = ('Error when booting vnf vnf_id')
300 self.assertTrue(msg in context)
302 def test_wait_for_vnf_vnf_timeout(self):
303 vnf = {'id': 'vnf_id',
304 'status': 'PENDING_CREATE'}
305 with mock.patch('functest.utils.openstack_tacker.get_vnf',
307 self.assertRaises(Exception) as context:
308 openstack_tacker.wait_for_vnf(self.tacker_client,
312 msg = ('Timeout when booting vnf vnf_id')
313 self.assertTrue(msg in context)
315 def test_delete_vnf(self):
316 with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
317 return_value=self.vnf), \
318 mock.patch.object(self.tacker_client, 'delete_vnf',
319 return_value=self.vnf):
320 resp = openstack_tacker.delete_vnf(self.tacker_client,
323 self.assertEqual(resp, self.vnf)
325 def test_delete_vnf_missing_vnf_name(self):
326 with self.assertRaises(Exception) as context:
327 openstack_tacker.delete_vnf(self.tacker_client,
330 msg = 'You need to provide a VNF id or name'
331 self.assertTrue(msg in context)
333 @mock.patch('functest.utils.openstack_tacker.logger.error')
334 def test_delete_vnf_exception(self, mock_logger_error):
335 with mock.patch('functest.utils.openstack_tacker.get_vnf_id',
336 return_value=self.vnf), \
337 mock.patch.object(self.tacker_client, 'delete_vnf',
338 side_effect=Exception):
339 resp = openstack_tacker.delete_vnf(self.tacker_client,
342 self.assertIsNone(resp)
343 self.assertTrue(mock_logger_error.called)
345 def test_list_sfcs(self):
346 with mock.patch.object(self.tacker_client, 'list_sfcs',
347 return_value=self.sfclist):
348 resp = openstack_tacker.list_sfcs(self.tacker_client,
350 self.assertEqual(resp, ['test_sfc1', 'test_sfc2'])
352 def test_list_sfcs_verbose(self):
353 with mock.patch.object(self.tacker_client, 'list_sfcs',
354 return_value=self.sfclist):
355 resp = openstack_tacker.list_sfcs(self.tacker_client,
357 self.assertEqual(resp, self.sfclist)
359 @mock.patch('functest.utils.openstack_tacker.logger.error')
360 def test_list_sfcs_exception(self, mock_logger_error):
361 with mock.patch.object(self.tacker_client, 'list_sfcs',
362 side_effect=Exception):
363 resp = openstack_tacker.list_sfcs(self.tacker_client,
365 mock_logger_error.assert_called_once_with(test_utils.
371 self.assertIsNone(resp)
373 def test_create_sfc_default(self):
374 with mock.patch.object(self.tacker_client, 'create_sfc',
375 return_value=self.createsfc), \
376 mock.patch('functest.utils.openstack_tacker.get_vnf_id',
377 return_value=self.vnf):
378 resp = openstack_tacker.create_sfc(self.tacker_client,
380 chain_vnf_ids=['chain_vnf_id'],
381 chain_vnf_names=[self.vnf])
382 self.assertEqual(resp, self.createsfc)
384 @mock.patch('functest.utils.openstack_tacker.logger.error')
385 def test_create_sfc_exception(self, mock_logger_error):
386 with mock.patch.object(self.tacker_client, 'create_sfc',
387 side_effect=Exception):
388 resp = openstack_tacker.create_sfc(self.tacker_client,
390 chain_vnf_ids=['chain_vnf_id'],
391 chain_vnf_names=[self.vnf])
392 mock_logger_error.assert_called_once_with(test_utils.
398 self.assertIsNone(resp)
400 def test_delete_sfc(self):
401 with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
402 return_value=self.sfc), \
403 mock.patch.object(self.tacker_client, 'delete_sfc',
404 return_value=self.sfc):
405 resp = openstack_tacker.delete_sfc(self.tacker_client,
408 self.assertEqual(resp, self.sfc)
410 def test_delete_sfc_missing_sfc_name(self):
411 with self.assertRaises(Exception) as context:
412 openstack_tacker.delete_sfc(self.tacker_client,
415 msg = 'You need to provide an SFC id or name'
416 self.assertTrue(msg in context)
418 @mock.patch('functest.utils.openstack_tacker.logger.error')
419 def test_delete_sfc_exception(self, mock_logger_error):
420 with mock.patch('functest.utils.openstack_tacker.get_sfc_id',
421 return_value=self.sfc), \
422 mock.patch.object(self.tacker_client, 'delete_sfc',
423 side_effect=Exception):
424 resp = openstack_tacker.delete_sfc(self.tacker_client,
427 self.assertIsNone(resp)
428 self.assertTrue(mock_logger_error.called)
430 def test_list_sfc_classifiers(self):
431 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
432 return_value=self.sfc_classifierlist):
433 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
435 self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2'])
437 def test_list_sfc_classifiers_verbose(self):
438 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
439 return_value=self.sfc_classifierlist):
440 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
442 self.assertEqual(resp, self.sfc_classifierlist)
444 @mock.patch('functest.utils.openstack_tacker.logger.error')
445 def test_list_sfc_classifiers_exception(self, mock_logger_error):
446 with mock.patch.object(self.tacker_client, 'list_sfc_classifiers',
447 side_effect=Exception):
448 resp = openstack_tacker.list_sfc_classifiers(self.tacker_client,
450 mock_logger_error.assert_called_once_with(test_utils.
457 self.assertIsNone(resp)
459 def test_create_sfc_classifier_default(self):
460 with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
461 return_value=self.createsfc_clf), \
462 mock.patch('functest.utils.openstack_tacker.get_sfc_id',
463 return_value=self.sfc):
465 resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
469 self.assertEqual(resp, self.createsfc_clf)
471 @mock.patch('functest.utils.openstack_tacker.logger.error')
472 def test_sfc_classifier_exception(self, mock_logger_error):
473 with mock.patch.object(self.tacker_client, 'create_sfc_classifier',
474 side_effect=Exception):
476 resp = openstack_tacker.create_sfc_classifier(self.tacker_client,
480 mock_logger_error.assert_called_once_with(test_utils.
487 self.assertIsNone(resp)
489 def test_delete_sfc_classifier(self):
490 with mock.patch('functest.utils.openstack_tacker.get_sfc_'
492 return_value=self.sfc_clf), \
493 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
494 return_value=self.sfc_clf):
496 resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
499 self.assertEqual(resp, cl)
501 def test_delete_sfc_classifier_missing_sfc_name(self):
502 with self.assertRaises(Exception) as context:
503 openstack_tacker.delete_vnf(self.tacker_client,
506 msg = 'You need to provide an SFCclassifier id or name'
507 self.assertTrue(msg in context)
509 @mock.patch('functest.utils.openstack_tacker.logger.error')
510 def test_delete_sfc_classifier_exception(self, mock_logger_error):
511 with mock.patch('functest.utils.openstack_tacker.get_sfc_'
513 return_value=self.sfc_clf), \
514 mock.patch.object(self.tacker_client, 'delete_sfc_classifier',
515 side_effect=Exception):
517 resp = openstack_tacker.delete_sfc_classifier(self.tacker_client,
520 self.assertIsNone(resp)
521 self.assertTrue(mock_logger_error.called)
524 if __name__ == "__main__":
525 unittest.main(verbosity=2)