diff options
Diffstat (limited to 'sfc/unit_tests/unit')
| -rw-r--r-- | sfc/unit_tests/unit/__init__.py | 0 | ||||
| -rw-r--r-- | sfc/unit_tests/unit/lib/test_cleanup.py | 469 | ||||
| -rw-r--r-- | sfc/unit_tests/unit/lib/test_odl_utils.py | 817 | ||||
| -rw-r--r-- | sfc/unit_tests/unit/lib/test_openstack_utils.py | 2504 | ||||
| -rw-r--r-- | sfc/unit_tests/unit/lib/test_test_utils.py | 543 |
5 files changed, 4333 insertions, 0 deletions
diff --git a/sfc/unit_tests/unit/__init__.py b/sfc/unit_tests/unit/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/sfc/unit_tests/unit/__init__.py 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') diff --git a/sfc/unit_tests/unit/lib/test_odl_utils.py b/sfc/unit_tests/unit/lib/test_odl_utils.py new file mode 100644 index 00000000..1dfcf1ed --- /dev/null +++ b/sfc/unit_tests/unit/lib/test_odl_utils.py @@ -0,0 +1,817 @@ +#!/usr/bin/env python + +############################################################################### +# Copyright (c) 2018 Venkata Harshavardhan Reddy Allu 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 + +from mock import call +from mock import Mock +from mock import patch + +import sfc.lib.odl_utils as odl_utils + +__author__ = "Harshavardhan Reddy <venkataharshavardhan_ven@srmuniv.edu.in>" + + +class SfcOdlUtilsTesting(unittest.TestCase): + + @patch('re.compile', autospec=True) + @patch('opnfv.utils.ovs_logger.OVSLogger', autospec=True) + def test_actual_rsps_in_compute(self, mock_ovs_log, mock_compile): + """ + Checks the proper functionality of actual_rsps_in_compute + function + """ + + match_calls = [call('msg_1'), call('msg_2')] + + mf = Mock() + mf.group.side_effect = ['msg_p_1', 'msg_p_2'] + mock_compile.return_value.match.side_effect = [mf, None] + mock_ovs_log.ofctl_dump_flows.return_value = '\nflow_rep\nmsg_1\nmsg_2' + + result = odl_utils.actual_rsps_in_compute(mock_ovs_log, 'compute_ssh') + + self.assertEqual(['msg_p_1|msg_p_2'], result) + mock_compile.return_value.match.assert_has_calls(match_calls) + mock_ovs_log.ofctl_dump_flows.assert_called_once_with('compute_ssh', + 'br-int', '101') + + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.get_rsp', autospec=True) + @patch('sfc.lib.odl_utils.get_odl_acl_list', autospec=True) + @patch('sfc.lib.odl_utils.get_rsps_from_netvirt_acl_actions', + autospec=True) + def test_get_active_rsps_on_ports(self, + mock_rsps_from_netvirt_acl_actions, + mock_odl_acl_list, + mock_get_rsp, + mock_log): + """ + Checks the proper functionality of get_active_rsps_on_ports + function + """ + + log_calls = [call('ACL acl_obj_one does not have an ACE')] + + port_one = Mock() + port_two = Mock() + port_one.id = 's_p' + port_two.id = 'd_p' + neutron_ports = [port_one, port_two] + + mock_rsps_from_netvirt_acl_actions.return_value = ['rsp_obj_one', + 'rsp_obj_two'] + + mock_get_rsp.side_effect = [{'of-matches': ['of-match-one'], + 'reverse-path': 'r-path-one'}, + {'of-matches': ['of-match-two']}] + + mock_odl_acl_list.return_value = {'access-lists': {'acl': [ + {'acl-name': 'acl_obj_one', + 'access-list-entries': {'ace': []}}, + {'acl-name': 'acl_obj_two', + 'access-list-entries': {'ace': [{'matches': { + 'destination-port-range': None}}]}}, + {'acl-name': 'acl_obj_three', + 'access-list-entries': {'ace': [{'matches': { + 'destination-port-range': {'lower-port': 22}, + 'netvirt-sfc-acl:source-port-uuid': 's_p_uuid', + 'netvirt-sfc-acl:destination-port-uuid': 'd_p_uuid'}}]}}, + {'acl-name': 'acl_obj_four', + 'access-list-entries': {'ace': [{'matches': { + 'destination-port-range': {'lower-port': 22}, + 'netvirt-sfc-acl:source-port-uuid': 's_p', + 'netvirt-sfc-acl:destination-port-uuid': 'd_p'}, + 'actions': 'm_actions'}]}}]}} + + expected = [{'of-matches': ['of-match-two', 'tp_dst=22']}, + {'of-matches': ['of-match-one', 'tp_src=22'], + 'reverse-path': 'r-path-one'}] + + result = odl_utils.get_active_rsps_on_ports('odl_ip', + 'odl_port', + neutron_ports) + + self.assertEqual(sorted(expected), sorted(result)) + mock_log.warn.assert_has_calls(log_calls) + mock_rsps_from_netvirt_acl_actions.assert_called_once_with('odl_ip', + 'odl_port', + 'm_actions') + + @patch('sfc.lib.odl_utils.get_odl_resource_elem', autospec=True) + def test_get_rsps_from_netvirt_acl_actions(self, mock_odl_resource_elem): + """ + Checks the proper functionality of get_rsps_from_netvirt_acl_actions + function + """ + + netv = {'netvirt-sfc-acl:rsp-name': 'rsp-name', + 'netvirt-sfc-acl:sfp-name': 'sfp-name'} + + sfp_state = {'sfp-rendered-service-path': [{'name': 'sfp-rsp-one'}, + {'name': 'sfp-rsp-two'}]} + + mock_odl_resource_elem.return_value = sfp_state + rsp_names = ['rsp-name', 'sfp-rsp-one', 'sfp-rsp-two'] + + result = odl_utils.get_rsps_from_netvirt_acl_actions('odl_ip', + 'odl_port', + netv) + self.assertEqual(rsp_names, result) + mock_odl_resource_elem.assert_called_once_with('odl_ip', 'odl_port', + 'service-function-path-' + 'state', 'sfp-name', + datastore='operational') + + @patch('sfc.lib.odl_utils.get_odl_resource_elem', + autospec=True, return_value='mocked_rsp') + def test_get_rsp(self, mock_odl_resource_elem): + """ + Checks the proper functionality of get_rsp + function + """ + + result = odl_utils.get_rsp('odl_ip', 'odl_port', 'rsp_name') + self.assertEqual('mocked_rsp', result) + mock_odl_resource_elem.assert_called_once_with('odl_ip', 'odl_port', + 'rendered-service-path', + 'rsp_name', + datastore='operational') + + @patch('sfc.lib.odl_utils.get_active_rsps_on_ports', autospec=True) + def test_promised_rsps_in_compute(self, mock_active_rsps_on_ports): + """ + Checks the proper functionality of propmised_rsps_in_compute + function + """ + + mock_active_rsps_on_ports.return_value = [ + {'of-matches': {'one': 'one'}, 'path-id': 1}, + {'of-matches': {'two': 'two'}, 'path-id': 2}] + + result = odl_utils.promised_rsps_in_compute('odl_ip', 'odl_port', + 'compute_ports') + + self.assertEqual(['0x1|one', '0x2|two'], result) + mock_active_rsps_on_ports.assert_called_once_with('odl_ip', 'odl_port', + 'compute_ports') + + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('time.time', autospec=True, side_effect=[1, 2]) + def test_timethis(self, + mock_time, + mock_log): + """ + Checks the proper functionality of timethis + function + """ + + expected = ('mock_this', '1') + log_calls = [call("mock_func(*('mock',), **{'name': 'this'}) " + "took: 1 sec")] + + @odl_utils.timethis + def mock_func(msg, name=''): + return msg+'_'+name + + result = mock_func('mock', name='this') + self.assertEqual(result, expected) + mock_log.info.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.get_odl_items', autospec=True) + @patch('sfc.lib.odl_utils.promised_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.os_sfc_utils.get_tacker_items', autospec=True) + def test_wait_for_classification_rules_rsps_not_configured( + self, mock_get_tacker_items, mock_promised_rsps_in_compute, + mock_get_odl_items, mock_find_compute, mock_log, mock_sleep): + """ + Checks the proper functionality of wait_for_classification_rules + function when rsps are not configured in ODL + """ + + log_calls = [call("Error when waiting for classification rules: " + "RSPs not configured in ODL")] + + mock_find_compute.return_value = 'mock_compute' + mock_promised_rsps_in_compute.return_value = None + + odl_utils.wait_for_classification_rules('ovs_logger', + 'compute_nodes', + 'odl_ip', + 'odl_port', + 'compute_name', + 'neutron_ports') + mock_promised_rsps_in_compute.assert_called_with('odl_ip', + 'odl_port', + 'neutron_ports') + assert mock_promised_rsps_in_compute.call_count == 10 + mock_find_compute.assert_called_once_with('compute_name', + 'compute_nodes') + mock_sleep.assert_called_with(3) + assert mock_sleep.call_count == 9 + mock_get_tacker_items.assert_called_once_with() + mock_get_odl_items.assert_called_once_with('odl_ip', 'odl_port') + mock_log.error.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.actual_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.promised_rsps_in_compute', autospec=True) + def test_wait_for_classification_rules_timeout_not_updated( + self, mock_promised_rsps_in_compute, mock_actual_rsps_in_compute, + mock_find_compute, mock_log, mock_sleep): + """ + Checks the proper functionality of wait_for_classification_rules + function when classification rules are not updated in a given timeout + """ + + log_calls = [call("Timeout but classification rules are not updated"), + call("RSPs in ODL Operational DataStore" + "for compute 'compute_name':"), + call("['compute|rsps']"), + call("RSPs in compute nodes:"), + call("[]")] + + mock_compute = Mock() + mock_compute.ssh_client = 'mock_ssh_client' + mock_find_compute.return_value = mock_compute + mock_actual_rsps_in_compute.return_value = [] + mock_promised_rsps_in_compute.return_value = ['compute|rsps'] + + odl_utils.wait_for_classification_rules('ovs_logger', + 'compute_nodes', + 'odl_ip', + 'odl_port', + 'compute_name', + 'neutron_ports', + timeout=2) + mock_find_compute.assert_called_once_with('compute_name', + 'compute_nodes') + mock_log.error.assert_has_calls(log_calls[:1]) + mock_log.info.assert_has_calls(log_calls[1:]) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.actual_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.promised_rsps_in_compute', autospec=True) + def test_wait_for_classification_rules_updated( + self, mock_promised_rsps_in_compute, mock_actual_rsps_in_compute, + mock_find_compute, mock_log, mock_sleep): + """ + Checks the proper functionality of wait_for_classification_rules + function when classification rules are not updated in a given timeout + """ + + log_calls = [call("RSPs in ODL Operational DataStore" + "for compute 'compute_name':"), + call("['compute|rsps']"), + call("RSPs in compute nodes:"), + call("['compute|rsps']"), + call("Classification rules were updated")] + mock_compute = Mock() + mock_compute.ssh_client = 'mock_ssh_client' + mock_find_compute.return_value = mock_compute + mock_actual_rsps_in_compute.return_value = ['compute|rsps'] + mock_promised_rsps_in_compute.return_value = ['compute|rsps'] + + odl_utils.wait_for_classification_rules('ovs_logger', + 'compute_nodes', + 'odl_ip', + 'odl_port', + 'compute_name', + 'neutron_ports', + timeout=2) + mock_log.info.assert_has_calls(log_calls) + + @patch('re.search', autospec=True) + @patch('ConfigParser.RawConfigParser', autospec=True) + @patch('os.getcwd', autospec=True, return_value='/etc') + @patch('os.path.join', autospec=True, return_value='/etc/ml2_conf.ini') + def test_get_odl_ip_port(self, mock_join, + mock_getcwd, + mock_rawconfigparser, + mock_search): + """ + Checks the proper functionality of get_odl_ip_port + function + """ + + cmd_calls = [call('pwd'), + call('sudo cp /etc/neutron/plugins/ml2/ml2_conf.ini ' + '/etc/'), + call('sudo chmod 777 /etc/ml2_conf.ini')] + + n1 = Mock() + n2 = Mock() + nodes = [n1, n2] + mock_rawconfigparser.return_value.get.return_value = 'config' + mock_search.return_value.group.return_value = 'odl_ip:odl_port' + n1.run_cmd.side_effect = ['/etc', '', ''] + + result = odl_utils.get_odl_ip_port(nodes) + self.assertEqual(('odl_ip', 'odl_port'), result) + n1.run_cmd.assert_has_calls(cmd_calls) + n1.is_controller.assert_called_once_with() + mock_getcwd.assert_called_once_with() + mock_join.assert_called_once_with('/etc', 'ml2_conf.ini') + n1.get_file.assert_called_once_with('/etc/ml2_conf.ini', + '/etc/ml2_conf.ini') + mock_rawconfigparser.return_value.read.assert_called_once_with( + '/etc/ml2_conf.ini') + mock_rawconfigparser.return_value.get.assert_called_with( + 'ml2_odl', 'url') + mock_search.assert_called_once_with(r'[0-9]+(?:\.[0-9]+){3}\:[0-9]+', + 'config') + + @patch('re.search', autospec=True) + @patch('ConfigParser.RawConfigParser', autospec=True) + @patch('os.getcwd', autospec=True, return_value='/etc') + @patch('os.path.join', autospec=True, return_value='/etc/ml2_conf.ini') + def test_get_odl_username_password(self, mock_join, + mock_getcwd, + mock_rawconfigparser, + mock_search): + """ + Check the proper functionality of get odl_username_password + function + """ + + mock_rawconfigparser.return_value.get.return_value = 'odl_username' + result = odl_utils.get_odl_username_password() + self.assertEqual(('odl_username'), result[0]) + mock_getcwd.assert_called_once_with() + mock_join.assert_called_once_with('/etc', 'ml2_conf.ini') + mock_rawconfigparser.return_value.read.assert_called_once_with( + '/etc/ml2_conf.ini') + mock_rawconfigparser.return_value.get.return_value = 'odl_password' + result = odl_utils.get_odl_username_password() + self.assertEqual(('odl_password'), result[1]) + + def test_pluralize(self): + """ + Checks the proper functionality of pluralize + function + """ + + result = odl_utils.pluralize('service-function-path') + self.assertEqual('service-function-paths', result) + + def test_get_module(self): + """ + Checks the proper functionality of get_module + function + """ + + result = odl_utils.get_module('service-function-path') + self.assertEqual('service-function-path', result) + + @patch('sfc.lib.odl_utils.get_module', + autospec=True, return_value='mocked_module') + @patch('sfc.lib.odl_utils.pluralize', + autospec=True, return_value='resources') + def test_format_odl_resource_list_url(self, mock_plularize, + mock_get_module): + """ + Checks the proper functionality of format_odl_resource_list_url + function + """ + + result = odl_utils.format_odl_resource_list_url('odl_ip', + 'odl_port', + 'resource') + formatted_url = ('http://admin:admin@odl_ip:' + 'odl_port/restconf/config/mocked_module:' + 'resources') + self.assertEqual(formatted_url, result) + mock_plularize.assert_called_once_with('resource') + mock_get_module.assert_called_once_with('resource') + + @patch('sfc.lib.odl_utils.format_odl_resource_list_url', + autospec=True, return_value='list_u/r/l') + def test_format_odl_resource_elem_url(self, mock_odl_resource_list_url): + """ + Checks the proper functionality of format_odl_resource_elem_url + function + """ + + result = odl_utils.format_odl_resource_elem_url('odl_ip', 'odl_port', + 'resource', + 'elem_name') + formatted_url = ('list_u/r/l/resource/elem_name') + self.assertEqual(formatted_url, result) + mock_odl_resource_list_url.assert_called_once_with('odl_ip', + 'odl_port', + 'resource', + 'config') + + @patch('sfc.lib.odl_utils.pluralize', + autospec=True, return_value='resources') + def test_odl_resource_list_names_returns_empty_list(self, mock_plularize): + """ + Checks the proper functionality of odl_resource_list_names + function when resources are empty + """ + + resource_json = {'resources': {}} + result = odl_utils.odl_resource_list_names('resource', resource_json) + self.assertEqual([], result) + + @patch('sfc.lib.odl_utils.pluralize', + autospec=True, return_value='resources') + def test_odl_resource_list_names(self, mock_plularize): + """ + Checks the proper functionality of odl_resource_list_names + function + """ + + resource_json = {'resources': {'resource': [{'name': 'resource_one'}, + {'name': 'resource_two'}]}} + result = odl_utils.odl_resource_list_names('resource', resource_json) + self.assertEqual(['resource_one', 'resource_two'], result) + + @patch('requests.get', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_resource_list_url', autospec=True) + def test_get_odl_resource_list(self, + mock_odl_resource_list_url, + mock_get): + """ + Checks the proper functionality of get_odl_resource_list + function + """ + + mock_odl_resource_list_url.return_value = 'u/r/l' + mock_get.return_value.json.return_value = {'key': 'value'} + + result = odl_utils.get_odl_resource_list('odl_ip', + 'odl_port', + 'resource') + + self.assertEqual({'key': 'value'}, result) + mock_odl_resource_list_url.assert_called_once_with('odl_ip', + 'odl_port', + 'resource', + datastore='config') + mock_get.assert_called_once_with('u/r/l') + + @patch('requests.get', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_resource_elem_url', autospec=True) + def test_get_odl_resource_elem(self, + mock_odl_resource_elem_url, + mock_get): + """ + Checks the proper functionality of get_odl_resource_elem + function + """ + + mock_response = Mock() + mock_response.get.return_value = ['elem_one', 'elem_two'] + mock_get.return_value.json.return_value = mock_response + mock_odl_resource_elem_url.return_value = 'u/r/l' + + result = odl_utils.get_odl_resource_elem( + 'odl_ip', 'odl_port', 'resource', 'elem_name') + + self.assertEqual('elem_one', result) + mock_odl_resource_elem_url.assert_called_once_with( + 'odl_ip', 'odl_port', 'resource', 'elem_name', 'config') + mock_get.assert_called_once_with('u/r/l') + mock_response.get.assert_called_once_with('resource', [{}]) + + @patch('requests.delete', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_resource_elem_url', + autospec=True, return_value='u/r/l') + def test_delete_odl_resource_elem(self, + mock_odl_resource_elem_url, + mock_delete): + """ + Checks the proper functionality of delete_odl_resource_elem + function + """ + + odl_utils.delete_odl_resource_elem('odl_ip', 'odl_port', 'resource', + 'elem_name') + + mock_odl_resource_elem_url('odl_ip', 'odl_port', 'resource', + 'elem_name', 'config') + mock_delete.assert_called_once_with('u/r/l') + + def test_odl_acl_types_names_returns_empty_list(self): + """ + Checks the proper functionality of odl_acl_types_names + function when access lists are empty + """ + + acl_json = {'access-lists': {}} + result = odl_utils.odl_acl_types_names(acl_json) + self.assertEqual([], result) + + def test_odl_acl_types_names(self): + """ + Checks the proper functionality of odl_acl_types_names + function + """ + + acl_json = {'access-lists': {'acl': [{'acl-type': 'type-one', + 'acl-name': 'name-one'}, + {'acl-type': 'type-two', + 'acl-name': 'name-two'}]}} + acl_types = [('type-one', 'name-one'), + ('type-two', 'name-two')] + + result = odl_utils.odl_acl_types_names(acl_json) + self.assertEqual(acl_types, result) + + def test_format_odl_acl_list_url(self): + """ + Checks the proper functionality of format_odl_acl_list_url + function + """ + + formatted_url = ('http://admin:admin@odl_ip:odl_port/restconf/config/' + 'ietf-access-control-list:access-lists') + result = odl_utils.format_odl_acl_list_url('odl_ip', 'odl_port') + self.assertEqual(formatted_url, result) + + @patch('json.dumps', + autospec=True, return_value='{\n "key": "value"\n}') + def test_improve_json_layout(self, mock_dumps): + """ + Checks the proper functionality of improve_json_layout + function + """ + + result = odl_utils.improve_json_layout({'key': 'value'}) + + self.assertEqual('{\n "key": "value"\n}', result) + mock_dumps.assert_called_once_with({'key': 'value'}, + indent=4, + separators=(',', ': ')) + + @patch('requests.get', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_acl_list_url', + autospec=True, return_value='acl_list_u/r/l') + @patch('sfc.lib.odl_utils.improve_json_layout', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_resource_list_url', autospec=True) + def test_get_odl_items(self, + mock_odl_resource_list_url, + mock_json_layout, + mock_odl_acl_list_url, + mock_log, + mock_get): + """ + Checks the proper functionality of get_odl_items + function + """ + + log_calls = [call('Configured ACLs in ODL: r_acl_j_s_o_n'), + call('Configured SFs in ODL: r_sf_j_s_o_n'), + call('Configured SFFs in ODL: r_sff_j_s_o_n'), + call('Configured SFCs in ODL: r_sfc_j_s_o_n'), + call('Configured RSPs in ODL: r_sp_j_s_o_n')] + + resource_list_url_calls = [call('odl_ip', 'odl_port', + 'service-function'), + call('odl_ip', 'odl_port', + 'service-function-forwarder'), + call('odl_ip', 'odl_port', + 'service-function-chain'), + call('odl_ip', 'odl_port', + 'rendered-service-path', + datastore='operational')] + + resource_list_urls = ['sf_list_u/r/l', 'sff_list_u/r/l', + 'sfc_list_u/r/l', 'rsp_list_u/r/l'] + + get_calls = [call(url) for url in resource_list_urls] + + mock_odl_resource_list_url.side_effect = resource_list_urls + + mock_get.return_value.json.side_effect = ['r_acl_json', 'r_sf_json', + 'r_sff_json', 'r_sfc_json', + 'r_rsp_json'] + + mock_json_layout.side_effect = ['r_acl_j_s_o_n', 'r_sf_j_s_o_n', + 'r_sff_j_s_o_n', 'r_sfc_j_s_o_n', + 'r_sp_j_s_o_n'] + + odl_utils.get_odl_items('odl_ip', 'odl_port') + + mock_odl_acl_list_url.assert_called_once_with('odl_ip', 'odl_port') + mock_odl_resource_list_url.assert_has_calls(resource_list_url_calls) + mock_get.assert_has_calls(get_calls, any_order=True) + mock_log.debug.assert_has_calls(log_calls) + + @patch('requests.get', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_acl_list_url', autospec=True) + def test_get_odl_acl_list(self, + mock_acl_list_url, + mock_get): + """ + Checks the proper functionality of get_odl_acl_list + function + """ + + mock_acl_list_url.return_value = 'acl_list/url' + mock_get.return_value.json.return_value = {'key': 'value'} + result = odl_utils.get_odl_acl_list('odl_ip', 'odl_port') + mock_acl_list_url.assert_called_once_with('odl_ip', 'odl_port') + mock_get.assert_called_once_with('acl_list/url') + self.assertEqual({'key': 'value'}, result) + + @patch('requests.delete', autospec=True) + @patch('sfc.lib.odl_utils.format_odl_acl_list_url', autospec=True) + def test_delete_odl_acl(self, + mock_acl_list_url, + mock_delete): + """ + Checks the proper functionality of delete_odl_acl + function + """ + + mock_acl_list_url.return_value = 'acl_list/url' + + odl_utils.delete_odl_acl('odl_ip', 'odl_port', 'acl_type', 'acl_name') + + mock_acl_list_url.assert_called_once_with('odl_ip', 'odl_port') + mock_delete.assert_called_once_with( + 'acl_list/url/acl/acl_type/acl_name') + + @patch('sfc.lib.odl_utils.delete_odl_acl', autospec=True) + def test_delete_acl(self, mock_delete_odl_acl): + """ + Checks the proper fucntionality of delete_acl + function + """ + + odl_utils.delete_acl('clf_name', 'odl_ip', 'odl_port') + mock_delete_odl_acl.assert_called_once_with( + 'odl_ip', + 'odl_port', + 'ietf-access-control-list:ipv4-acl', + 'clf_name') + + @patch('sfc.lib.odl_utils.logger', autospec=True) + def test_find_compute_raises_exception(self, mock_log): + """ + Checks the proper functionality of find_compute + function when compute was not found in the client + """ + + ErrorMSG = 'No compute, where the client is, was found' + compute_node_one = Mock() + compute_node_two = Mock() + compute_nodes = [compute_node_one, compute_node_two] + compute_node_one.name = 'compute_one' + compute_node_two.name = 'compute_two' + + with self.assertRaises(Exception) as cm: + odl_utils.find_compute('compute_client', compute_nodes) + + self.assertEqual(ErrorMSG, cm.exception.message) + mock_log.debug.assert_called_once_with(ErrorMSG) + + @patch('sfc.lib.odl_utils.logger', autospec=True) + def test_find_compute(self, mock_log): + """ + Checks the proper functionality of find_compute + function when compute was not found in the client + """ + + compute_node_one = Mock() + compute_node_two = Mock() + compute_nodes = [compute_node_one, compute_node_two] + compute_node_one.name = 'compute_one' + compute_node_two.name = 'compute_two' + + result = odl_utils.find_compute('compute_two', compute_nodes) + + self.assertEqual(compute_node_two, result) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.get_active_rsps_on_ports', autospec=True) + def test_check_vnffg_deletion_returns_false_rsps_still_active( + self, mock_active_rsps_on_ports, + mock_log, mock_sleep): + """ + Checks the proper functionality of check_vnffg_deletion + function to verify that it returns false on the given condition + """ + + log_calls = [call('RSPs are still active in the MD-SAL')] + mock_active_rsps_on_ports.return_value = True + result = odl_utils.check_vnffg_deletion('odl_ip', 'odl_port', + 'ovs_logger', 'neutron_ports', + 'compute_client_name', + 'compute_nodes', retries=1) + self.assertFalse(result) + mock_active_rsps_on_ports.assert_called_once_with('odl_ip', 'odl_port', + 'neutron_ports') + mock_sleep.assert_called_once_with(3) + mock_log.debug.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.actual_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.get_active_rsps_on_ports', autospec=True) + def test_check_vnffg_deletion_returns_false_error_getting_compute( + self, mock_active_rsps_on_ports, mock_actual_rsps, + mock_find_compute, mock_log, mock_sleep): + """ + Checks the proper functionality of check_vnffg_deletion + function to verify that it returns false on the given condition + """ + + log_calls = [call('There was an error getting the compute: ErrorMSG')] + mock_compute = Mock() + mock_compute.ssh_client = 'mock_ssh_client' + mock_find_compute.side_effect = [Exception('ErrorMSG'), mock_compute] + mock_active_rsps_on_ports.side_effect = [True, False] + result = odl_utils.check_vnffg_deletion('odl_ip', 'odl_port', + 'ovs_logger', 'neutron_ports', + 'compute_client_name', + 'compute_nodes', retries=2) + self.assertFalse(result) + mock_sleep.assert_called_once_with(3) + mock_find_compute.assert_called_once_with('compute_client_name', + 'compute_nodes') + mock_log.debug.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.actual_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.get_active_rsps_on_ports', autospec=True) + def test_check_vnffg_deletion_returns_false_classification_flow_in_compute( + self, mock_active_rsps_on_ports, mock_actual_rsps, + mock_find_compute, mock_log, mock_sleep): + """ + Checks the proper functionality of check_vnffg_deletion + function to verify that it returns false on the given condition + """ + + log_calls = [call('Classification flows still in the compute')] + mock_compute = Mock() + mock_compute.ssh_client = 'mock_ssh_client' + mock_find_compute.return_value = mock_compute + mock_actual_rsps.side_effect = [True, True] + mock_active_rsps_on_ports.side_effect = [True, False] + result = odl_utils.check_vnffg_deletion('odl_ip', 'odl_port', + 'ovs_logger', 'neutron_ports', + 'compute_client_name', + 'compute_nodes', retries=2) + self.assertFalse(result) + mock_actual_rsps.assert_called_with('ovs_logger', 'mock_ssh_client') + mock_sleep.assert_called_with(3) + mock_find_compute.assert_called_once_with('compute_client_name', + 'compute_nodes') + assert mock_sleep.call_count == 3 + mock_log.debug.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.odl_utils.logger', autospec=True) + @patch('sfc.lib.odl_utils.find_compute', autospec=True) + @patch('sfc.lib.odl_utils.actual_rsps_in_compute', autospec=True) + @patch('sfc.lib.odl_utils.get_active_rsps_on_ports', autospec=True) + def test_check_vnffg_deletion_returns_true(self, + mock_active_rsps_on_ports, + mock_actual_rsps, + mock_find_compute, + mock_log, mock_sleep): + """ + Checks the proper functionality of check_vnffg_deletion + function to verify that it returns true + """ + + mock_compute = Mock() + mock_compute.ssh_client = 'mock_ssh_client' + mock_active_rsps_on_ports.side_effect = [True, False] + + mock_actual_rsps.side_effect = [True, False] + + mock_find_compute.return_value = mock_compute + + result = odl_utils.check_vnffg_deletion('odl_ip', 'odl_port', + 'ovs_logger', 'neutron_ports', + 'compute_client_name', + 'compute_nodes', retries=2) + self.assertTrue(result) + mock_find_compute.assert_called_once_with('compute_client_name', + 'compute_nodes') + assert mock_sleep.call_count == 2 + mock_log.assert_not_called() diff --git a/sfc/unit_tests/unit/lib/test_openstack_utils.py b/sfc/unit_tests/unit/lib/test_openstack_utils.py new file mode 100644 index 00000000..bdd53d36 --- /dev/null +++ b/sfc/unit_tests/unit/lib/test_openstack_utils.py @@ -0,0 +1,2504 @@ +#!/usr/bin/env python + +############################################################################### +# Copyright (c) 2018 Venkata Harshavardhan Reddy Allu 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 + +from mock import call +from mock import Mock +from mock import patch +from mock import mock_open +from mock import DEFAULT +from mock import MagicMock + +import sfc.lib.openstack_utils as os_sfc_utils +from tackerclient.v1_0 import client as tacker_client + +__author__ = "Harshavardhan Reddy <venkataharshavardhan_ven@srmuniv.edu.in>" + + +class SfcOpenStackUtilsTesting(unittest.TestCase): + + def setUp(self): + self.patcher1 = patch.object(os_sfc_utils.constants, + 'ENV_FILE', autospec=True) + self.patcher2 = patch.object(os_sfc_utils.openstack_tests, + 'get_credentials', autospec=True) + self.patcher3 = patch.object(os_sfc_utils.nova_utils, + 'nova_client', autospec=True) + self.patcher4 = patch.object(os_sfc_utils.neutron_utils, + 'neutron_client', autospec=True) + self.patcher5 = patch.object(os_sfc_utils.heat_utils, + 'heat_client', autospec=True) + self.patcher6 = patch.object(os_sfc_utils.keystone_utils, + 'keystone_client', autospec=True) + self.patcher7 = patch.object(os_sfc_utils.connection, + 'from_config', autospec=True,) + self.patcher8 = patch.object(os_sfc_utils.neutronclient, + 'Client', autospec=True,) + + self.env_file = self.patcher1.start().return_value + self.os_creds = self.patcher2.start().return_value + self.nova = self.patcher3.start().return_value + self.neutron = self.patcher4.start().return_value + self.heat = self.patcher5.start().return_value + self.keystone = self.patcher6.start().return_value + self.conn = self.patcher7.start().return_value + self.neutron_client = self.patcher8.start().return_value + + self.os_sfc = os_sfc_utils.OpenStackSFC() + + def tearDown(self): + self.patcher1.stop() + self.patcher2.stop() + self.patcher3.stop() + self.patcher4.stop() + self.patcher5.stop() + self.patcher6.stop() + self.patcher7.stop() + self.patcher8.stop() + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('os.environ', {'OS_NETWORK_API_VERSION': '1'}) + def test_get_neutron_client_version(self, + mock_log): + """ + Checks the proper functionality of get_neutron_client_version + """ + log_calls = [call("OS_NETWORK_API_VERSION is 1")] + result = self.os_sfc.get_neutron_client_version() + assert result == '1' + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_register_glance_image_already_exists(self, + mock_log): + """ + Checks the proper functionality of register_glance_image + function when the image is local + """ + image_obj = Mock() + image_obj.name = 'name' + log_calls = [call('Registering the image...'), + call('Image ' + image_obj.name + ' already exists.')] + + self.conn.image.find_image.return_value = image_obj + result = self.os_sfc.register_glance_image('name', + 'url', + 'img_format', + 'public') + + self.conn.image.find_image.assert_called_once_with(image_obj.name) + + assert result is image_obj + + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch("__builtin__.open", autospec=True) + def test_register_glance_image_is_local(self, + mock_open_fn, + mock_log): + """ + Checks the proper functionality of register_glance_image + function when the image is local + """ + log_calls = [call('Registering the image...'), + call('Image created')] + + image_obj_None = None + image_obj_name = 'name' + image_obj = Mock() + mocked_file = mock_open(read_data='url').return_value + mock_open_fn.return_value = mocked_file + + self.conn.image.find_image.return_value = image_obj_None + self.conn.image.upload_image.return_value = image_obj + result = self.os_sfc.register_glance_image('name', + 'url', + 'img_format', + 'public') + assert result is image_obj + + self.conn.image.find_image.assert_called_once_with(image_obj_name) + + self.conn.image.upload_image.\ + assert_called_once_with(name='name', + disk_format='img_format', + data='url', + is_public='public', + container_format='bare') + + self.assertEqual([image_obj], self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.urllib2.urlopen', autospec=True) + def test_register_glance_image_is_not_local(self, + mock_urlopen, + mock_log): + """ + Checks the proper functionality of register_glance_image + function when the image is not local + """ + log_calls = [call('Registering the image...'), + call('Downloading image'), + call('Image created')] + + image_obj_None = None + image_obj_name = 'name' + image_obj = Mock() + mock_file = Mock() + mock_file.read.side_effect = ['http://url'] + mock_urlopen.return_value = mock_file + + self.conn.image.find_image.return_value = image_obj_None + self.conn.image.upload_image.return_value = image_obj + + result = self.os_sfc.register_glance_image('name', + 'http://url', + 'img_format', + 'public') + + assert result is image_obj + + self.conn.image.find_image.assert_called_once_with(image_obj_name) + + self.conn.image.upload_image.\ + assert_called_once_with(name='name', + disk_format='img_format', + data='http://url', + is_public='public', + container_format='bare') + + self.assertEqual([image_obj], self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_flavour(self, + mock_log): + """ + Checks the proper functionality of create_flavor + function + """ + + mock_openstack_flavor_ins = self.conn.compute.\ + create_flavor.return_value + log_calls = [call('Creating flavor...')] + + result = self.os_sfc.create_flavor('name', + 'ram', + 'disk', + 'vcpus') + assert result is mock_openstack_flavor_ins + self.assertEqual([mock_openstack_flavor_ins], + self.os_sfc.creators) + self.conn.compute.create_flavor.\ + assert_called_once_with(name='name', + ram='ram', + disk='disk', + vcpus='vcpus') + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.env.get', autospec=True) + def test_create_network_infrastructure(self, mock_env_get, mock_log): + log_calls = [call('Creating Networks...'), + call('Creating Router...')] + network_obj = Mock() + network_obj.id = '1' + subnet_obj = Mock() + subnet_obj.id = '2' + ext_network_obj = Mock() + ext_network_obj.id = '3' + router_obj = Mock() + router_obj.id = '4' + + self.conn.network.create_network.return_value = network_obj + self.conn.network.create_subnet.return_value = subnet_obj + self.conn.network.find_network.return_value = ext_network_obj + self.conn.network.create_router.return_value = router_obj + self.conn.network.get_router.return_value = router_obj + mock_env_get.return_value = 'ext_net_name' + + expected = (network_obj, router_obj) + result = self.os_sfc.create_network_infrastructure('net_name', + 'sn_name', + 'subnet_cidr', + 'router_name') + self.conn.network.create_network.\ + assert_called_once_with(name='net_name') + self.conn.network.create_subnet.\ + assert_called_once_with(name='sn_name', cidr='subnet_cidr', + network_id=network_obj.id, ip_version='4') + self.conn.network.find_network.\ + assert_called_once_with('ext_net_name') + self.conn.network.create_router.\ + assert_called_once_with(name='router_name') + self.conn.network.add_interface_to_router.\ + assert_called_once_with(router_obj.id, subnet_id=subnet_obj.id) + self.conn.network.update_router.\ + assert_called_once_with( + router_obj.id, + external_gateway_info={'network_id': ext_network_obj.id}) + self.conn.network.get_router.assert_called_once_with(router_obj.id) + + self.assertEqual(expected, result) + self.assertEqual([network_obj, subnet_obj, router_obj], + self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_security_group(self, + mock_log): + """ + Checks the proper functionality of create_security_group + function + """ + log_calls = [call('Creating the security groups...')] + sec_group_obj = Mock() + sec_group_obj.id = '1' + + self.conn.network.create_security_group.return_value = sec_group_obj + + result = self.os_sfc.create_security_group('sec_grp_name') + assert result is sec_group_obj + + self.conn.network.create_security_group.\ + assert_called_once_with(name='sec_grp_name') + + pc_calls = [call(security_group_id=sec_group_obj.id, + direction='ingress', + protocol='icmp'), + call(security_group_id=sec_group_obj.id, + direction='ingress', + protocol='tcp', + port_range_min=22, + port_range_max=22), + call(security_group_id=sec_group_obj.id, + direction='ingress', + protocol='tcp', + port_range_min=80, + port_range_max=80)] + + self.conn.network.create_security_group_rule.\ + assert_has_calls(pc_calls) + + self.assertEqual([sec_group_obj], self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_instance_port_security_false(self, + mock_log): + """ + Checks the proper functionality of create_instance + function + """ + + keypair_obj = Mock() + keypair_obj.name = 'keypair_name' + flavor_obj = Mock() + flavor_obj.id = '1' + port_obj1 = Mock() + port_obj1.id = '2' + port_obj2 = Mock() + port_obj2.id = '3' + instance_obj = Mock() + instance_obj.name = 'instance_name' + secgrp = Mock() + secgrp.name = 'sec_grp' + secgrp.id = '4' + img_cre = Mock() + img_cre.id = '5' + network = Mock() + network.id = '6' + ports = ['port1', 'port2'] + port_security = False + + log_calls = [call('Creating Key Pair vm_name...'), + call('Creating Port ' + str(ports) + '...'), + call('Creating the instance vm_name...'), + call('Waiting for instance_name to become Active'), + call('instance_name is active')] + + self.conn.compute.create_keypair.return_value = keypair_obj + self.conn.compute.find_flavor.return_value = flavor_obj + self.conn.network.create_port.side_effect = [port_obj1, port_obj2] + self.conn.compute.create_server.return_value = instance_obj + + port_obj_list = [port_obj1, port_obj2] + + expected = (instance_obj, port_obj_list) + result = self.os_sfc.create_instance('vm_name', + 'flavor_name', + img_cre, + network, + secgrp, + 'av_zone', + ports, + port_security=port_security) + self.assertEqual(expected, result) + + pc_calls = [call(name=ports[0], + is_port_security_enabled=port_security, + network_id=network.id), + call(name=ports[1], + is_port_security_enabled=port_security, + network_id=network.id)] + + self.conn.compute.create_keypair.\ + assert_called_once_with(name='vm_name' + "_keypair") + + self.conn.compute.find_flavor.assert_called_once_with('flavor_name') + + self.conn.network.create_port.\ + assert_has_calls(pc_calls) + + self.conn.compute.create_server.\ + assert_called_once_with(name='vm_name', + image_id=img_cre.id, + flavor_id=flavor_obj.id, + networks=[{"port": port_obj1.id}, + {"port": port_obj2.id}], + key_name=keypair_obj.name, + availability_zone='av_zone') + + self.conn.compute.wait_for_server.\ + assert_called_once_with(instance_obj) + + self.assertEqual([keypair_obj, port_obj1, port_obj2, instance_obj], + self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_instance(self, + mock_log): + """ + Checks the proper functionality of create_instance + function + """ + + keypair_obj = Mock() + keypair_obj.name = 'keypair_name' + flavor_obj = Mock() + flavor_obj.id = '1' + port_obj = Mock() + port_obj.id = '2' + instance_obj = Mock() + instance_obj.name = 'instance_name' + secgrp = Mock() + secgrp.name = 'sec_grp' + secgrp.id = '4' + img_cre = Mock() + img_cre.id = '5' + network = Mock() + network.id = '6' + ports = ['port1'] + port_obj_list = [port_obj] + port_security = True + + log_calls = [call('Creating Key Pair vm_name...'), + call('Creating Port ' + str(ports) + '...'), + call('Creating the instance vm_name...'), + call('Waiting for instance_name to become Active'), + call('instance_name is active')] + + self.conn.compute.create_keypair.return_value = keypair_obj + self.conn.compute.find_flavor.return_value = flavor_obj + self.conn.network.create_port.return_value = port_obj + self.conn.compute.create_server.return_value = instance_obj + # self.conn.compute.wait_for_server.return_value = wait_ins_obj + + expected = (instance_obj, port_obj_list) + result = self.os_sfc.create_instance('vm_name', + 'flavor_name', + img_cre, + network, + secgrp, + 'av_zone', + ports, + port_security=port_security) + self.assertEqual(expected, result) + + pc_calls = [call(name=ports[0], + is_port_security_enabled=port_security, + network_id=network.id, + security_group_ids=[secgrp.id])] + + self.conn.compute.create_keypair.\ + assert_called_once_with(name='vm_name' + "_keypair") + + self.conn.compute.find_flavor.assert_called_once_with('flavor_name') + + self.conn.network.create_port.\ + assert_has_calls(pc_calls) + + self.conn.compute.create_server.\ + assert_called_once_with(name='vm_name', + image_id=img_cre.id, + flavor_id=flavor_obj.id, + networks=[{"port": port_obj.id}], + key_name=keypair_obj.name, + availability_zone='av_zone') + + self.conn.compute.wait_for_server.\ + assert_called_once_with(instance_obj) + + self.assertEqual([keypair_obj, port_obj, instance_obj], + self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_instance_port_security_false_one_port(self, + mock_log): + """ + Checks the proper functionality of create_instance + function + """ + + keypair_obj = Mock() + keypair_obj.name = 'keypair_name' + flavor_obj = Mock() + flavor_obj.id = '1' + port_obj = Mock() + port_obj.id = '2' + instance_obj = Mock() + instance_obj.name = 'instance_name' + secgrp = Mock() + secgrp.name = 'sec_grp' + secgrp.id = '4' + img_cre = Mock() + img_cre.id = '5' + network = Mock() + network.id = '6' + ports = ['port1'] + port_obj_list = [port_obj] + port_security = False + + log_calls = [call('Creating Key Pair vm_name...'), + call('Creating Port ' + str(ports) + '...'), + call('Creating the instance vm_name...'), + call('Waiting for instance_name to become Active'), + call('instance_name is active')] + + self.conn.compute.create_keypair.return_value = keypair_obj + self.conn.compute.find_flavor.return_value = flavor_obj + self.conn.network.create_port.return_value = port_obj + self.conn.compute.create_server.return_value = instance_obj + + expected = (instance_obj, port_obj_list) + result = self.os_sfc.create_instance('vm_name', + 'flavor_name', + img_cre, + network, + secgrp, + 'av_zone', + ports, + port_security=port_security) + self.assertEqual(expected, result) + + pc_calls = [call(name=ports[0], + is_port_security_enabled=port_security, + network_id=network.id)] + + self.conn.compute.create_keypair.\ + assert_called_once_with(name='vm_name' + "_keypair") + + self.conn.compute.find_flavor.assert_called_once_with('flavor_name') + + self.conn.network.create_port.\ + assert_has_calls(pc_calls) + + self.conn.compute.create_server.\ + assert_called_once_with(name='vm_name', + image_id=img_cre.id, + flavor_id=flavor_obj.id, + networks=[{"port": port_obj.id}], + key_name=keypair_obj.name, + availability_zone='av_zone') + + self.conn.compute.wait_for_server.\ + assert_called_once_with(instance_obj) + + self.assertEqual([keypair_obj, port_obj, instance_obj], + self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + def test_get_instance(self): + """ + Checks the proper functionality of get_instance function + """ + + mock_instance_id = 'instance-abyz' + mock_instance = Mock() + mock_instance.id = mock_instance_id + mock_instance.name = 'test-instance' + mock_instance.hypervisor_hostname = 'nova-abyz' + self.conn.compute.get_server_metadata.return_value = mock_instance + result = self.os_sfc.get_instance(mock_instance_id) + self.assertEqual(result, mock_instance) + + @patch.object(os_sfc_utils.OpenStackSFC, 'get_hypervisor_hosts') + def test_get_av_zones(self, mock_hosts): + """ + Checks the proper functionality of get_av_zone + function + """ + mock_hosts.return_value = ['host1', 'host2'] + result = self.os_sfc.get_av_zones() + mock_hosts.assert_called_once() + self.assertEqual(['nova::host1', 'nova::host2'], result) + + def test_get_hypervisor_hosts(self): + """ + Checks the proper functionality of get_av_zone + function + """ + from openstack.compute.v2 import hypervisor + + hypervisor1 = Mock() + hypervisor1.state = 'up' + hypervisor1.name = 'compute00' + hypervisor2 = Mock() + hypervisor2.state = 'up' + hypervisor2.name = 'compute01' + nodes = [hypervisor1.name, hypervisor2.name] + hypervisors_list = MagicMock() + mock_obj = patch.object(hypervisor, 'Hypervisor') + mock_obj.side_effect = [hypervisor1, hypervisor2] + self.conn.compute.hypervisors.return_value = hypervisors_list + hypervisors_list.__iter__.return_value = [hypervisor1, hypervisor2] + + result = self.os_sfc.get_hypervisor_hosts() + self.conn.compute.hypervisors.assert_called_once() + self.assertEqual(nodes, result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_hypervisor_hosts_exception(self, mock_log): + """ + Checks the proper functionality of get_av_zone + function when an exception appears + """ + log_calls = [call('Error [get_hypervisors(compute)]: Error MSG')] + self.conn.compute.hypervisors.side_effect = Exception('Error MSG') + result = self.os_sfc.get_hypervisor_hosts() + mock_log.error.assert_has_calls(log_calls) + self.assertIsNone(result) + + @patch('sfc.lib.openstack_utils.OpenStackSFC.get_vm_compute', + autospec=True, return_value='mock_client') + def test_compute_client(self, mock_get_vm_compute): + """ + Checks the proper functionality of get_compute_client + function + """ + + result = self.os_sfc.get_compute_client() + self.assertEqual('mock_client', result) + mock_get_vm_compute.assert_called_once_with(self.os_sfc, 'client') + + @patch('sfc.lib.openstack_utils.OpenStackSFC.get_vm_compute', + autospec=True, return_value='mock_server') + def test_get_compute_server(self, mock_get_vm_compute): + """ + Checks the proper functionality of get_compute_server + function + """ + + result = self.os_sfc.get_compute_server() + self.assertEqual('mock_server', result) + mock_get_vm_compute.assert_called_once_with(self.os_sfc, 'server') + + def test_get_vm_compute_raised_exception(self): + """ + Checks the proper functionality of get_vm_compute + function when no VM with the given name is found + """ + + ErrorMSG = "There is no VM with name 'mock_vm_name'!!" + with self.assertRaises(Exception) as cm: + self.os_sfc.get_vm_compute('mock_vm_name') + + self.assertEqual(cm.exception.message, ErrorMSG) + + def test_get_vm_compute(self): + """ + Checks the proper functionality of get_vm_compute + function + """ + + mock_cre_obj_1 = Mock() + mock_cre_obj_2 = Mock() + mock_cre_obj_1.get_vm_inst.return_value.name = 'pro_vm' + mock_cre_obj_2.get_vm_inst.return_value.name = 'dev_vm' + mock_cre_obj_2.get_vm_inst.return_value.compute_host = 'mock_host' + self.os_sfc.creators = [mock_cre_obj_1, mock_cre_obj_2] + + result = self.os_sfc.get_vm_compute('dev_vm') + self.assertEqual('mock_host', result) + + def test_get_port_by_ip(self): + """ + Checks the proper functonality of get_port_by_ip function + """ + + mock_port_ip_address = 'e.f.g.h' + mock_port_one, mock_port_two = Mock(), Mock() + mock_port_one.id = 'port-abcd' + mock_port_two.id = 'port-efgz' + mock_port_one.fixed_ips = [{'ip_address': 'a.b.c.d'}] + mock_port_two.fixed_ips = [{'ip_address': 'e.f.g.h'}] + self.conn.network.ports.return_value = [mock_port_one, mock_port_two] + self.conn.network.get_port.return_value = mock_port_two + result = self.os_sfc.get_port_by_ip(mock_port_ip_address) + self.assertEqual(result, mock_port_two) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance', + autospec=True) + def test_get_instance_port_raised_exceptioin(self, + mock_os_vm, + mock_log): + """ + Checks the proper functionality of get_client_port + function when no port is returned + """ + + mock_os_vm_ins = mock_os_vm.return_value + mock_vm = Mock() + mock_vm.name = 'mock_vm_name' + mock_os_vm_ins.get_port_by_name.return_value = None + ErrorMSG = 'Client VM does not have the desired port' + log_calls = [call("The VM mock_vm_name does not have any port" + " with name mock_vm_name-port")] + + with self.assertRaises(Exception) as cm: + self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins) + + self.assertEqual(cm.exception.message, ErrorMSG) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance', + autospec=True) + def test_get_instance_port(self, + mock_os_vm, + mock_log): + """ + Checks the proper functionality of get_client_port + function when no port is returned + """ + + mock_os_vm_ins = mock_os_vm.return_value + mock_vm = Mock() + mock_vm.name = 'mock_vm_name' + mock_os_vm_ins.get_port_by_name.return_value = 'mock_port' + result = self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins) + self.assertEqual('mock_port', result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.env.get', autospec=True) + def test_assign_floating_ip(self, + mock_env_get, + mock_log): + """ + Checks the proper functionality of assigning_floating_ip + function + """ + ext_network_obj = Mock() + ext_network_obj.id = '1' + fip_obj = Mock() + fip_obj.floating_ip_address = 'floating_ip_address' + port_obj = Mock() + port_obj.id = '2' + instance_obj = Mock() + instance_obj.id = '3' + + log_calls = [call(' Creating floating ips '), + call(' FLoating IP address ' + + fip_obj.floating_ip_address + + ' created'), + call(' Adding Floating IPs to instances ')] + + mock_env_get.return_value = 'ext_net_name' + self.conn.network.find_network.return_value = ext_network_obj + self.conn.network.create_ip.return_value = fip_obj + self.conn.netwotk.get_port.return_value = port_obj + self.conn.compute.get_server.return_value = instance_obj + + result = self.os_sfc.assign_floating_ip(instance_obj, port_obj) + assert result is fip_obj.floating_ip_address + + self.conn.network.find_network.assert_called_once_with('ext_net_name') + self.conn.network.create_ip.\ + assert_called_once_with(floating_network_id=ext_network_obj.id, + port_id=port_obj.id) + self.conn.compute.add_floating_ip_to_server.\ + assert_called_once_with(instance_obj.id, + fip_obj.floating_ip_address) + + self.assertEqual([fip_obj], + self.os_sfc.creators) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.heat_utils.get_stack_servers', + autospec=True) + @patch('sfc.lib.openstack_utils.cr_inst.generate_creator', autospec=True) + def test_assign_floating_ip_vnfs_raised_exception_ips_provided( + self, mock_generate_creator, mock_get_stack_servers, mock_log): + """ + Checks the proper functionality of assign_floating_ip_vnfs + function when server name does not have any floating IP assignment + """ + + ErrorMSG = "The VNF server_name-float does not have any suitable" + \ + " port with ip any of ['floating_ip', 'other_ip'] for" + \ + " floating IP assignment" + log_calls = [call(ErrorMSG)] + self.os_sfc.image_settings = 'image_settings' + self.heat.stacks.list.return_value = ['stack_obj'] + mock_ips = ['floating_ip', 'other_ip'] + mock_server_obj = Mock() + mock_port_obj = Mock() + mock_server_obj.name = 'server_name' + mock_server_obj.ports = [mock_port_obj] + mock_port_obj.name = None + mock_port_obj.ips = [{'ip_address': 'floating_ip'}] + mock_get_stack_servers.return_value = [mock_server_obj] + + with self.assertRaises(Exception) as cm: + self.os_sfc.assign_floating_ip_vnfs('router', mock_ips) + + self.assertEqual(cm.exception.message, ErrorMSG) + mock_get_stack_servers.assert_called_once_with(self.heat, + self.nova, + self.neutron_client, + self.keystone, + 'stack_obj', + 'admin') + mock_generate_creator.assert_called_once_with(self.os_creds, + mock_server_obj, + 'image_settings', + 'admin') + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.heat_utils.get_stack_servers', + autospec=True) + @patch('sfc.lib.openstack_utils.cr_inst.generate_creator', autospec=True) + def test_assign_floating_ip_vnfs_raised_exception_ips_not_provided( + self, mock_generate_creator, mock_get_stack_servers, mock_log): + """ + Checks the proper functionality of assign_floating_ip_vnfs + function when server name does not have any floating IP assignment + """ + + ErrorMSG = "The VNF server_name-float does not have any suitable" + \ + " port for floating IP assignment" + log_calls = [call(ErrorMSG)] + self.os_sfc.image_settings = 'image_settings' + self.heat.stacks.list.return_value = ['stack_obj'] + mock_server_obj = Mock() + mock_port_obj = Mock() + mock_server_obj.name = 'server_name' + mock_server_obj.ports = [mock_port_obj] + mock_port_obj.name = None + mock_port_obj.ips = [{'ip_address': 'floating_ip'}] + mock_get_stack_servers.return_value = [mock_server_obj] + + with self.assertRaises(Exception) as cm: + self.os_sfc.assign_floating_ip_vnfs('router') + + mock_get_stack_servers.assert_called_once_with(self.heat, + self.nova, + self.neutron_client, + self.keystone, + 'stack_obj', + 'admin') + mock_generate_creator.assert_called_once_with(self.os_creds, + mock_server_obj, + 'image_settings', + 'admin') + self.assertEqual(cm.exception.message, ErrorMSG) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.FloatingIpConfig', autospec=True) + @patch('sfc.lib.openstack_utils.cr_inst.generate_creator', + autospec=True) + @patch('sfc.lib.openstack_utils.heat_utils.get_stack_servers', + autospec=True) + def test_assign_floating_ip_vnfs(self, + mock_get_stack_servers, + mock_generate_creator, + mock_floating_ip_config): + """ + Checks the proper functionality of assign_floating_ip_vnfs + function + """ + + self.os_sfc.image_settings = 'image_settings' + self.heat.stacks.list.return_value = ['stack_obj'] + + mock_router = Mock() + mock_server_obj = Mock() + mock_ip_obj = Mock() + mock_port_obj = Mock() + mock_router.name = 'm_router' + mock_server_obj.name = 'serv_obj' + mock_server_obj.ports = [mock_port_obj] + mock_ips = ['floating_ip', 'other_ip'] + mock_ip_obj.ip = 'mocked_ip' + mock_port_obj.name = 'port_obj' + mock_port_obj.ips = [{'ip_address': 'floating_ip'}] + mock_get_stack_servers.return_value = [mock_server_obj] + mock_os_vm_ins = mock_generate_creator.return_value + float_ip_ins = mock_floating_ip_config.return_value + mock_os_vm_ins.add_floating_ip.return_value = mock_ip_obj + + result = self.os_sfc.assign_floating_ip_vnfs(mock_router, mock_ips) + self.assertEqual(['mocked_ip'], result) + self.assertEqual([mock_os_vm_ins], self.os_sfc.creators) + mock_get_stack_servers.assert_called_once_with(self.heat, + self.nova, + self.neutron_client, + self.keystone, + 'stack_obj', 'admin') + mock_generate_creator.assert_called_once_with(self.os_creds, + mock_server_obj, + 'image_settings', + 'admin') + mock_floating_ip_config.assert_called_once_with(name='serv_obj-float', + port_name='port_obj', + router_name='m_router') + mock_os_vm_ins.add_floating_ip.assert_called_once_with(float_ip_ins) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_all_security_groups(self, mock_log): + """ + Checks the proper functionality of delete_all_security_groups + function + """ + + log_calls_info = [call('Deleting remaining security groups...')] + secgrp1_obj = Mock() + secgrp2_obj = Mock() + secgrp_list = MagicMock() + + self.conn.network.create_security_groups.side_effect = [secgrp1_obj, + secgrp2_obj] + self.conn.network.security_groups.return_value = secgrp_list + + secgrp_list.__iter__.return_value = [secgrp1_obj, secgrp2_obj] + del_calls = [call(secgrp1_obj), + call(secgrp2_obj)] + + self.os_sfc.delete_all_security_groups() + self.conn.network.security_groups.assert_called_once() + self.conn.network.delete_security_group.assert_has_calls(del_calls) + mock_log.info.assert_has_calls(log_calls_info) + + @patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance', + autospec=True) + def test_wait_for_vnf(self, mock_os_vm): + """ + Checks the proper functionality of wait_for_vnf function + """ + + mock_os_vm.vm_active.return_value = "x" + result = self.os_sfc.wait_for_vnf(mock_os_vm) + self.assertEqual('x', result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_port_groups_raises_exception(self, mock_log): + """ + Checks the create_port_groups when length of ports is greater than 2 + """ + instance_obj = Mock() + instance_obj.name = 'name' + self.conn.compute.get_server.return_value = instance_obj + + log_calls_info = [call('Creating the port pairs...')] + log_calls_err = [call('Only SFs with one or two ports are supported')] + exception_message = "Failed to create port pairs" + vnf_ports = ['p1', 'p2', 'p3'] + with self.assertRaises(Exception) as cm: + self.os_sfc.create_port_groups(vnf_ports, instance_obj) + self.assertEqual(exception_message, cm.exception.message) + mock_log.info.assert_has_calls(log_calls_info) + mock_log.error.assert_has_calls(log_calls_err) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_port_groups_returns_none_from_pp(self, mock_log): + """ + Checks the create_port_groups when something goes wrong in port pair + creation + """ + instance_obj = Mock() + instance_obj.name = 'name' + port_obj1 = Mock() + port_obj2 = Mock() + port_obj1.id = '123abc' + port_obj2.id = '456def' + + self.conn.compute.get_server.return_value = instance_obj + self.conn.network.get_port.return_value = port_obj1 + self.conn.network.get_port.return_value = port_obj2 + + log_calls_info = [call('Creating the port pairs...')] + log_calls_warn = [call('Chain creation failed due to port pair ' + 'creation failed for vnf %(vnf)s', + {'vnf': instance_obj.name})] + self.neutron_client.create_sfc_port_pair.return_value = None + result = self.os_sfc.create_port_groups( + [port_obj1, port_obj2], instance_obj) + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls_info) + mock_log.warning.assert_has_calls(log_calls_warn) + + @patch('snaps.domain.network.Port', autospec=True) + @patch('snaps.domain.vm_inst.VmInst', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_port_groups_exception_nopp(self, mock_log, mock_osvm, + mock_port): + """ + Checks the create_port_groups when openstack does not commit the pp + """ + + log_calls_info = [call('Creating the port pairs...')] + mock_port_ins = mock_port.return_value + mock_port_ins.id = '123abc' + mock_vm_ins = mock_osvm.return_value + mock_vm_ins.name = 'vm' + exception_message = "Port pair was not committed in openstack" + expected_port_pair = {'name': 'vm-connection-points', + 'description': 'port pair for vm', + 'ingress': '123abc', + 'egress': '123abc'} + self.neutron_client.create_sfc_port_pair.return_value = \ + {'port_pair': {'id': 'pp_id'}} + self.neutron_client.list_sfc_port_pairs.return_value = \ + {'port_pairs': [{'id': 'xxxx'}]} + with self.assertRaises(Exception) as cm: + self.os_sfc.create_port_groups([mock_port_ins], mock_vm_ins) + self.assertEqual(exception_message, cm.exception.message) + self.neutron_client.create_sfc_port_pair.assert_has_calls( + [call({'port_pair': expected_port_pair})]) + mock_log.info.assert_has_calls(log_calls_info) + + @patch('snaps.domain.network.Port', autospec=True) + @patch('snaps.domain.vm_inst.VmInst', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_port_groups_returns_none_from_ppg(self, mock_log, + mock_vm, + mock_port): + """ + Checks the create_port_groups when something goes wrong in port pair + group creation + """ + + instance_obj = Mock() + instance_obj.name = 'name' + port_obj = Mock() + port_obj.id = '123abc' + + self.conn.compute.get_server.return_value = instance_obj + self.conn.network.get_port.return_value = port_obj + + log_calls_info = [call('Creating the port pairs...'), + call('Creating the port pair groups for name')] + log_calls_warn = [call('Chain creation failed due to port pair group ' + 'creation failed for vnf ' + '{}'.format(instance_obj.name))] + self.neutron_client.create_sfc_port_pair.return_value = \ + {'port_pair': {'id': 'pp_id'}} + self.neutron_client.list_sfc_port_pairs.return_value = \ + {'port_pairs': [{'id': 'pp_id'}]} + self.neutron_client.create_sfc_port_pair_group.return_value = None + result = self.os_sfc.create_port_groups([port_obj], instance_obj) + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls_info) + mock_log.warning.assert_has_calls(log_calls_warn) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_port_groups_returns_id(self, mock_log): + """ + Checks the create_port_groups when everything goes as expected + """ + + log_calls_info = [call('Creating the port pairs...')] + + instance_obj = Mock() + instance_obj.name = 'vm' + port_obj = Mock() + port_obj.id = '123abc' + self.conn.compute.get_server.return_value = instance_obj + self.conn.network.get_port.return_value = port_obj + + expected_port_pair = {'name': 'vm-connection-points', + 'description': 'port pair for vm', + 'ingress': '123abc', + 'egress': '123abc'} + self.neutron_client.create_sfc_port_pair.return_value = \ + {'port_pair': {'id': 'pp_id'}} + self.neutron_client.list_sfc_port_pairs.return_value = \ + {'port_pairs': [{'id': 'pp_id'}]} + self.neutron_client.create_sfc_port_pair_group.return_value = \ + {'port_pair_group': {'id': 'pp_id'}} + expected_port_pair_gr = {'name': 'vm-port-pair-group', + 'description': 'port pair group for vm', + 'port_pairs': ['pp_id']} + + self.os_sfc.create_port_groups([port_obj], instance_obj) + self.neutron_client.create_sfc_port_pair.assert_has_calls( + [call({'port_pair': expected_port_pair})]) + self.neutron_client.create_sfc_port_pair_group.assert_has_calls( + [call({'port_pair_group': expected_port_pair_gr})]) + mock_log.info.assert_has_calls(log_calls_info) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_classifier(self, mock_log): + """ + Checks the create_classifier method + """ + + log_calls = [call('Creating the classifier...')] + neutron_port = 'neutron_port_id' + port = 80 + protocol = 'tcp' + fc_name = 'red_http' + symmetrical = False + self.neutron_client.create_sfc_flow_classifier.return_value = \ + {'flow_classifier': {'id': 'fc_id'}} + + expected_sfc_classifier_params = {'name': fc_name, + 'logical_source_port': neutron_port, + 'destination_port_range_min': port, + 'destination_port_range_max': port, + 'protocol': protocol} + self.os_sfc.create_classifier(neutron_port, port, + protocol, fc_name, symmetrical) + self.neutron_client.create_sfc_flow_classifier.assert_has_calls( + [call({'flow_classifier': expected_sfc_classifier_params})]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_classifier_symmetric(self, mock_log): + """ + Checks the create_chain method + """ + + log_calls = [call('Creating the classifier...')] + neutron_port = 'neutron_port_id' + port = 80 + protocol = 'tcp' + fc_name = 'red_http' + symmetrical = True + serv_p = '123' + server_ip = '1.1.1.2' + self.neutron_client.create_sfc_flow_classifier.return_value = \ + {'flow_classifier': {'id': 'fc_id'}} + + expected_sfc_classifier_params = {'name': fc_name, + 'logical_source_port': neutron_port, + 'destination_port_range_min': port, + 'destination_port_range_max': port, + 'destination_ip_prefix': server_ip, + 'logical_destination_port': serv_p, + 'protocol': protocol} + self.os_sfc.create_classifier(neutron_port, port, + protocol, fc_name, symmetrical, + server_port='123', + server_ip='1.1.1.2') + self.neutron_client.create_sfc_flow_classifier.assert_has_calls( + [call({'flow_classifier': expected_sfc_classifier_params})]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_chain(self, mock_log): + """ + Checks the create_chain method + """ + + log_calls = [call('Creating the classifier...'), + call('Creating the chain...')] + port_groups = ['1a', '2b'] + neutron_port = 'neutron_port_id' + port = 80 + protocol = 'tcp' + vnffg_name = 'red_http' + symmetrical = False + self.neutron_client.create_sfc_flow_classifier.return_value = \ + {'flow_classifier': {'id': 'fc_id'}} + self.neutron_client.create_sfc_port_chain.return_value = \ + {'port_chain': {'id': 'pc_id'}} + + expected_sfc_classifier_params = {'name': vnffg_name + '-classifier', + 'logical_source_port': neutron_port, + 'destination_port_range_min': port, + 'destination_port_range_max': port, + 'protocol': protocol} + expected_chain_config = {'name': vnffg_name + '-port-chain', + 'description': 'port-chain for SFC', + 'port_pair_groups': port_groups, + 'flow_classifiers': ['fc_id']} + + self.os_sfc.create_chain(port_groups, neutron_port, port, + protocol, vnffg_name, symmetrical) + + self.neutron_client.create_sfc_flow_classifier.assert_has_calls( + [call({'flow_classifier': expected_sfc_classifier_params})]) + self.neutron_client.create_sfc_port_chain.assert_has_calls( + [call({'port_chain': expected_chain_config})]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_chain_symmetric(self, mock_log): + """ + Checks the create_chain method + """ + + log_calls = [call('Creating the classifier...'), + call('Creating the chain...')] + port_groups = ['1a', '2b'] + neutron_port = 'neutron_port_id' + port = 80 + protocol = 'tcp' + vnffg_name = 'red_http' + symmetrical = True + serv_p = '123abc' + server_ip = '1.1.1.2' + self.neutron_client.create_sfc_flow_classifier.return_value = \ + {'flow_classifier': {'id': 'fc_id'}} + self.neutron_client.create_sfc_port_chain.return_value = \ + {'port_chain': {'id': 'pc_id'}} + + expected_sfc_classifier_params = {'name': vnffg_name + '-classifier', + 'logical_source_port': neutron_port, + 'destination_port_range_min': port, + 'destination_port_range_max': port, + 'destination_ip_prefix': server_ip, + 'logical_destination_port': serv_p, + 'protocol': protocol} + expected_chain_config = {'name': vnffg_name + '-port-chain', + 'description': 'port-chain for SFC', + 'port_pair_groups': port_groups, + 'flow_classifiers': ['fc_id'], + 'chain_parameters': {'symmetric': True}} + + self.os_sfc.create_chain(port_groups, neutron_port, port, + protocol, vnffg_name, symmetrical, + server_port=serv_p, server_ip=server_ip) + + self.neutron_client.create_sfc_flow_classifier.assert_has_calls( + [call({'flow_classifier': expected_sfc_classifier_params})]) + self.neutron_client.create_sfc_port_chain.assert_has_calls( + [call({'port_chain': expected_chain_config})]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_update_chain_symmetric(self, mock_log): + """ + Checks the update_chain method + """ + + log_calls = [call('Update the chain...')] + vnffg_name = 'red_http' + fc_name = 'blue_ssh' + symmetrical = True + self.neutron_client.find_resource.return_value = \ + {'id': 'fc_id'} + expected_chain_config = {'name': vnffg_name + '-port-chain', + 'flow_classifiers': ['fc_id'], + 'chain_parameters': {'symmetric': True}} + self.os_sfc.update_chain(vnffg_name, fc_name, symmetrical) + self.neutron_client.update_sfc_port_chain.assert_has_calls( + [call('fc_id', {'port_chain': expected_chain_config})]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_swap_classifiers(self, mock_log): + """ + Checks the swap_classifiers method + """ + + log_calls = [call('Swap classifiers...')] + vnffg_1_name = 'red_http' + vnffg_2_name = 'blue_ssh' + symmetrical = False + self.os_sfc.swap_classifiers(vnffg_1_name, vnffg_2_name, symmetrical) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_port_groups(self, mock_log): + """ + Checks the delete_port_groups method + """ + log_calls = [call('Deleting the port groups...'), + call('Deleting the port pairs...')] + self.neutron_client.list_sfc_port_pair_groups.return_value = \ + {'port_pair_groups': [{'id': 'id_ppg1'}, {'id': 'id_ppg2'}]} + self.neutron_client.list_sfc_port_pairs.return_value = \ + {'port_pairs': [{'id': 'id_pp1'}, {'id': 'id_pp2'}]} + self.os_sfc.delete_port_groups() + + self.neutron_client.delete_sfc_port_pair_group.assert_has_calls( + [call('id_ppg1'), call('id_ppg2')]) + self.neutron_client.delete_sfc_port_pair.assert_has_calls( + [call('id_pp1'), call('id_pp2')]) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_chain(self, mock_log): + """ + Checks the delete_chain method + """ + log_calls = [call('Deleting the chain...'), + call('Deleting the classifiers...')] + self.neutron_client.list_sfc_port_chains.return_value = \ + {'port_chains': [{'id': 'id_pc1'}]} + self.neutron_client.list_sfc_flow_classifiers.return_value = \ + {'flow_classifiers': [{'id': 'id_fc1'}]} + self.os_sfc.delete_chain() + + self.neutron_client.delete_sfc_port_chain.\ + assert_has_calls([call('id_pc1')]) + self.neutron_client.delete_sfc_flow_classifier.assert_has_calls( + [call('id_fc1')]) + mock_log.info.assert_has_calls(log_calls) + + +class SfcTackerSectionTesting(unittest.TestCase): + def setUp(self): + self.patcher = patch.object(tacker_client, 'Client', autospec=True) + self.mock_tacker_client = self.patcher.start().return_value + + def tearDown(self): + self.patcher.stop() + + @patch('os.getenv', autospec=True, return_value=None) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_tacker_client_version_returned_default(self, + mock_log, + mock_getenv): + """ + Checks the proper functionality of get_tacker_client_version + function when the os.getenv returns none + """ + result = os_sfc_utils.get_tacker_client_version() + self.assertEqual(result, '1.0') + mock_getenv.assert_called_once_with('OS_TACKER_API_VERSION') + mock_log.info.assert_not_called() + + @patch('os.getenv', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_tacker_client_version(self, + mock_log, + mock_getenv): + """ + Checks the proper functionality of get_tacker_client_version + function when the os.getenv returns version + """ + + ver = '2.0' + mock_getenv.return_value = ver + log_calls = [call("OS_TACKER_API_VERSION is set in env as '%s'", ver)] + + result = os_sfc_utils.get_tacker_client_version() + self.assertEqual(result, ver) + mock_getenv.assert_called_once_with('OS_TACKER_API_VERSION') + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_id_from_name_returned_none(self, mock_log): + """ + Checks the proper functionality of get_id_from_name + function when tacker_client.list returns None + """ + + resource_name = 'mock_resource_name' + resource_type = 'mock_resource_type' + params = {'fields': 'id', 'name': resource_name} + collection = resource_type + 's' + path = '/' + collection + self.mock_tacker_client.list.side_effect = Exception('ErrorMSG') + log_calls = [call('Error [get_id_from_name(tacker_client, ' + 'resource_type, resource_name)]: ErrorMSG')] + + result = os_sfc_utils.get_id_from_name(self.mock_tacker_client, + resource_type, + resource_name) + self.assertIsNone(result) + self.mock_tacker_client.list.assert_called_once_with(collection, + path, + **params) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.openstack_tests.get_credentials', + autospec=True, return_value='os_creds') + @patch('sfc.lib.openstack_utils.keystone_utils.keystone_session', + autospec=True, return_value='keystone_session_obj') + @patch('sfc.lib.openstack_utils.constants.ENV_FILE', autospec=True) + @patch('sfc.lib.openstack_utils.tackerclient.Client', autospec=True) + def test_get_tacker_client(self, mock_tacker_client, + mock_env_file, + mock_keystone_session, + mock_get_credentials): + """ + checks the proper functionality of get_tacker_client + function + """ + + mock_tacker_client_ins = mock_tacker_client.return_value + result = os_sfc_utils.get_tacker_client() + assert result is mock_tacker_client_ins + mock_get_credentials.assert_called_once_with(os_env_file=mock_env_file, + overrides=None) + mock_keystone_session.assert_called_once_with('os_creds') + mock_tacker_client.assert_called_once_with( + '1.0', session='keystone_session_obj') + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_id_from_name(self, mock_log): + """ + Checks the proper functionality of get_id_from_name + function when tacker_client.list returns id + """ + + resource_name = 'mock_resource_name' + resource_type = 'mock_resource_type' + params = {'fields': 'id', 'name': resource_name} + collection = resource_type + 's' + self.mock_tacker_client.list.return_value = {collection: {0: {'id': + 'mock_id'}}} + path = '/' + collection + result = os_sfc_utils.get_id_from_name(self.mock_tacker_client, + resource_type, + resource_name) + self.assertEqual('mock_id', result) + self.mock_tacker_client.list.assert_called_once_with(collection, + path, + **params) + mock_log.error.assert_not_called() + + @patch('sfc.lib.openstack_utils.get_id_from_name', autospec=True) + def test_get_vnfd_id(self, mock_get_id): + """ + Checks the proper functionality of get_vnfd_id + function + """ + + mock_get_id.return_value = 'id' + result = os_sfc_utils.get_vnfd_id(self.mock_tacker_client, + 'vnfd_name') + self.assertEqual('id', result) + mock_get_id.assert_called_once_with(self.mock_tacker_client, + 'vnfd', + 'vnfd_name') + + @patch('sfc.lib.openstack_utils.get_id_from_name', autospec=True) + def test_get_vim_id(self, mock_get_id): + """ + Checks the proper fucntionality of get_vim_id + function + """ + + mock_get_id.return_value = 'id' + result = os_sfc_utils.get_vim_id(self.mock_tacker_client, 'vim_name') + mock_get_id.assert_called_once_with(self.mock_tacker_client, + 'vim', + 'vim_name') + self.assertEqual('id', result) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_id_from_name', autospec=True) + def test_get_vnf_id(self, + mock_get_id, + mock_log, + mock_sleep): + """ + Checks the proper functionality of get_vnf_id + function + """ + + vnf_name = 'mock_vnf' + log_calls = [call("Could not retrieve ID for vnf with name [%s]." + " Retrying." % vnf_name)] + + get_id_calls = [call(self.mock_tacker_client, 'vnf', vnf_name)] * 2 + + mock_get_id.side_effect = [None, 'vnf_id'] + + result = os_sfc_utils.get_vnf_id(self.mock_tacker_client, vnf_name, 2) + self.assertEqual('vnf_id', result) + mock_sleep.assert_called_once_with(1) + mock_log.info.assert_has_calls(log_calls) + mock_get_id.assert_has_calls(get_id_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_id_from_name', autospec=True) + def test_get_vnffg_id(self, + mock_get_id, + mock_log, + mock_sleep): + """ + Checks the proper functionality of get_vnffg_id + function + """ + + vnffg_name = 'mock_vnffg' + log_calls = [call("Could not retrieve ID for vnffg with name [%s]." + " Retrying." % vnffg_name)] + + get_id_calls = [call(self.mock_tacker_client, 'vnffg', vnffg_name)] * 2 + + mock_get_id.side_effect = [None, 'vnf_id'] + + result = os_sfc_utils.get_vnffg_id(self.mock_tacker_client, + vnffg_name, + 2) + self.assertEqual('vnf_id', result) + mock_sleep.assert_called_once_with(1) + mock_log.info.assert_has_calls(log_calls) + mock_get_id.assert_has_calls(get_id_calls) + + @patch('sfc.lib.openstack_utils.get_id_from_name', autospec=True) + def test_get_vnffgd_id(self, mock_get_id): + """ + Checks the proper functionality of get_vnffgd_id + function + """ + + mock_get_id.return_value = 'id' + result = os_sfc_utils.get_vnffgd_id(self.mock_tacker_client, + 'vnffgd_name') + mock_get_id.assert_called_once_with(self.mock_tacker_client, + 'vnffgd', + 'vnffgd_name') + self.assertEqual('id', result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vnfds_returned_none(self, mock_log): + """ + Checks the proper functionality of list_vnfds + function when the list_vnfds returns none + """ + + log_calls = [call('Error [list_vnfds(tacker_client)]: ErrorMSG')] + self.mock_tacker_client.list_vnfds.side_effect = Exception('ErrorMSG') + result = os_sfc_utils.list_vnfds(self.mock_tacker_client) + mock_log.error.assert_has_calls(log_calls) + self.mock_tacker_client.list_vnfds.assert_called_once_with( + retrieve_all=True) + self.assertIsNone(result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vnfds(self, mock_log): + """ + Checks the proper functionality of list_vnfds + function when the list_vnfds returns vnfds + """ + + vnfds = { + 'vnfds': [{'id': 1}, + {'id': 2}] + } + self.mock_tacker_client.list_vnfds.return_value = vnfds + result = os_sfc_utils.list_vnfds(self.mock_tacker_client) + self.mock_tacker_client.list_vnfds.assert_called_once_with( + retrieve_all=True) + mock_log.assert_not_called() + self.assertEqual([1, 2], result) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnfd_returned_none_tosca_file_not_provided(self, mock_log): + """ + Checks the proper functionality of create_vnfd + function when an exception is raised + """ + + log_calls = [call("Creating the vnfd..."), + call("Error [create_vnfd(tacker_client, 'None')]: " + "ErrorMSG")] + + self.mock_tacker_client.create_vnfd.side_effect = Exception('ErrorMSG') + result = os_sfc_utils.create_vnfd(self.mock_tacker_client, + None, + 'vnfd_name') + self.assertIsNone(result) + self.mock_tacker_client.create_vnfd.assert_called_once_with( + body={'vnfd': {'attributes': {'vnfd': {}}, + 'name': 'vnfd_name'}}) + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.error.assert_has_calls(log_calls[1:]) + + @patch('yaml.safe_load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnfd_returned_none_tosca_file_provided(self, + mock_log, + mock_open, + mock_safe_load): + """ + Checks the proper functionality of create_vnfd + function when an exception is raised + """ + + log_calls = [call("Creating the vnfd..."), + call("VNFD template:\nmock_vnfd"), + call("Error [create_vnfd(tacker_client, 'tosca_file')]: " + "ErrorMSG")] + + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'mock_vnfd' + mock_safe_load.return_value = 'mock_vnfd_body' + self.mock_tacker_client.create_vnfd.side_effect = Exception('ErrorMSG') + result = os_sfc_utils.create_vnfd(self.mock_tacker_client, + 'tosca_file', + 'vnfd_name') + self.assertIsNone(result) + mock_open.assert_called_once_with('tosca_file') + open_handler.read.assert_called_once_with() + mock_safe_load.assert_called_once_with('mock_vnfd') + mock_log.info.assert_has_calls(log_calls[:2]) + mock_log.error.assert_has_calls(log_calls[2:]) + + @patch('yaml.safe_load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnfd(self, + mock_log, + mock_open, + mock_safe_load): + """ + Checks the proper functionality of create_vnfd + function + """ + + log_calls = [call("VNFD template:\nmock_vnfd")] + + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'mock_vnfd' + mock_safe_load.return_value = 'mock_vnfd_body' + result = os_sfc_utils.create_vnfd(self.mock_tacker_client, + 'tosca_file', + 'vnfd_name') + assert result is self.mock_tacker_client.create_vnfd.return_value + self.mock_tacker_client.create_vnfd.assert_called_once_with( + body={"vnfd": {"attributes": {"vnfd": "mock_vnfd_body"}, + "name": "vnfd_name"}}) + mock_open.assert_called_once_with('tosca_file') + open_handler.read.assert_called_once_with() + mock_safe_load.assert_called_once_with('mock_vnfd') + mock_log.info.assert_has_calls(log_calls) + mock_log.error.assert_not_called() + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_vnfd_returned_none(self, mock_log): + """ + Checks the proper functionality of delete_vnfd + function when an exception is raised + """ + + log_calls = [call("Error [delete_vnfd(tacker_client, 'None', 'None')]:" + " You need to provide VNFD id or VNFD name")] + + result = os_sfc_utils.delete_vnfd(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.get_vnfd_id', + autospec=True, return_value='vnfd') + def test_delete_vnfd(self, mock_get_vnfd_id): + """ + Checks the proper functionality of delete_vnfd + function + """ + + result = os_sfc_utils.delete_vnfd(self.mock_tacker_client, + None, + 'vnfd_name') + assert result is self.mock_tacker_client.delete_vnfd.return_value + mock_get_vnfd_id.assert_called_once_with(self.mock_tacker_client, + 'vnfd_name') + self.mock_tacker_client.delete_vnfd.assert_called_once_with('vnfd') + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vnfs_returned_none(self, mock_log): + """ + Checks the proper functionality of list_vnfs + function + """ + + log_calls = [call("Error [list_vnfs(tacker_client)]: ErrorMSG")] + + self.mock_tacker_client.list_vnfs.side_effect = Exception('ErrorMSG') + result = os_sfc_utils.list_vnfs(self.mock_tacker_client) + self.assertIsNone(result) + self.mock_tacker_client.list_vnfs.assert_called_once_with( + retrieve_all=True) + mock_log.error.assert_has_calls(log_calls) + + def test_list_vnfs(self): + """ + Checks the proper functionality of list_vnfs + function + """ + vnfs = {'vnfs': [{'id': 1}, + {'id': 2}]} + + self.mock_tacker_client.list_vnfs.return_value = vnfs + result = os_sfc_utils.list_vnfs(self.mock_tacker_client) + self.assertEqual([1, 2], result) + self.mock_tacker_client.list_vnfs.assert_called_once_with( + retrieve_all=True) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnf_returned_none_vnfd_not_provided(self, mock_log): + """ + Checks the proper functionality of create_vnf + function when an exception is raised + """ + + log_calls = [call("Creating the vnf..."), + call("error [create_vnf(tacker_client," + " 'vnf_name', 'None', 'None')]: " + "vnfd id or vnfd name is required")] + result = os_sfc_utils.create_vnf(self.mock_tacker_client, 'vnf_name') + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.error.assert_has_calls(log_calls[1:]) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnf_returned_none_vnfd_provided(self, mock_log): + """ + Checks the proper functionality of create_vnf + function when an exception is raised + """ + + log_calls = [call("Creating the vnf..."), + call("error [create_vnf(tacker_client," + " 'vnf_name', 'None', 'vnfd_name')]: " + "vim id or vim name is required")] + result = os_sfc_utils.create_vnf(self.mock_tacker_client, + 'vnf_name', + None, + 'vnfd_name', + None, + None) + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.error.assert_has_calls(log_calls[1:]) + + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vim_id', + autospec=True, return_value='vim_id') + @patch('sfc.lib.openstack_utils.get_vnfd_id', + autospec=True, return_value='vnfd_id') + def test_create_vnf_vim_id_not_provided(self, + mock_get_vnfd_id, + mock_get_vim_id, + mock_log, + mock_open): + """ + Checks the proper functionality of create_vnf + function + """ + mock_body = {'vnf': {'attributes': {'param_values': 'mock_data'}, + 'vim_id': 'vim_id', + 'name': 'vnf_name', + 'vnfd_id': 'vnfd_id'}} + log_calls = [call('Creating the vnf...')] + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'mock_data' + result = os_sfc_utils.create_vnf(self.mock_tacker_client, + 'vnf_name', + None, + 'vnfd_name', + None, + 'vim_name', + 'param_file') + + assert result is self.mock_tacker_client.create_vnf.return_value + mock_get_vnfd_id.assert_called_once_with(self.mock_tacker_client, + 'vnfd_name') + mock_get_vim_id.assert_called_once_with(self.mock_tacker_client, + 'vim_name') + mock_log.info.assert_has_calls(log_calls) + self.mock_tacker_client.create_vnf.assert_called_once_with( + body=mock_body) + + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnf_vim_id_provided(self, mock_log, mock_open): + """ + Checks the proper functionality of create_vnf + function + """ + mock_body = {'vnf': {'attributes': {}, + 'vim_id': 'vim_id', + 'name': 'vnf_name', + 'vnfd_id': 'vnfd_id'}} + log_calls = [call('Creating the vnf...')] + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'mock_data' + + result = os_sfc_utils.create_vnf(self.mock_tacker_client, + 'vnf_name', + 'vnfd_id', + 'vnfd_name', + 'vim_id', + 'vim_name') + assert result is self.mock_tacker_client.create_vnf.return_value + mock_log.info.assert_has_calls(log_calls) + self.mock_tacker_client.create_vnf.assert_called_once_with( + body=mock_body) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_vnf_returned_none_vnf_not_provided(self, mock_log): + """ + Checks the proper functionality of get_vnf + functionality when an exception is raised + """ + + log_calls = [call("Could not retrieve VNF [vnf_id=None, vnf_name=None]" + " - You must specify vnf_id or vnf_name")] + + result = os_sfc_utils.get_vnf(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.get_vnf_id', + autospec=True, return_value=None) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_vnf_returned_none_vnf_provided(self, + mock_log, + mock_get_vnf_id): + """ + Checks the proper functionality of get_vnf + functionality when an exception is raised + """ + + log_calls = [call("Could not retrieve VNF [vnf_id=None, " + "vnf_name=vnf_name] - Could not retrieve ID from " + "name [vnf_name]")] + result = os_sfc_utils.get_vnf(self.mock_tacker_client, + None, + 'vnf_name') + self.assertIsNone(result) + mock_get_vnf_id.assert_called_once_with(self.mock_tacker_client, + 'vnf_name') + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.list_vnfs', autospec=True) + def test_get_vnf(self, + mock_list_vnfs, + mock_log): + """ + Checks the proper functionality of get_vnf + function + """ + + vnf = {'vnfs': [{'id': 'default'}, + {'id': 'vnf_id'}]} + + mock_list_vnfs.return_value = vnf + result = os_sfc_utils.get_vnf(self.mock_tacker_client, 'vnf_id', None) + self.assertDictEqual(vnf['vnfs'][1], result) + mock_log.error.assert_not_called() + + @patch('json.loads', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf', autospe=True) + def test_get_vnf_ip(self, + mock_get_vnf, + mock_json_loads): + """ + Checks the proper functionality of get_vnf_ip + function + """ + + vnf = {"mgmt_url": {"VDU1": "192.168.120.3"}} + mock_get_vnf.return_value = vnf + mock_json_loads.return_value = vnf['mgmt_url'] + result = os_sfc_utils.get_vnf_ip(self.mock_tacker_client) + self.assertEqual("192.168.120.3", result) + mock_get_vnf.assert_called_once_with(self.mock_tacker_client, + None, + None) + mock_json_loads.assert_called_once_with(vnf['mgmt_url']) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf', autospec=True) + def test_wait_for_vnf_returned_none_unable_to_retrieve_vnf(self, + mock_get_vnf, + mock_log): + """ + Checks the proper functionality of wait_for_vnf + function when an Exception is raised + """ + + mock_get_vnf.return_value = None + log_calls = [call("error [wait_for_vnf(tacker_client, 'vnf_id', " + "'vnf_name')]: Could not retrieve VNF - id='vnf_id'," + " name='vnf_name'")] + + result = os_sfc_utils.wait_for_vnf(self.mock_tacker_client, + 'vnf_id', + 'vnf_name', + 0) + self.assertIsNone(result) + mock_get_vnf.assert_called_once_with(self.mock_tacker_client, + 'vnf_id', + 'vnf_name') + mock_log.error.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf', autospec=True) + def test_wait_for_vnf_returned_none_unable_to_boot_vnf(self, + mock_get_vnf, + mock_log, + mock_sleep): + """ + Checks the proper functionality of wait_for_vnf + function when an Exception is raised + """ + + mock_vnf_values = [{'id': 'vnf_id', + 'status': 'ERROR'}, + {'id': 'vnf_id', + 'status': 'PEDNING_CREATE'}] + mock_get_vnf.side_effect = mock_vnf_values + log_calls = [call("Waiting for vnf %s" % str(mock_vnf_values[0])), + call("error [wait_for_vnf(tacker_client, 'vnf_id', " + "'vnf_name')]: Error when booting vnf vnf_id")] + + result = os_sfc_utils.wait_for_vnf(self.mock_tacker_client, + 'vnf_id', + 'vnf_name', + 0) + self.assertIsNone(result) + mock_get_vnf.assert_called_once_with(self.mock_tacker_client, + 'vnf_id', + 'vnf_name') + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.error.assert_has_calls(log_calls[1:]) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf', autospec=True) + def test_wait_for_vnf_returned_none_timeout_booting_vnf(self, + mock_get_vnf, + mock_log, + mock_sleep): + """ + Checks the proper functionality of wait_for_vnf + function when an Exception is raised + """ + + mock_vnf_values = [{'id': 'vnf_id', + 'status': 'PENDING_CREATE'}, + {'id': 'vnf_id', + 'status': 'PENDING_CREATE'}] + mock_get_vnf.side_effect = mock_vnf_values + log_calls = [call("Waiting for vnf %s" % str(mock_vnf_values[1])), + call("error [wait_for_vnf(tacker_client, 'vnf_id', " + "'vnf_name')]: Timeout when booting vnf vnf_id")] + + result = os_sfc_utils.wait_for_vnf(self.mock_tacker_client, + 'vnf_id', + 'vnf_name', + 0) + self.assertIsNone(result) + mock_get_vnf.assert_called_with(self.mock_tacker_client, + 'vnf_id', + 'vnf_name') + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.error.assert_has_calls(log_calls[1:]) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf', autospec=True) + def test_wait_for_vnf(self, + mock_get_vnf, + mock_log, + mock_sleep): + """ + Checks for the proper functionality of wait_for_vnf + function + """ + + mock_vnf_values = [{'status': 'PENDING_CREATE', + 'id': 'vnf_id'}, + {'status': 'ACTIVE', + 'id': 'vnf_id'}] + + log_calls = [call("Waiting for vnf %s" % mock_vnf_values[0])] + + mock_get_vnf.side_effect = mock_vnf_values + + result = os_sfc_utils.wait_for_vnf(self.mock_tacker_client, + 'vnf_id', + 'vnf_name', + 3) + self.assertEqual('vnf_id', result) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_vnf_returned_none(self, mock_log): + """ + Checks the proper functionality of delete_vnf + function + """ + + log_calls = [call("Error [delete_vnf(tacker_client, 'None', 'None')]:" + " You need to provide a VNF id or name")] + result = os_sfc_utils.delete_vnf(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnf_id', autospec=True) + def test_delete_vnf(self, + mock_get_vnf_id, + mock_log): + """ + Checks the proper functionality of delete_vnf + function + """ + + mock_get_vnf_id.return_value = 'vnf' + result = os_sfc_utils.delete_vnf(self.mock_tacker_client, + None, + 'vnf_name') + assert result is self.mock_tacker_client.delete_vnf.return_value + mock_get_vnf_id.assert_called_once_with(self.mock_tacker_client, + 'vnf_name') + self.mock_tacker_client.delete_vnf.assert_called_once_with('vnf') + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vim_returned_none(self, + mock_log): + """ + Checks the proper functionality of create_vim + function when the vim_file is not provided + """ + + self.mock_tacker_client.create_vim.side_effect = Exception('ErrorMSG') + log_calls = [[call("Creating the vim...")], + [call("Error [create_vim(tacker_client, 'None')]" + ": ErrorMSG")]] + + result = os_sfc_utils.create_vim(self.mock_tacker_client) + self.assertIsNone(result) + self.mock_tacker_client.create_vim.assert_called_once_with(body={}) + mock_log.info.assert_has_calls(log_calls[0]) + mock_log.error.assert_has_calls(log_calls[1]) + + @patch('json.load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vim(self, + mock_log, + mock_open, + mock_json_loads): + """ + Checks the proper functionality of create_vim + function + """ + + log_calls = [call("Creating the vim..."), + call("VIM template:\nmock_data")] + + open_handler = mock_open.return_value.__enter__.return_value + mock_json_loads.return_value = 'mock_data' + result = os_sfc_utils.create_vim(self.mock_tacker_client, 'vim_file') + assert result is self.mock_tacker_client.create_vim.return_value + mock_log.info.assert_has_calls(log_calls) + mock_open.assert_called_once_with('vim_file') + mock_json_loads.assert_called_once_with(open_handler) + mock_log.error.assert_not_called() + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnffgd_returned_none(self, mock_log): + """ + Checks the proper functionality of create_vnffgd + function when create_vnffgd raises an Exception + """ + + self.mock_tacker_client.create_vnffgd.side_effect = Exception( + 'ErrorMSG') + log_calls = [[call("Creating the vnffgd...")], + [call("Error [create_vnffgd(tacker_client, 'None')]" + ": ErrorMSG")]] + + result = os_sfc_utils.create_vnffgd(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls[0]) + mock_log.error.assert_has_calls(log_calls[1]) + + @patch('yaml.safe_load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnffgd(self, + mock_log, + mock_open, + mock_safe_load): + """ + Checks the proper functionality of create_vnffgd + function + """ + + log_calls = [call('Creating the vnffgd...'), + call('VNFFGD template:\nmock_data')] + + vnffgd_body = {'id': 0, 'type': 'dict'} + + mock_vim_body = {'vnffgd': {'name': 'vnffgd_name', + 'template': {'vnffgd': vnffgd_body}}} + + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'mock_data' + mock_safe_load.return_value = {'id': 0, 'type': 'dict'} + result = os_sfc_utils.create_vnffgd(self.mock_tacker_client, + 'tosca_file', + 'vnffgd_name') + assert result is self.mock_tacker_client.create_vnffgd.return_value + mock_open.assert_called_once_with('tosca_file') + mock_safe_load.assert_called_once_with('mock_data') + self.mock_tacker_client.create_vnffgd.assert_called_once_with( + body=mock_vim_body) + mock_log.info.assert_has_calls(log_calls) + mock_log.error.assert_not_called() + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnffg_returned_none(self, mock_log): + """ + Checks the proper functionality of create_vnffg + function when the vnffgd id or vnffg name is not provided + """ + + log_calls = [[call("Creating the vnffg...")], + [call("error [create_vnffg(tacker_client," + " 'None', 'None', 'None')]: " + "vnffgd id or vnffgd name is required")]] + + result = os_sfc_utils.create_vnffg(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.info.assert_has_calls(log_calls[0]) + mock_log.error.assert_has_calls(log_calls[1]) + + @patch('yaml.safe_load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + @patch('sfc.lib.openstack_utils.get_vnffgd_id', autospec=True) + def test_create_vnffg_vnffgd_id_not_provided(self, + mock_get_vnffgd_id, + mock_log, + mock_open, + mock_safe_load): + """ + Checks the proper functionality of create_vnffg + function when the vnffgd id or vnffg name is not provided + """ + + log_calls = [call('Creating the vnffg...')] + vnffg_calls = [call(body={ + 'vnffg': { + 'attributes': {'param_values': {'type': 'dict', + 'id': 0}}, + 'vnffgd_id': 'mocked_vnffg_id', + 'name': 'vnffg_name', + 'symmetrical': False}})] + mock_get_vnffgd_id.return_value = 'mocked_vnffg_id' + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'data' + mock_safe_load.return_value = {'id': 0, 'type': 'dict'} + + result = os_sfc_utils.create_vnffg(self.mock_tacker_client, + 'vnffg_name', + None, + 'vnffgd_name', + 'param_file') + assert result is self.mock_tacker_client.create_vnffg.return_value + mock_open.assert_called_once_with('param_file') + open_handler.read.assert_called_once_with() + mock_get_vnffgd_id.assert_called_once_with(self.mock_tacker_client, + 'vnffgd_name') + mock_safe_load.assert_called_once_with('data') + mock_log.info.assert_has_calls(log_calls) + self.mock_tacker_client.create_vnffg.assert_has_calls(vnffg_calls) + + @patch('yaml.safe_load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_create_vnffg_vnffgd_id_provided(self, + mock_log, + mock_open, + mock_safe_load): + """ + Checks the proper functionality of create_vnffg + function when the vnffgd id or vnffg name is not provided + """ + + log_calls = [call('Creating the vnffg...')] + vnffg_calls = [call(body={ + 'vnffg': { + 'attributes': {'param_values': {'type': 'dict', + 'id': 0}}, + 'vnffgd_id': 'vnffgd_id', + 'name': 'vnffg_name', + 'symmetrical': False}})] + open_handler = mock_open.return_value.__enter__.return_value + open_handler.read.return_value = 'data' + mock_safe_load.return_value = {'id': 0, 'type': 'dict'} + + result = os_sfc_utils.create_vnffg(self.mock_tacker_client, + 'vnffg_name', + 'vnffgd_id', + 'vnffgd_name', + 'param_file') + assert result is self.mock_tacker_client.create_vnffg.return_value + mock_open.assert_called_once_with('param_file') + open_handler.read.assert_called_once_with() + mock_safe_load.assert_called_once_with('data') + mock_log.info.assert_has_calls(log_calls) + self.mock_tacker_client.create_vnffg.assert_has_calls(vnffg_calls) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vnffgds_returned_none(self, mock_log): + """ + Checks the proper functionality of list_vnffgds + function when list_vnffgds raises an Exception + """ + + self.mock_tacker_client.list_vnffgds.side_effect = Exception( + 'ErrorMSG') + log_calls = [call('Error [list_vnffgds(tacker_client)]: ErrorMSG')] + + result = os_sfc_utils.list_vnffgds(self.mock_tacker_client) + self.assertIsNone(result) + self.mock_tacker_client.list_vnffgds.assert_called_once_with( + retrieve_all=True) + mock_log.error.assert_has_calls(log_calls) + + def test_list_vnffgds(self): + """ + Checks the proper functtionality of list_vnffgds + function + """ + + vnffgds = {'vnffgds': [{'id': 'vnffgd_obj_one'}, + {'id': 'vnffgd_obj_two'}]} + + mock_vnffgds = ['vnffgd_obj_one', 'vnffgd_obj_two'] + + self.mock_tacker_client.list_vnffgds.return_value = vnffgds + result = os_sfc_utils.list_vnffgds(self.mock_tacker_client) + self.assertEqual(mock_vnffgds, result) + self.mock_tacker_client.list_vnffgds.assert_called_once_with( + retrieve_all=True) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vnffgs_returned_none(self, mock_log): + """ + Checks the proper functionality of list_vnffgs + function when list_vnffgs raises an Exception + """ + + self.mock_tacker_client.list_vnffgs.side_effect = Exception('ErrorMSG') + log_calls = [call('Error [list_vnffgs(tacker_client)]: ErrorMSG')] + + result = os_sfc_utils.list_vnffgs(self.mock_tacker_client) + self.assertIsNone(result) + self.mock_tacker_client.list_vnffgs.assert_called_once_with( + retrieve_all=True) + mock_log.error.assert_has_calls(log_calls) + + def test_list_vnffgs(self): + """ + Checks the proper functionality of list_vnffgs + function + """ + + vnffgs = {'vnffgs': [{'id': 'vnffg_obj_one'}, + {'id': 'vnffg_obj_two'}]} + + mock_vnffgs = ['vnffg_obj_one', 'vnffg_obj_two'] + + self.mock_tacker_client.list_vnffgs.return_value = vnffgs + result = os_sfc_utils.list_vnffgs(self.mock_tacker_client) + self.assertEqual(mock_vnffgs, result) + self.mock_tacker_client.list_vnffgs.assert_called_once_with( + retrieve_all=True) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_vnffg_returned_none(self, mock_log): + """ + Checks the proper functionality of delete_vnffg + function + """ + + log_calls = [call("Error [delete_vnffg(tacker_client, 'None', 'None')]" + ": You need to provide a VNFFG id or name")] + + result = os_sfc_utils.delete_vnffg(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.get_vnffg_id', + autospec=True, return_value='vnffg') + def test_delete_vnffg(self, mock_get_vnffg_id): + """ + Checks the proper functionality of delete_vnffg + function + """ + + self.mock_tacker_client.delete_vnffg.return_value = 'deleted' + result = os_sfc_utils.delete_vnffg(self.mock_tacker_client, + None, + 'vnffg_name') + self.assertEqual('deleted', result) + mock_get_vnffg_id.assert_called_once_with(self.mock_tacker_client, + 'vnffg_name') + self.mock_tacker_client.delete_vnffg.assert_called_once_with('vnffg') + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_vnffgd_returned_none(self, mock_log): + """ + Checks the proper functionality of delete_vnffgd + function + """ + + log_calls = [call("Error [delete_vnffgd(tacker_client, 'None', 'None')" + "]: You need to provide VNFFGD id or VNFFGD name")] + + result = os_sfc_utils.delete_vnffgd(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.get_vnffgd_id', + autospec=True, return_value='vnffgd') + def test_delete_vnffgd(self, mock_get_vnffgd_id): + """ + Checks the proper functionality of delete_vnffgd + function + """ + + self.mock_tacker_client.delete_vnffgd.return_value = 'deleted' + result = os_sfc_utils.delete_vnffgd(self.mock_tacker_client, + None, + 'vnffgd_name') + self.assertEqual('deleted', result) + mock_get_vnffgd_id.assert_called_once_with(self.mock_tacker_client, + 'vnffgd_name') + self.mock_tacker_client.delete_vnffgd.assert_called_once_with('vnffgd') + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_list_vims_returned_none(self, mock_log): + """ + Checks the proper functionality of list_vims + function when VNFFGD id is not provided + """ + + self.mock_tacker_client.list_vims.side_effect = Exception('ErrorMSG') + log_calls = [call('Error [list_vims(tacker_client)]: ErrorMSG')] + + result = os_sfc_utils.list_vims(self.mock_tacker_client) + self.assertIsNone(result) + self.mock_tacker_client.list_vims.assert_called_once_with( + retrieve_all=True) + mock_log.error.assert_has_calls(log_calls) + + def test_list_vims(self): + """ + Checks the proper functionality list_vims + function + """ + + vims = {'vims': [{'id': 'vim_obj_1'}, + {'id': 'vim_obj_2'}]} + + mock_vims = ['vim_obj_1', 'vim_obj_2'] + + self.mock_tacker_client.list_vims.return_value = vims + result = os_sfc_utils.list_vims(self.mock_tacker_client) + self.assertEqual(mock_vims, result) + self.mock_tacker_client.list_vims.assert_called_once_with( + retrieve_all=True) + + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_delete_vim_returned_none(self, mock_log): + """ + Checks the proper functionality of delete_vim + function when VIM id and VIM name is not provided + """ + + log_calls = [call("Error [delete_vim(tacker_client, '%s', '%s')]: %s" + % (None, None, 'You need to provide ' + 'VIM id or VIM name'))] + + result = os_sfc_utils.delete_vim(self.mock_tacker_client) + self.assertIsNone(result) + mock_log.error.assert_has_calls(log_calls) + + @patch('sfc.lib.openstack_utils.get_vim_id', + autospec=True, return_value='vim_id') + def test_delete_vim(self, mock_get_vim_id): + """ + Checks the proper functionality of delete_vim + function + """ + + result = os_sfc_utils.delete_vim(self.mock_tacker_client, + None, + 'vim_name') + assert result is self.mock_tacker_client.delete_vim.return_value + mock_get_vim_id.assert_called_once_with(self.mock_tacker_client, + 'vim_name') + self.mock_tacker_client.delete_vim.assert_called_once_with('vim_id') + + @patch('sfc.lib.openstack_utils.get_tacker_client', + autospec=True, return_value='tacker_client_obj') + @patch('sfc.lib.openstack_utils.logger', autospec=True) + def test_get_tacker_items(self, + mock_log, + mock_tacker_client): + """ + Checks the proper functionality of get_tacker_items + function + """ + + mock_dict = {'list_vims': DEFAULT, + 'list_vnfds': DEFAULT, + 'list_vnfs': DEFAULT, + 'list_vnffgds': DEFAULT, + 'list_vnffgs': DEFAULT} + with patch.multiple('sfc.lib.openstack_utils', + **mock_dict) as mock_values: + + os_sfc_utils.get_tacker_items() + + mock_tacker_client.assert_called_once_with() + self.assertEqual(5, mock_log.debug.call_count) + for key in mock_values: + mock_values[key].assert_called_once_with('tacker_client_obj') + + @patch('json.dump', autospec=True) + @patch('json.load', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.create_vim', autospec=True) + def test_register_vim(self, + mock_create_vim, + mock_open, + mock_json_loads, + mock_json_dump): + """ + Checks the proper functionality of register_vim + function + """ + + tmp_file = '/tmp/register-vim.json' + open_handler = mock_open.return_value.__enter__.return_value + open_calls = [call('vim_file'), + call(tmp_file, 'w')] + + mock_json_loads.return_value = {'vim': {'auth_cred': + {'password': None}, + 'auth_url': None}} + + json_dict = {'vim': {'auth_cred': {'password': 'os_auth_cred'}, + 'auth_url': 'os_auth_url'}} + + patch_dict = {'OS_AUTH_URL': 'os_auth_url', + 'OS_PASSWORD': 'os_auth_cred'} + + with patch.dict('os.environ', patch_dict): + os_sfc_utils.register_vim(self.mock_tacker_client, 'vim_file') + mock_json_loads.assert_called_once_with(open_handler) + mock_json_dump.assert_called_once_with(json_dict, + mock_open(tmp_file, 'w')) + mock_open.assert_has_calls(open_calls, any_order=True) + mock_create_vim.assert_called_once_with(self.mock_tacker_client, + vim_file=tmp_file) + + @patch('json.dump', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.create_vnf', autospec=True) + @patch('os.path.join', + autospec=True, return_value='/tmp/param_av_zone.json') + def test_create_vnf_in_av_zone(self, + mock_path_join, + mock_create_vnf, + mock_open, + mock_json_dump): + """ + Checks the proper fucntionality of test_create_vnf_in_av_zone + fucntion + """ + + data = {'zone': 'av::zone'} + param_file = '/tmp/param_av_zone.json' + os_sfc_utils.create_vnf_in_av_zone(self.mock_tacker_client, + 'vnf_name', + 'vnfd_name', + 'vim_name', + 'param_file', + 'av::zone') + open_handler = mock_open.return_value.__enter__.return_value + mock_path_join.assert_called_once_with('/tmp', 'param_av_zone.json') + mock_open.assert_called_once_with(param_file, 'w+') + mock_json_dump.assert_called_once_with(data, open_handler) + mock_create_vnf.assert_called_once_with(self.mock_tacker_client, + 'vnf_name', + vnfd_name='vnfd_name', + vim_name='vim_name', + param_file=param_file) + + @patch('json.dump', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('sfc.lib.openstack_utils.create_vnffg', autospec=True) + @patch('os.path.join', autospec=True, return_value='/tmp/param_name.json') + def test_create_vnffg_with_param_file(self, + mock_path_join, + mock_create_vnffg, + mock_open, + mock_json_dump): + """ + Checks the proper functionality of create_vnffg_with_param_file + function + """ + + data = { + 'ip_dst_prefix': 'server_ip', + 'net_dst_port_id': 'server_port', + 'net_src_port_id': 'client_port' + } + param_file = '/tmp/param_name.json' + os_sfc_utils.create_vnffg_with_param_file(self.mock_tacker_client, + 'vnffgd_name', + 'vnffg_name', + 'default_param_file', + 'client_port', + 'server_port', + 'server_ip') + open_handler = mock_open.return_value.__enter__.return_value + mock_path_join.assert_called_once_with('/tmp', 'param_vnffg_name.json') + mock_open.assert_called_once_with(param_file, 'w+') + mock_json_dump.assert_called_once_with(data, open_handler) + mock_create_vnffg.assert_called_once_with(self.mock_tacker_client, + vnffgd_name='vnffgd_name', + vnffg_name='vnffg_name', + param_file=param_file, + symmetrical=True) diff --git a/sfc/unit_tests/unit/lib/test_test_utils.py b/sfc/unit_tests/unit/lib/test_test_utils.py new file mode 100644 index 00000000..a7d2bfde --- /dev/null +++ b/sfc/unit_tests/unit/lib/test_test_utils.py @@ -0,0 +1,543 @@ +#!/usr/bin/env python + +############################################################################### +# Copyright (c) 2018 Venkata Harshavardhan Reddy Allu 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 + +from mock import Mock +from mock import call +from mock import patch + +import sfc.lib.test_utils as test_utils + +__author__ = "Harshavardhan Reddy <venkataharshavardhan_ven@srmuniv.edu.in>" + + +class SfcTestUtilsTesting(unittest.TestCase): + + def setUp(self): + self.ip = '10.10.10.10' + + @patch('subprocess.PIPE', autospec=True) + @patch('subprocess.Popen', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + def test_run_cmd(self, + mock_log, + mock_popen, + mock_pipe): + """ + Checks the proper functionality of run_cmd + function + """ + + cmd = 'mock_command' + log_calls = [call('Running [mock_command] returns: [0] ' + + '- STDOUT: "output" - STDERR: "output"')] + + pipe_mock = Mock() + attrs = {'communicate.return_value': ('output', 'error'), + 'returncode': 0} + pipe_mock.configure_mock(**attrs) + mock_popen.return_value = pipe_mock + result = test_utils.run_cmd(cmd) + self.assertTupleEqual(result, (0, 'output', 'error')) + mock_popen.assert_called_once_with(cmd, + shell=True, + stdout=mock_pipe, + stderr=mock_pipe) + mock_popen.return_value.communicate.assert_called_once_with() + mock_log.debug.assert_has_calls(log_calls) + + @patch('sfc.lib.test_utils.run_cmd', autospec=True) + def test_run_cmd_remote(self, mock_run_cmd): + """ + Checks the proper functionality of the run_cmd_remote + function + """ + + cmd = 'mock_command' + mock_rc = 'sshpass -p opnfv ssh -q -o UserKnownHostsFile=/dev/null' + \ + ' -o StrictHostKeyChecking=no -o ConnectTimeout=50 ' + \ + ' root@10.10.10.10 mock_command' + test_utils.run_cmd_remote(self.ip, cmd) + mock_run_cmd.assert_called_once_with(mock_rc) + + @patch('shutil.copyfileobj') + @patch('urllib.urlopen', autospec=True) + @patch('__builtin__.open', autospec=True) + def test_download_url_with_exception(self, + mock_open, + mock_urlopen, + mock_copyfileobj): + """ + Checks the proper functionality of download_url + function when an exception is raised + """ + + dest_path = 'mocked_/dest_/path' + url = 'mocked_/url' + mock_urlopen.side_effect = Exception('HttpError') + self.assertFalse(test_utils.download_url(url, dest_path)) + mock_urlopen.assert_called_once_with(url) + mock_open.assert_not_called() + mock_copyfileobj.assert_not_called() + + @patch('urllib.urlopen', autospec=True) + @patch('__builtin__.open', autospec=True) + @patch('shutil.copyfileobj', autospec=True) + def test_download_url_without_exception(self, + mock_copyfileobj, + mock_open, + mock_urlopen): + """ + Checks the proper functionality of download_url + function when any exception isn't raised + """ + + response = '<mocked_response>' + dest_path = 'mocked_/dest_/path' + url = 'mocked_/url' + open_handler = mock_open.return_value.__enter__.return_value + mock_urlopen.return_value = response + self.assertTrue(test_utils.download_url(url, dest_path)) + mock_urlopen.assert_called_once_with(url) + mock_open.assert_called_once_with('mocked_/dest_/path/url', 'wb') + mock_copyfileobj.assert_called_once_with(response, open_handler) + + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.download_url', autospec=True) + @patch('os.path.isfile', autospec=True, return_value=False) + @patch('os.path.dirname', autospec=True, return_value='mocked_') + @patch('os.path.basename', autospec=True, return_value='image_path') + def test_download_image_file_not_found(self, + mock_basename, + mock_dirname, + mock_isfile, + mock_download_url, + mock_log): + """ + Checks the proper functionality of download_image + function when the image file was not found locally + """ + + url = 'mocked_/url' + image_path = 'mocked_/image_path' + log_calls = [call('Downloading image')] + test_utils.download_image(url, image_path) + mock_log.info.assert_has_calls(log_calls) + mock_basename.assert_called_once_with(image_path) + mock_dirname.assert_called_once_with(image_path) + mock_isfile.assert_called_once_with(image_path) + mock_download_url.assert_called_once_with('mocked_/url/image_path', + 'mocked_') + + @patch('sfc.lib.test_utils.download_url') + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('os.path.isfile', autospec=True, return_value=True) + @patch('os.path.dirname', autospec=True, return_value='mocked_') + @patch('os.path.basename', autospec=True, return_value='image_path') + def test_download_image_file_found(self, + mock_basename, + mock_dirname, + mock_isfile, + mock_log, + mock_download_url): + """ + Checks the proper functionality of download_image + function when the image file was found locally + """ + + url = 'mocked_/url' + image_path = 'mocked_/image_path' + log_calls = [call('Using old image')] + test_utils.download_image(url, image_path) + mock_log.info.assert_has_calls(log_calls) + mock_basename.assert_called_once_with(image_path) + mock_dirname.assert_called_once_with(image_path) + mock_isfile.assert_called_once_with(image_path) + mock_download_url.assert_not_called() + + @patch('sfc.lib.test_utils.run_cmd', autospec=True) + def test_ping_gets_error(self, mock_run_cmd): + """ + Checks the proper functionality of ping + function when run_cmd returns non-zero returncode + """ + + mock_cmd = 'ping -c1 -w1 %s' % self.ip + mock_run_cmd.return_value = (1, '', '') + self.assertFalse(test_utils.ping(self.ip, 1)) + mock_run_cmd.assert_called_once_with(mock_cmd) + + @patch('sfc.lib.test_utils.run_cmd', autospec=True) + def test_ping_gets_no_error(self, mock_run_cmd): + """ + Checks the proper functionality of ping + function when run_cmd returns zero as returncode + """ + + mock_cmd = 'ping -c1 -w1 %s' % self.ip + mock_run_cmd.return_value = (0, '', '') + self.assertTrue(test_utils.ping(self.ip, 1)) + mock_run_cmd.assert_called_once_with(mock_cmd) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_start_http_server_returned_false_failed_to_start( + self, mock_run_cmd_remote, mock_log, mock_sleep): + """ + Checks the proper functionality of start_http_server + function when http_server is failed to start + """ + + cmd = "\'python -m SimpleHTTPServer 80 " + \ + "> /dev/null 2>&1 &\'" + + rcr_calls = [call(self.ip, cmd), + call(self.ip, 'ps aux | grep SimpleHTTPServer')] + log_calls = [call('Failed to start http server')] + + mock_run_cmd_remote.side_effect = [('', '', ''), + ('', '', '')] + + result = test_utils.start_http_server(self.ip, 1) + self.assertFalse(result) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_sleep.assert_called_once_with(3) + mock_log.error.assert_has_calls(log_calls) + mock_log.info.assert_not_called() + mock_log.debug.assert_not_called() + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_start_http_server_returned_false_port_is_down( + self, mock_run_cmd_remote, mock_log, mock_sleep): + """ + Checks the proper functionality of start_http_server + function when port 80 is down + """ + + cmd = "\'python -m SimpleHTTPServer 80 " + \ + "> /dev/null 2>&1 &\'" + + rcr_calls = [call(self.ip, cmd), + call(self.ip, 'ps aux | grep SimpleHTTPServer'), + call(self.ip, 'netstat -pntl | grep :80')] + + log_calls = [call('output'), + call('Port 80 is not up yet')] + + mock_run_cmd_remote.side_effect = [('', '', ''), + ('', 'output', ''), + ('', '', '')] + + result = test_utils.start_http_server(self.ip, 1) + self.assertFalse(result) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_sleep.assert_called_with(5) + mock_log.info.assert_has_calls(log_calls[:1]) + mock_log.debug.assert_has_calls(log_calls[1:]) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_start_http_server_returned_true(self, + mock_run_cmd_remote, + mock_log, + mock_sleep): + """ + Checks the proper functionality of start_http_server + function when the port 80 is up + """ + + cmd = "\'python -m SimpleHTTPServer 80 " + \ + "> /dev/null 2>&1 &\'" + + rcr_calls = [call(self.ip, cmd), + call(self.ip, 'ps aux | grep SimpleHTTPServer'), + call(self.ip, 'netstat -pntl | grep :80')] + + log_calls = [call('output')] + + mock_run_cmd_remote.side_effect = [('', '', ''), + ('', 'output', ''), + ('', 'output', '')] + + self.assertTrue(test_utils.start_http_server(self.ip, 1)) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_sleep.assert_called_once_with(3) + mock_log.info.assert_has_calls(log_calls) + mock_log.debug.assert_not_called() + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_start_vxlan_tool_returned_false(self, + mock_run_cmd_remote, + mock_log, + mock_sleep): + """ + Checks the proper functionality of start_vxlan_tool + function when no output is returned on ps command + """ + + mock_run_cmd_remote.side_effect = [('', 'output', ''), + ('', '', '')] + + mock_rc = 'nohup python /root/vxlan_tool.py --do ' + \ + 'forward --interface eth0 > /dev/null 2>&1 &' + + rcr_calls = [call(self.ip, mock_rc), + call(self.ip, 'ps aux | grep vxlan_tool')] + + log_calls = [call('Failed to start the vxlan tool')] + + self.assertFalse(test_utils.start_vxlan_tool(self.ip)) + mock_sleep.assert_called_once_with(3) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_log.error.assert_has_calls(log_calls) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_start_vxlan_tool_returned_output(self, + mock_run_cmd_remote, + mock_log, + mock_sleep): + """ + Checks the proper functionality of start_vxlan_tool + function when output is returned on ps command + """ + + mock_run_cmd_remote.side_effect = [('', 'output', ''), + ('', 'output', '')] + + mock_rc = 'nohup python /root/vxlan_tool.py --do ' + \ + 'forward --interface eth0 > /dev/null 2>&1 &' + + rcr_calls = [call(self.ip, mock_rc), + call(self.ip, 'ps aux | grep vxlan_tool')] + + self.assertIsNotNone(test_utils.start_vxlan_tool(self.ip)) + mock_sleep.assert_called_once_with(3) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_log.error.assert_not_called() + + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_stop_vxlan_tool(self, mock_run_cmd_remote): + """ + Checks the proper functionality of stop_vxlan_tool + function + """ + + mock_rc = 'pkill -f vxlan_tool.py' + test_utils.stop_vxlan_tool(self.ip) + mock_run_cmd_remote.assert_called_once_with(self.ip, mock_rc) + + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_netcat(self, + mock_run_cmd_remote, + mock_log): + """ + Checks the proper functionality of netcat + function + """ + + dest_ip = 'mock_destination_ip' + c = 'nc -z -w 5 %s 1234' % dest_ip + log_calls = [call('Running [%s] from [%s] returns [0]' % (c, self.ip))] + mock_run_cmd_remote.return_value = (0, '', '') + result = test_utils.netcat(self.ip, dest_ip, 1234) + self.assertEqual(result, 0) + mock_log.info.assert_has_calls(log_calls) + + @patch('sfc.lib.test_utils.netcat', autospec=True) + def test_is_ssh_blocked_returned_false(self, mock_netcat): + """ + Checks the proper funcitonality of is_ssh_blocked + function when the returncode is zero + """ + + dest_ip = 'mock_destination_ip' + nc_calls = [call('10.10.10.10', + 'mock_destination_ip', + destination_port='22', + source_port=None)] + + mock_netcat.return_value = 0 + self.assertFalse(test_utils.is_ssh_blocked(self.ip, dest_ip)) + mock_netcat.assert_has_calls(nc_calls) + + @patch('sfc.lib.test_utils.netcat', autospec=True) + def test_is_ssh_blocked_returned_true(self, mock_netcat): + """ + Checks the proper funcitonality of is_ssh_blocked + function when the returncode is non-zero integer + """ + + dest_ip = 'mock_destination_ip' + nc_calls = [call('10.10.10.10', + 'mock_destination_ip', + destination_port='22', + source_port=None)] + + mock_netcat.return_value = 1 + self.assertTrue(test_utils.is_ssh_blocked(self.ip, dest_ip)) + mock_netcat.assert_has_calls(nc_calls) + + @patch('sfc.lib.test_utils.netcat', autospec=True) + def test_is_http_blocked_returned_false(self, mock_netcat): + """ + Checks the proper funcitonality of is_http_blocked + function when the returncode is zero + """ + + dest_ip = 'mock_destination_ip' + nc_calls = [call('10.10.10.10', + 'mock_destination_ip', + destination_port='80', + source_port=None)] + + mock_netcat.return_value = 0 + self.assertFalse(test_utils.is_http_blocked(self.ip, dest_ip)) + mock_netcat.assert_has_calls(nc_calls) + + @patch('sfc.lib.test_utils.netcat', autospec=True) + def test_is_http_blocked_returned_true(self, mock_netcat): + """ + Checks the proper funcitonality of is_http_blocked + function when the returncode is non-zero integer + """ + + dest_ip = 'mock_destination_ip' + nc_calls = [call('10.10.10.10', + 'mock_destination_ip', + destination_port='80', + source_port=None)] + + mock_netcat.return_value = 1 + self.assertTrue(test_utils.is_http_blocked(self.ip, dest_ip)) + mock_netcat.assert_has_calls(nc_calls) + + @patch('time.strftime', autospec=True) + @patch('opnfv.utils.ovs_logger.OVSLogger', autospec=True) + def test_capture_ovs_logs(self, + mock_ovs_log, + mock_strftime): + """ + Checks the proper functionality of capture_ovs_logs + function + """ + + log_calls = [call('controller_clients', + 'compute_clients', + 'error', + 'date_time')] + + mock_strftime.return_value = 'date_time' + test_utils.capture_ovs_logs(mock_ovs_log, + 'controller_clients', + 'compute_clients', + 'error') + + mock_strftime.assert_called_once_with('%Y%m%d-%H%M%S') + mock_ovs_log.dump_ovs_logs.assert_has_calls(log_calls) + + def test_get_ssh_clients(self): + """ + Checks the proper functionality of get_ssh_clients + fucntion + """ + + mock_node_obj_one = Mock() + mock_node_obj_two = Mock() + mock_node_obj_one.ssh_client = 'ssh_client_one' + mock_node_obj_two.ssh_client = 'ssh_client_two' + nodes = [mock_node_obj_one, mock_node_obj_two] + result = test_utils.get_ssh_clients(nodes) + self.assertEqual(result, ['ssh_client_one', 'ssh_client_two']) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_check_ssh_returned_false(self, + mock_run_cmd_remote, + mock_log, + mock_sleep): + """ + Checks the proper functionality of check_ssh + fucntion when few VNFs can't establish SSH connectivity + """ + + ips = ["ip_address-1", + "ip_address-2"] + + rcr_calls = [call(ips[0], 'exit'), + call(ips[1], 'exit')] + + log_calls = [call('Checking SSH connectivity ' + + 'to the SFs with ips %s' % str(ips))] + + mock_run_cmd_remote.side_effect = [(1, '', ''), + (0, '', '')] + + self.assertFalse(test_utils.check_ssh(ips, retries=1)) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_log.info.assert_has_calls(log_calls) + mock_sleep.assert_called_once_with(3) + + @patch('time.sleep', autospec=True) + @patch('sfc.lib.test_utils.logger', autospec=True) + @patch('sfc.lib.test_utils.run_cmd_remote', autospec=True) + def test_check_ssh_returned_true(self, + mock_run_cmd_remote, + mock_log, + mock_sleep): + """ + Checks the proper functionality of check_ssh + fucntion when all VNFs can establish SSH connectivity + """ + + ips = ["ip_address-1", + "ip_address-2"] + + rcr_calls = [call(ips[0], 'exit'), + call(ips[1], 'exit')] + + log_calls = [call('Checking SSH connectivity to ' + + 'the SFs with ips %s' % str(ips)), + call('SSH connectivity to the SFs established')] + + mock_run_cmd_remote.side_effect = [(0, '', ''), + (0, '', '')] + + self.assertTrue(test_utils.check_ssh(ips, retries=1)) + mock_run_cmd_remote.assert_has_calls(rcr_calls) + mock_log.info.assert_has_calls(log_calls) + mock_sleep.assert_not_called() + + def test_fill_installer_dict(self): + """ + Checks the proper functionality of fill_installer_dict + function + """ + + installer_type = 'mock_installer' + installer_yaml_fields = { + 'user': 'defaults.installer.mock_installer.user', + 'password': 'defaults.installer.mock_installer.password', + 'cluster': 'defaults.installer.mock_installer.cluster', + 'pkey_file': 'defaults.installer.mock_installer.pkey_file' + } + result = test_utils.fill_installer_dict(installer_type) + self.assertDictEqual(result, installer_yaml_fields) |