diff options
Diffstat (limited to 'functest/tests/unit/utils/test_openstack_tacker.py')
| -rw-r--r-- | functest/tests/unit/utils/test_openstack_tacker.py | 455 |
1 files changed, 455 insertions, 0 deletions
diff --git a/functest/tests/unit/utils/test_openstack_tacker.py b/functest/tests/unit/utils/test_openstack_tacker.py new file mode 100644 index 00000000..a8330c0e --- /dev/null +++ b/functest/tests/unit/utils/test_openstack_tacker.py @@ -0,0 +1,455 @@ +#!/usr/bin/env python + +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +import logging +import mock +import unittest + +from functest.utils import openstack_tacker +from functest.tests.unit import test_utils + + +class OSTackerTesting(unittest.TestCase): + + logging.disable(logging.CRITICAL) + + def setUp(self): + self.tacker_client = mock.Mock() + self.getresponse = {'vnfds': [{'id': 'test_id'}], + 'vnfs': [{'id': 'test_id'}], + 'sfcs': [{'id': 'test_id'}]} + self.vnfdlist = {'vnfds': [{'id': 'test_vnfd1'}, {'id': 'test_vnfd2'}]} + self.vnflist = {'vnfs': [{'id': 'test_vnf1'}, {'id': 'test_vnf2'}]} + self.sfclist = {'sfcs': [{'id': 'test_sfc1'}, {'id': 'test_sfc2'}]} + self.sfc_classifierlist = {'sfc_classifiers': [{'id': 'test_sfc_cl1'}, + {'id': 'test_sfc_cl2'}]} + + self.createvnfd = {"vnfd": {"attributes": {"vnfd": 'vnfd_body'}}} + self.createvnf = {"vnf": {"attributes": {"vnf": 'vnf_body'}}} + self.createsfc = {"sfc": {"attributes": {"sfc": 'sfc_body'}}} + self.createsfc_clf = {"sfc_classifier": {"attributes": + {"sfc_clf": 'sfc_clf_body'}}} + + self.resource_type = 'vnfd' + self.resource_name = 'resource_name' + self.tosca_file = 'test_tosca_file' + self.vnfd = 'test_vnfd' + self.vnf = 'test_vnf' + self.sfc = 'test_sfc' + self.sfc_clf = 'test_sfc_clf' + + def _get_creds(self): + cred_dict = { + 'OS_USERNAME': 'username', + 'OS_PASSWORD': 'password', + 'OS_AUTH_URL': 'auth_url', + 'OS_TENANT_NAME': 'tenant_name', + 'OS_USER_DOMAIN_NAME': 'user_domain_name', + 'OS_PROJECT_DOMAIN_NAME': 'project_domain_name', + 'OS_PROJECT_NAME': 'project_name', + 'OS_ENDPOINT_TYPE': 'endpoint_type', + 'OS_REGION_NAME': 'region_name' + } + return cred_dict + + def test_get_id_from_name(self): + with mock.patch.object(self.tacker_client, 'get', + return_value=self.getresponse): + resp = openstack_tacker.get_id_from_name(self.tacker_client, + self.resource_type, + self.resource_name) + self.assertEqual(resp, 'test_id') + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_get_id_from_name_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'get', + side_effect=Exception): + resp = openstack_tacker.get_id_from_name(self.tacker_client, + self.resource_type, + self.resource_name) + self.assertIsNone(resp) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error [get" + "_id_from_" + "name(tacker" + "_client" + ", resource_" + "type, " + "resource_" + "name)]:")) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_vnfd_id(self, mock_get_id_from_name): + openstack_tacker.get_vnfd_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'vnfd', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_vnf_id(self, mock_get_id_from_name): + openstack_tacker.get_vnf_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'vnf', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_sfc_id(self, mock_get_id_from_name): + openstack_tacker.get_sfc_id(self.tacker_client, self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'sfc', + self.resource_name) + + @mock.patch('functest.utils.openstack_tacker.get_id_from_name') + def test_get_sfc_classifier_id(self, mock_get_id_from_name): + openstack_tacker.get_sfc_classifier_id(self.tacker_client, + self.resource_name) + mock_get_id_from_name.assert_called_once_with(self.tacker_client, + 'sfc-classifier', + self.resource_name) + + def test_list_vnfds(self): + with mock.patch.object(self.tacker_client, 'list_vnfds', + return_value=self.vnfdlist): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_vnfd1', 'test_vnfd2']) + + def test_list_vnfds_verbose(self): + with mock.patch.object(self.tacker_client, 'list_vnfds', + return_value=self.vnfdlist): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.vnfdlist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_vnfds_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_vnfds', + side_effect=Exception): + resp = openstack_tacker.list_vnfds(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_vnfds(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_vnfd_missing_file(self): + with mock.patch.object(self.tacker_client, 'create_vnfd', + return_value=self.createvnfd): + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=None) + self.assertEqual(resp, self.createvnfd) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_vnfd_default(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_vnfd', + return_value=self.createvnfd), \ + mock.patch('__builtin__.open', mock.mock_open(read_data='1')) \ + as m: + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=self.tosca_file) + m.assert_called_once_with(self.tosca_file) + mock_logger_error.assert_called_once_with('1') + self.assertEqual(resp, self.createvnfd) + + @mock.patch('functest.utils.openstack_tacker.logger.exception') + def test_create_vnfd_exception(self, mock_logger_excep): + with mock.patch.object(self.tacker_client, 'create_vnfd', + side_effect=Exception): + resp = openstack_tacker.create_vnfd(self.tacker_client, + tosca_file=self.tosca_file) + mock_logger_excep.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [create" + "_vnfd(" + "tacker_" + "client, " + "'%s')]" + % self. + tosca_file)) + self.assertIsNone(resp) + + def test_delete_vnfd(self): + with mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnfd), \ + mock.patch.object(self.tacker_client, 'delete_vnfd', + return_value=self.vnfd): + resp = openstack_tacker.delete_vnfd(self.tacker_client, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + self.assertEqual(resp, self.vnfd) + + # TODO: Exception('You need to provide an VNFD' + # 'id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_vnfd_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnfd), \ + mock.patch.object(self.tacker_client, 'delete_vnfd', + side_effect=Exception): + resp = openstack_tacker.delete_vnfd(self.tacker_client, + vnfd_id=None, + vnfd_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_vnfs(self): + with mock.patch.object(self.tacker_client, 'list_vnfs', + return_value=self.vnflist): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_vnf1', 'test_vnf2']) + + def test_list_vnfs_verbose(self): + with mock.patch.object(self.tacker_client, 'list_vnfs', + return_value=self.vnflist): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.vnflist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_vnfs_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_vnfs', + side_effect=Exception): + resp = openstack_tacker.list_vnfs(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_vnfs(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_vnf_default(self): + with mock.patch.object(self.tacker_client, 'create_vnf', + return_value=self.createvnf), \ + mock.patch('functest.utils.openstack_tacker.get_vnfd_id', + return_value=self.vnf): + resp = openstack_tacker.create_vnf(self.tacker_client, + vnf_name=self.vnf, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + self.assertEqual(resp, self.createvnf) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_vnf_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_vnf', + side_effect=Exception): + resp = openstack_tacker.create_vnf(self.tacker_client, + vnf_name=self.vnf, + vnfd_id='vnfd_id', + vnfd_name=self.vnfd) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_vnf(" + "tacker_" + "client")) + self.assertIsNone(resp) + + # TODO: wait_for_vnf + + def test_delete_vnf(self): + with mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf), \ + mock.patch.object(self.tacker_client, 'delete_vnf', + return_value=self.vnf): + resp = openstack_tacker.delete_vnf(self.tacker_client, + vnf_id='vnf_id', + vnf_name=self.vnf) + self.assertEqual(resp, self.vnf) + + # TODO: Exception('You need to provide an VNF' + # 'classifier id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_vnf_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf), \ + mock.patch.object(self.tacker_client, 'delete_vnf', + side_effect=Exception): + resp = openstack_tacker.delete_vnf(self.tacker_client, + vnf_id=None, + vnf_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_sfcs(self): + with mock.patch.object(self.tacker_client, 'list_sfcs', + return_value=self.sfclist): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_sfc1', 'test_sfc2']) + + def test_list_sfcs_verbose(self): + with mock.patch.object(self.tacker_client, 'list_sfcs', + return_value=self.sfclist): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.sfclist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_sfcs_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_sfcs', + side_effect=Exception): + resp = openstack_tacker.list_sfcs(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_sfcs(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_sfc_default(self): + with mock.patch.object(self.tacker_client, 'create_sfc', + return_value=self.createsfc), \ + mock.patch('functest.utils.openstack_tacker.get_vnf_id', + return_value=self.vnf): + resp = openstack_tacker.create_sfc(self.tacker_client, + sfc_name=self.sfc, + chain_vnf_ids=['chain_vnf_id'], + chain_vnf_names=[self.vnf]) + self.assertEqual(resp, self.createsfc) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_create_sfc_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_sfc', + side_effect=Exception): + resp = openstack_tacker.create_sfc(self.tacker_client, + sfc_name=self.sfc, + chain_vnf_ids=['chain_vnf_id'], + chain_vnf_names=[self.vnf]) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_sfc(" + "tacker_" + "client")) + self.assertIsNone(resp) + + def test_delete_sfc(self): + with mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc), \ + mock.patch.object(self.tacker_client, 'delete_sfc', + return_value=self.sfc): + resp = openstack_tacker.delete_sfc(self.tacker_client, + sfc_id='sfc_id', + sfc_name=self.sfc) + self.assertEqual(resp, self.sfc) + + # TODO: Exception('You need to provide an SFC' + # 'id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_sfc_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc), \ + mock.patch.object(self.tacker_client, 'delete_sfc', + side_effect=Exception): + resp = openstack_tacker.delete_sfc(self.tacker_client, + sfc_id=None, + sfc_name=None) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + def test_list_sfc_classifiers(self): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + return_value=self.sfc_classifierlist): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=False) + self.assertEqual(resp, ['test_sfc_cl1', 'test_sfc_cl2']) + + def test_list_sfc_classifiers_verbose(self): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + return_value=self.sfc_classifierlist): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=True) + self.assertEqual(resp, self.sfc_classifierlist) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_list_sfc_classifiers_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'list_sfc_classifiers', + side_effect=Exception): + resp = openstack_tacker.list_sfc_classifiers(self.tacker_client, + verbose=False) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("Error" + " [list" + "_sfc_cl" + "assifiers(" + "tacker_" + "client)]:")) + self.assertIsNone(resp) + + def test_create_sfc_classifier_default(self): + with mock.patch.object(self.tacker_client, 'create_sfc_classifier', + return_value=self.createsfc_clf), \ + mock.patch('functest.utils.openstack_tacker.get_sfc_id', + return_value=self.sfc): + cl = self.sfc_clf + resp = openstack_tacker.create_sfc_classifier(self.tacker_client, + sfc_clf_name=cl, + sfc_id='sfc_id', + sfc_name=self.sfc) + self.assertEqual(resp, self.createsfc_clf) + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_sfc_classifier_exception(self, mock_logger_error): + with mock.patch.object(self.tacker_client, 'create_sfc_classifier', + side_effect=Exception): + cl = self.sfc_clf + resp = openstack_tacker.create_sfc_classifier(self.tacker_client, + sfc_clf_name=cl, + sfc_id='sfc_id', + sfc_name=self.sfc) + mock_logger_error.assert_called_once_with(test_utils. + SubstrMatch("error" + " [create" + "_sfc_cl" + "assifier(" + "tacker_" + "client")) + self.assertIsNone(resp) + + def test_delete_sfc_classifier(self): + with mock.patch('functest.utils.openstack_tacker.get_sfc_' + 'classifier_id', + return_value=self.sfc_clf), \ + mock.patch.object(self.tacker_client, 'delete_sfc_classifier', + return_value=self.sfc_clf): + cl = self.sfc_clf + resp = openstack_tacker.delete_sfc_classifier(self.tacker_client, + sfc_clf_id='sfc_id', + sfc_clf_name=cl) + self.assertEqual(resp, cl) + + # TODO: Exception('You need to provide an SFC' + # 'classifier id or name') AssertionError + + @mock.patch('functest.utils.openstack_tacker.logger.error') + def test_delete_sfc_classifier_exception(self, mock_logger_error): + with mock.patch('functest.utils.openstack_tacker.get_sfc_' + 'classifier_id', + return_value=self.sfc_clf), \ + mock.patch.object(self.tacker_client, 'delete_sfc_classifier', + side_effect=Exception): + cl = self.sfc_clf + resp = openstack_tacker.delete_sfc_classifier(self.tacker_client, + sfc_clf_id='sfc_id', + sfc_clf_name=cl) + self.assertIsNone(resp) + self.assertTrue(mock_logger_error.called) + + +if __name__ == "__main__": + unittest.main(verbosity=2) |