diff options
Diffstat (limited to 'sfc/unit_tests/unit/lib/test_cleanup.py')
| -rw-r--r-- | sfc/unit_tests/unit/lib/test_cleanup.py | 469 |
1 files changed, 469 insertions, 0 deletions
diff --git a/sfc/unit_tests/unit/lib/test_cleanup.py b/sfc/unit_tests/unit/lib/test_cleanup.py new file mode 100644 index 00000000..e6f59d23 --- /dev/null +++ b/sfc/unit_tests/unit/lib/test_cleanup.py @@ -0,0 +1,469 @@ +#!/usr/bin/env python + +############################################################################### +# Copyright (c) 2018 Intracom Telecom and others. +# +# 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 unittest +import sfc.lib.cleanup as cleanup + +from mock import patch +from mock import call +from mock import DEFAULT +from mock import Mock + + +__author__ = "Dimitrios Markou <mardim@intracom-telecom.com>" + + +class SfcCleanupTesting(unittest.TestCase): + + def setUp(self): + self.odl_ip = '10.10.10.10' + self.odl_port = '8081' + self.patcher = patch('sfc.lib.openstack_utils.get_tacker_client') + self.mock_tacker_client = self.patcher.start() + self.mock_tacker_client.return_value = 'tacker_client_obj' + + def tearDown(self): + self.patcher.stop() + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.odl_utils.delete_odl_resource_elem') + @patch('sfc.lib.odl_utils.odl_resource_list_names') + @patch('sfc.lib.odl_utils.get_odl_resource_list') + def test_delete_odl_resource(self, mock_resource_list, + mock_resource_list_name, + mock_del_resource_elem, + mock_log): + """ + Checks if the functions which belong to the odl_utils + library are getting called. + """ + + resource = 'mock_resource' + log_calls = [call("Removing ODL resource: mock_resource/elem_one"), + call("Removing ODL resource: mock_resource/elem_two")] + + del_calls = [call(self.odl_ip, self.odl_port, resource, 'elem_one'), + call(self.odl_ip, self.odl_port, resource, 'elem_two')] + + mock_resource_list_name.return_value = ['elem_one', 'elem_two'] + mock_resource_list.return_value = ['rsrc_one', + 'rsrc_two', + 'rsrc_three'] + + cleanup.delete_odl_resources(self.odl_ip, self.odl_port, resource) + + mock_resource_list.assert_called_once_with(self.odl_ip, + self.odl_port, + resource) + mock_resource_list_name.assert_called_once_with( + resource, ['rsrc_one', 'rsrc_two', 'rsrc_three']) + mock_del_resource_elem.assert_has_calls(del_calls) + mock_log.assert_has_calls(log_calls) + + @patch('sfc.lib.odl_utils.get_odl_acl_list') + @patch('sfc.lib.odl_utils.odl_acl_types_names') + @patch('sfc.lib.odl_utils.delete_odl_acl') + def test_delete_odl_ietf_access_lists(self, + mock_del_acl, + mock_acl_types, + mock_get_acls): + """ + Ckecks the proper functionality of the delete_odl_ietf_access_lists + function + """ + + mock_acl_type_name_list = [('acl_type_one', 'name_one'), + ('acl_type_two', 'name_two')] + mock_get_acls.return_value = ['acl_one', 'acl_two'] + mock_acl_types.return_value = mock_acl_type_name_list + del_calls = [call(self.odl_ip, self.odl_port, key, value) + for key, value in mock_acl_type_name_list] + + cleanup.delete_odl_ietf_access_lists(self.odl_ip, self.odl_port) + + mock_get_acls.assert_called_once_with(self.odl_ip, self.odl_port) + mock_acl_types.assert_called_once_with(['acl_one', 'acl_two']) + mock_del_acl.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.list_vnfds', return_value=None) + def test_delete_vnfds_returned_list_is_none(self, mock_list_vnfds): + """ + Check the proper functionality of the delete_vnfds + function when the returned vnfds list is None + """ + + self.assertIsNone(cleanup.delete_vnfds()) + mock_list_vnfds.assert_called_once_with('tacker_client_obj') + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.openstack_utils.delete_vnfd') + @patch('sfc.lib.openstack_utils.list_vnfds') + def test_delete_vnfds_not_empty_list(self, + mock_list_vnfds, + mock_del_vnfd, + mock_log): + """ + Check the proper functionality of the delete_vnfds + function when the returned vnfds list is not empty + """ + + mock_list_vnfds.return_value = ['vnfd_one', 'vnfd_two'] + log_calls = [call("Removing vnfd: vnfd_one"), + call("Removing vnfd: vnfd_two")] + + del_calls = [call('tacker_client_obj', vnfd_id='vnfd_one'), + call('tacker_client_obj', vnfd_id='vnfd_two')] + + cleanup.delete_vnfds() + mock_list_vnfds.assert_called_once_with('tacker_client_obj') + mock_log.assert_has_calls(log_calls) + mock_del_vnfd.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.list_vnfs', return_value=None) + def test_delete_vnfs_returned_list_is_none(self, mock_list_vnfs): + """ + Check the proper functionality of the delete_vnfs + function when the returned vnfs list is None + """ + + self.assertIsNone(cleanup.delete_vnfs()) + mock_list_vnfs.assert_called_once_with('tacker_client_obj') + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.openstack_utils.delete_vnf') + @patch('sfc.lib.openstack_utils.list_vnfs') + def test_delete_vnfs_not_empty_list(self, + mock_list_vnfs, + mock_del_vnf, + mock_log): + """ + Check the proper functionality of the delete_vnfs + function when the returned vnfs list is not empty + """ + + mock_list_vnfs.return_value = ['vnf_one', 'vnf_two'] + log_calls = [call("Removing vnf: vnf_one"), + call("Removing vnf: vnf_two")] + + del_calls = [call('tacker_client_obj', vnf_id='vnf_one'), + call('tacker_client_obj', vnf_id='vnf_two')] + + cleanup.delete_vnfs() + mock_list_vnfs.assert_called_once_with('tacker_client_obj') + mock_log.assert_has_calls(log_calls) + mock_del_vnf.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.list_vnffgs', return_value=None) + def test_delete_vnffgs_returned_list_is_none(self, mock_list_vnffgs): + """ + Check the proper functionality of the delete_vnffgs + function when the returned vnffgs list is None + """ + + self.assertIsNone(cleanup.delete_vnffgs()) + mock_list_vnffgs.assert_called_once_with('tacker_client_obj') + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.openstack_utils.delete_vnffg') + @patch('sfc.lib.openstack_utils.list_vnffgs') + def test_delete_vnffgs_not_empty_list(self, + mock_list_vnffgs, + mock_del_vnffg, + mock_log): + """ + Check the proper functionality of the delete_vnffgs + function when the returned vnffgs list is not empty + """ + + mock_list_vnffgs.return_value = ['vnffg_one', 'vnffg_two'] + log_calls = [call("Removing vnffg: vnffg_two"), + call("Removing vnffg: vnffg_one")] + + del_calls = [call('tacker_client_obj', vnffg_id='vnffg_two'), + call('tacker_client_obj', vnffg_id='vnffg_one')] + + cleanup.delete_vnffgs() + mock_list_vnffgs.assert_called_once_with('tacker_client_obj') + mock_log.assert_has_calls(log_calls) + mock_del_vnffg.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.list_vnffgds', return_value=None) + def test_delete_vnffgds_returned_list_is_none(self, mock_list_vnffgds): + """ + Check the proper functionality of the delete_vnffgds + function when the returned vnffgds list is None + """ + + self.assertIsNone(cleanup.delete_vnffgds()) + mock_list_vnffgds.assert_called_once_with('tacker_client_obj') + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.openstack_utils.delete_vnffgd') + @patch('sfc.lib.openstack_utils.list_vnffgds') + def test_delete_vnffgds_not_empty_list(self, + mock_list_vnffgds, + mock_del_vnffgd, + mock_log): + """ + Check the proper functionality of the delete_vnffgds + function when the returned vnffgds list is not empty + """ + + mock_list_vnffgds.return_value = ['vnffgd_one', 'vnffgd_two'] + log_calls = [call("Removing vnffgd: vnffgd_one"), + call("Removing vnffgd: vnffgd_two")] + + del_calls = [call('tacker_client_obj', vnffgd_id='vnffgd_one'), + call('tacker_client_obj', vnffgd_id='vnffgd_two')] + + cleanup.delete_vnffgds() + mock_list_vnffgds.assert_called_once_with('tacker_client_obj') + mock_log.assert_has_calls(log_calls) + mock_del_vnffgd.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.list_vims', return_value=None) + def test_delete_vims_returned_list_is_none(self, mock_list_vims): + """ + Check the proper functionality of the delete_vims + function when the returned vims list is None + """ + + self.assertIsNone(cleanup.delete_vims()) + mock_list_vims.assert_called_once_with('tacker_client_obj') + + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.openstack_utils.delete_vim') + @patch('sfc.lib.openstack_utils.list_vims') + def test_delete_vims_not_empty_list(self, + mock_list_vims, + mock_del_vim, + mock_log): + """ + Check the proper functionality of the delete_vims + function when the returned vims list is not empty + """ + + mock_list_vims.return_value = ['vim_one', 'vim_two'] + log_calls = [call("Removing vim: vim_one"), + call("Removing vim: vim_two")] + + del_calls = [call('tacker_client_obj', vim_id='vim_one'), + call('tacker_client_obj', vim_id='vim_two')] + + cleanup.delete_vims() + mock_list_vims.assert_called_once_with('tacker_client_obj') + mock_log.assert_has_calls(log_calls) + mock_del_vim.assert_has_calls(del_calls) + + @patch('sfc.lib.openstack_utils.OpenStackSFC', autospec=True) + def test_delete_untracked_security_groups(self, + mock_obj): + instance = mock_obj.return_value + cleanup.delete_untracked_security_groups() + instance.delete_all_security_groups.assert_called_once() + + @patch('sfc.lib.cleanup.delete_odl_resources') + @patch('sfc.lib.cleanup.delete_odl_ietf_access_lists') + def test_cleanup_odl(self, + mock_del_odl_ietf, + mock_del_odl_res): + resources = ['service-function-forwarder'] + + odl_res_calls = [call(self.odl_ip, self.odl_port, item) + for item in resources] + + cleanup.cleanup_odl(self.odl_ip, self.odl_port) + + mock_del_odl_res.assert_has_calls(odl_res_calls) + mock_del_odl_ietf.assert_called_once_with(self.odl_ip, self.odl_port) + + @patch('sfc.lib.openstack_utils.OpenStackSFC', autospec=True) + def test_cleanup_nsfc_objects(self, mock_os_sfc): + mock_os_sfc_ins = mock_os_sfc.return_value + cleanup.cleanup_nsfc_objects() + mock_os_sfc_ins.delete_chain.assert_called_once() + mock_os_sfc_ins.delete_port_groups.assert_called_once() + + @patch('time.sleep') + def test_cleanup_tacker_objects(self, mock_time): + + mock_dict = {'delete_vnffgs': DEFAULT, + 'delete_vnffgds': DEFAULT, + 'delete_vnfs': DEFAULT, + 'delete_vnfds': DEFAULT, + 'delete_vims': DEFAULT} + with patch.multiple('sfc.lib.cleanup', + **mock_dict) as mock_values: + cleanup.cleanup_tacker_objects() + + for key in mock_values: + mock_values[key].assert_called_once() + + mock_time.assert_called_once_with(20) + + @patch('sfc.lib.cleanup.cleanup_tacker_objects') + def test_cleanup_mano_objects_tacker(self, mock_cleanup_tacker): + cleanup.cleanup_mano_objects('tacker') + mock_cleanup_tacker.assert_called_once() + + @patch('sfc.lib.cleanup.cleanup_nsfc_objects') + def test_cleanup_mano_objects_nsfc(self, mock_cleanup_nsfc): + cleanup.cleanup_mano_objects('no-mano') + mock_cleanup_nsfc.assert_called_once() + + @patch('sfc.lib.cleanup.connection') + @patch('sfc.lib.cleanup.logger.info') + def test_delete_openstack_objects(self, mock_log, mock_conn): + """ + Checks the delete_chain method + """ + testcase_config = Mock() + conn = Mock() + mock_creator_obj_one = Mock() + mock_creator_obj_one.name = 'subnet_name' + mock_creator_obj_two = Mock() + mock_creator_obj_two.name = 'creator_name' + mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two] + + mock_conn.from_config.return_value = conn + testcase_config.subnet_name = mock_creator_obj_one.name + log_calls = [call('Deleting ' + mock_creator_obj_two.name), + call('Deleting ' + mock_creator_obj_one.name)] + + cleanup.delete_openstack_objects(testcase_config, + mock_creator_objs_list) + mock_creator_obj_one.delete.\ + assert_called_once_with(conn.session) + mock_creator_obj_two.delete.\ + assert_called_once_with(conn.session) + mock_log.assert_has_calls(log_calls) + + @patch('sfc.lib.cleanup.connection') + @patch('sfc.lib.cleanup.logger.info') + def test_delete_openstack_objects_router(self, mock_log, mock_conn): + """ + Checks the delete_chain method + """ + testcase_config = Mock() + conn = Mock() + mock_creator_obj = Mock() + mock_creator_obj.name = 'creator_name' + mock_creator_router = Mock() + mock_creator_router.name = 'router_name' + mock_creator_router.id = '1' + mock_creator_subnet = Mock() + mock_creator_subnet.name = 'subnet_name' + mock_creator_subnet.id = '2' + mock_creator_objs_list = [mock_creator_subnet, + mock_creator_router, + mock_creator_obj] + + mock_conn.from_config.return_value = conn + testcase_config.router_name = mock_creator_router.name + testcase_config.subnet_name = mock_creator_subnet.name + + conn.network.get_subnet.return_value = mock_creator_subnet + log_calls = [call('Deleting ' + mock_creator_obj.name), + call('Deleting ' + mock_creator_router.name), + call('Removing subnet from router'), + call('Deleting router'), + call('Deleting ' + mock_creator_subnet.name)] + + cleanup.delete_openstack_objects(testcase_config, + mock_creator_objs_list) + conn.network.remove_interface_from_router.\ + assert_called_once_with(mock_creator_router.id, + mock_creator_subnet.id) + conn.network.delete_router.\ + assert_called_once_with(mock_creator_router) + mock_creator_obj.delete.\ + assert_called_once_with(conn.session) + mock_creator_subnet.delete.\ + assert_called_once_with(conn.session) + mock_log.assert_has_calls(log_calls) + + @patch('sfc.lib.cleanup.connection') + @patch('sfc.lib.cleanup.logger.info') + @patch('sfc.lib.cleanup.logger.error') + def test_delete_openstack_objects_exception(self, mock_log_err, + mock_log_info, mock_conn): + """ + Check the proper functionality of the delete_openstack_objects + function when exception occurs. + """ + testcase_config = Mock() + conn = Mock() + mock_creator_obj_one = Mock() + mock_creator_obj_one.name = 'subnet_name' + mock_creator_obj_two = Mock() + mock_creator_obj_two.name = 'creator_name' + exception_one = Exception('First Boom!') + exception_two = Exception('Second Boom!') + attrs_list = [{'delete.side_effect': exception_one}, + {'delete.side_effect': exception_two}] + + mock_creator_obj_one.configure_mock(**attrs_list[0]) + mock_creator_obj_two.configure_mock(**attrs_list[1]) + + mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two] + mock_conn.from_config.return_value = conn + testcase_config.subnet_name = mock_creator_obj_one.name + + log_calls = [call('Deleting ' + mock_creator_obj_two.name), + call('Deleting ' + mock_creator_obj_one.name), + call('Unexpected error cleaning - %s', exception_two), + call('Unexpected error cleaning - %s', exception_one)] + + cleanup.delete_openstack_objects(testcase_config, + mock_creator_objs_list) + mock_creator_obj_one.delete.\ + assert_called_once_with(conn.session) + mock_creator_obj_two.delete.\ + assert_called_once_with(conn.session) + + mock_log_info.assert_has_calls(log_calls[:2]) + mock_log_err.assert_has_calls(log_calls[2:]) + + @patch('sfc.lib.cleanup.delete_untracked_security_groups') + @patch('sfc.lib.cleanup.cleanup_mano_objects') + @patch('sfc.lib.cleanup.delete_openstack_objects') + @patch('sfc.lib.cleanup.cleanup_odl') + def test_cleanup(self, + mock_cleanup_odl, + mock_del_os_obj, + mock_cleanup_mano, + mock_untr_sec_grps): + + cleanup.cleanup('testcase_config', ['creator_one', 'creator_two'], + 'mano', + self.odl_ip, + self.odl_port) + + mock_cleanup_odl.assert_called_once_with(self.odl_ip, + self.odl_port) + mock_del_os_obj.assert_called_once_with('testcase_config', + ['creator_one', 'creator_two']) + mock_cleanup_mano.assert_called_once_with('mano') + mock_untr_sec_grps.assert_called_once() + + @patch('sfc.lib.cleanup.cleanup_mano_objects') + @patch('sfc.lib.cleanup.cleanup_odl') + def test_cleanup_from_bash(self, + mock_cleanup_odl, + mock_cleanup_mano): + + cleanup.cleanup_from_bash(self.odl_ip, + self.odl_port, + 'mano') + + mock_cleanup_odl.assert_called_once_with(self.odl_ip, + self.odl_port) + mock_cleanup_mano.assert_called_once_with(mano='mano') |