aboutsummaryrefslogtreecommitdiffstats
path: root/sfc/unit_tests
diff options
context:
space:
mode:
Diffstat (limited to 'sfc/unit_tests')
-rw-r--r--sfc/unit_tests/unit/lib/test_cleanup.py84
-rw-r--r--sfc/unit_tests/unit/lib/test_odl_utils.py40
-rw-r--r--sfc/unit_tests/unit/lib/test_openstack_utils.py264
3 files changed, 304 insertions, 84 deletions
diff --git a/sfc/unit_tests/unit/lib/test_cleanup.py b/sfc/unit_tests/unit/lib/test_cleanup.py
index 5ec4261e..b83e229f 100644
--- a/sfc/unit_tests/unit/lib/test_cleanup.py
+++ b/sfc/unit_tests/unit/lib/test_cleanup.py
@@ -262,8 +262,10 @@ class SfcCleanupTesting(unittest.TestCase):
mock_log.assert_has_calls(log_calls)
mock_del_vim.assert_has_calls(del_calls)
+ @patch('sfc.lib.cleanup.logger.info')
@patch('sfc.lib.cleanup.logger.error')
- def test_delete_openstack_objects_exception(self, mock_log):
+ def test_delete_openstack_objects_exception(self, mock_log_err,
+ mock_log_info):
"""
Check the proper functionality of the delete_openstack_objects
@@ -283,11 +285,13 @@ class SfcCleanupTesting(unittest.TestCase):
mock_creator_objs_list = [mock_creator_obj_one, mock_creator_obj_two]
log_calls = [call('Unexpected error cleaning - %s', exception_two),
- call('Unexpected error cleaning - %s', exception_one)]
+ call('Unexpected error cleaning - %s', exception_one),
+ call('Deleting the openstack objects...')]
cleanup.delete_openstack_objects(mock_creator_objs_list)
- mock_log.assert_has_calls(log_calls)
+ mock_log_err.assert_has_calls(log_calls[:2])
+ mock_log_info.assert_has_calls(log_calls[2:])
@patch('sfc.lib.openstack_utils.OpenStackSFC', autospec=True)
def test_delete_untracked_security_groups(self,
@@ -314,53 +318,71 @@ class SfcCleanupTesting(unittest.TestCase):
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')
- @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_time):
+ 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,
- 'delete_untracked_security_groups': DEFAULT}
+ 'delete_vims': DEFAULT}
with patch.multiple('sfc.lib.cleanup',
**mock_dict) as mock_values:
-
- cleanup.cleanup(['creator_one', 'creator_two'],
- self.odl_ip,
- self.odl_port)
+ 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.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(['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(['creator_one', 'creator_two'])
- mock_time.assert_called_once_with(20)
+ mock_cleanup_mano.assert_called_once_with('mano')
+ mock_untr_sec_grps.assert_called_once()
- @patch('time.sleep')
+ @patch('sfc.lib.cleanup.cleanup_mano_objects')
@patch('sfc.lib.cleanup.cleanup_odl')
def test_cleanup_from_bash(self,
mock_cleanup_odl,
- 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:
+ mock_cleanup_mano):
- cleanup.cleanup_from_bash(self.odl_ip,
- self.odl_port)
+ cleanup.cleanup_from_bash(self.odl_ip,
+ self.odl_port,
+ 'mano')
- for key in mock_values:
- mock_values[key].assert_called_once()
mock_cleanup_odl.assert_called_once_with(self.odl_ip,
self.odl_port)
- mock_time.assert_called_once_with(20)
+ 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
index d151a1ca..17ad88f5 100644
--- a/sfc/unit_tests/unit/lib/test_odl_utils.py
+++ b/sfc/unit_tests/unit/lib/test_odl_utils.py
@@ -733,43 +733,3 @@ class SfcOdlUtilsTesting(unittest.TestCase):
'compute_nodes')
assert mock_sleep.call_count == 2
mock_log.assert_not_called()
-
- @patch('os.path.join', autospec=True)
- @patch('sfc.lib.odl_utils.os_sfc_utils.create_vnffg_with_param_file',
- autospec=True)
- @patch('sfc.lib.odl_utils.os_sfc_utils.create_vnffgd', autospec=True)
- def test_create_chain(self, mock_create_vnffgd,
- mock_create_vnffg_with_param_file,
- mock_join):
- """
- Checks the proper functionality of create_chain
- function
- """
-
- mock_join.return_value = '/tosca_file'
- mock_neutron = Mock()
- mock_common_config = Mock()
- mock_testcase_config = Mock()
- mock_common_config.vnffgd_dir = 'mock_vnffgd_dir'
- mock_common_config.sfc_test_dir = 'mock_sfc_test_dir'
- mock_testcase_config.test_vnffgd_red = 'mock_test_vnffgd_red'
- mock_neutron.id = 'mock_id'
-
- odl_utils.create_chain('tacker_client',
- 'default_param_file',
- mock_neutron,
- mock_common_config,
- mock_testcase_config)
-
- mock_join.assert_called_once_with('mock_sfc_test_dir',
- 'mock_vnffgd_dir',
- 'mock_test_vnffgd_red')
- mock_create_vnffgd.assert_called_once_with('tacker_client',
- tosca_file='/tosca_file',
- vnffgd_name='red')
- mock_create_vnffg_with_param_file.assert_called_once_with(
- 'tacker_client',
- 'red',
- 'red_http',
- 'default_param_file',
- 'mock_id')
diff --git a/sfc/unit_tests/unit/lib/test_openstack_utils.py b/sfc/unit_tests/unit/lib/test_openstack_utils.py
index bc08a184..595f09b5 100644
--- a/sfc/unit_tests/unit/lib/test_openstack_utils.py
+++ b/sfc/unit_tests/unit/lib/test_openstack_utils.py
@@ -265,7 +265,8 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
"""
vm_con_ins = mock_vm_instance_config.return_value
- pc_ins = mock_port_config.return_value
+ pc_inss = ['pc_config1', 'pc_config2']
+ mock_port_config.side_effect = pc_inss
os_vm_ins = mock_os_vm_instance.return_value
os_vm_ins_cre = os_vm_ins.create.return_value
expected = (os_vm_ins_cre, os_vm_ins)
@@ -277,19 +278,21 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
img_cre.image_settings = 'image_settings'
log_calls = [call('Creating the instance vm_name...')]
-
+ pc_calls = [call(name='port1', network_name='nw_name'),
+ call(name='port2', network_name='nw_name')]
result = self.os_sfc.create_instance('vm_name',
'flavor_name',
img_cre,
network,
secgrp,
- 'av_zone')
+ 'av_zone',
+ ['port1', 'port2'])
self.assertEqual(expected, result)
mock_vm_instance_config.assert_called_once_with(name='vm_name',
flavor='flavor_name',
security_group_names=''
'sec_grp',
- port_settings=[pc_ins],
+ port_settings=pc_inss,
availability_zone='av'
'_zone')
mock_os_vm_instance.assert_called_once_with(self.os_creds,
@@ -297,6 +300,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
'image_settings')
self.assertEqual([os_vm_ins], self.os_sfc.creators)
mock_log.info.assert_has_calls(log_calls)
+ mock_port_config.assert_has_calls(pc_calls)
@patch('sfc.lib.openstack_utils.nova_utils.get_hypervisor_hosts',
autospec=True)
@@ -510,9 +514,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
@patch('sfc.lib.openstack_utils.logger', autospec=True)
@patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
autospec=True)
- def test_get_client_port_raised_exceptioin(self,
- mock_os_vm,
- mock_log):
+ 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
@@ -527,7 +531,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
" with name mock_vm_name-port")]
with self.assertRaises(Exception) as cm:
- self.os_sfc.get_client_port(mock_vm, mock_os_vm_ins)
+ 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)
@@ -535,9 +539,9 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
@patch('sfc.lib.openstack_utils.logger', autospec=True)
@patch('sfc.lib.openstack_utils.cr_inst.OpenStackVmInstance',
autospec=True)
- def test_get_client_port(self,
- mock_os_vm,
- mock_log):
+ 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
@@ -547,7 +551,7 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
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_client_port(mock_vm, mock_os_vm_ins)
+ result = self.os_sfc.get_instance_port(mock_vm, mock_os_vm_ins)
self.assertEqual('mock_port', result)
@patch('sfc.lib.openstack_utils.neutron_utils.list_security_groups',
@@ -569,9 +573,243 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
mock_list_security_groups.assert_called_once_with(self.neutron)
mock_delete_security_group.assert_has_calls(del_calls)
+ @patch('snaps.openstack.create_instance.OpenStackVmInstance',
+ autospec=True)
+ def test_wait_for_vnf(self, mock_osvminstance):
+ """
+ Checks the method wait_for_vnf
+ """
-class SfcTackerSectionTesting(unittest.TestCase):
+ mock_osvm_ins = mock_osvminstance.return_value
+ mock_osvm_ins.vm_active.return_value = True
+ result = self.os_sfc.wait_for_vnf(mock_osvm_ins)
+ self.assertTrue(result)
+
+ @patch('snaps.domain.vm_inst.VmInst', autospec=True)
+ @patch('sfc.lib.openstack_utils.logger', autospec=True)
+ def test_create_port_groups_raises_exception(self, mock_log, mock_vm):
+ """
+ Checks the create_port_groups when length of ports is greater than 2
+ """
+ mock_vm_ins = mock_vm.return_value
+ mock_vm_ins.name = 'vm'
+ 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, mock_vm_ins)
+ 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('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_pp(self, mock_log,
+ mock_vm,
+ mock_port):
+ """
+ Checks the create_port_groups when something goes wrong in port pair
+ creation
+ """
+ mock_vm_ins = mock_vm.return_value
+ mock_vm_ins.name = 'vm'
+ 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': 'vm'})]
+ mock_port_ins = mock_port.return_value
+ mock_port_ins2 = mock_port.return_value
+ mock_port_ins.id = '123abc'
+ mock_port_ins2.id = '456def'
+ self.neutron.create_sfc_port_pair.return_value = None
+ result = self.os_sfc.create_port_groups(
+ [mock_port_ins, mock_port_ins2], mock_vm_ins)
+ 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_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
+ """
+ mock_vm_ins = mock_vm.return_value
+ mock_vm_ins.name = 'vm'
+ log_calls_info = [call('Creating the port pairs...'),
+ call('Creating the port pair groups...')]
+ log_calls_warn = [call('Chain creation failed due to port pair group '
+ 'creation failed for vnf %(vnf)', 'vm')]
+ mock_port_ins = mock_port.return_value
+ mock_port_ins.id = '123abc'
+ self.neutron.create_sfc_port_pair.return_value = \
+ {'port_pair': {'id': 'pp_id'}}
+ self.neutron.create_sfc_port_pair_group.return_value = None
+ result = self.os_sfc.create_port_groups([mock_port_ins], mock_vm_ins)
+ 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_returns_id(self, mock_log, mock_osvm,
+ mock_port):
+ """
+ Checks the create_port_groups when everything goes as expected
+ """
+
+ log_calls_info = [call('Creating the port pairs...'),
+ call('Creating the port pair groups...')]
+ mock_port_ins = mock_port.return_value
+ mock_port_ins.id = '123abc'
+ mock_osvm_ins = mock_osvm.return_value
+ mock_osvm_ins.name = 'vm'
+ expected_port_pair = {'name': 'vm-connection-points',
+ 'description': 'port pair for vm',
+ 'ingress': '123abc',
+ 'egress': '123abc'}
+ self.neutron.create_sfc_port_pair.return_value = \
+ {'port_pair': {'id': 'pp_id'}}
+ self.neutron.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([mock_port_ins], mock_osvm_ins)
+ self.neutron.create_sfc_port_pair.assert_has_calls(
+ [call({'port_pair': expected_port_pair})])
+ self.neutron.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_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.create_sfc_flow_classifier.return_value = \
+ {'flow_classifier': {'id': 'fc_id'}}
+ self.neutron.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.create_sfc_flow_classifier.assert_has_calls(
+ [call({'flow_classifier': expected_sfc_classifier_params})])
+ self.neutron.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.create_sfc_flow_classifier.return_value = \
+ {'flow_classifier': {'id': 'fc_id'}}
+ self.neutron.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.create_sfc_flow_classifier.assert_has_calls(
+ [call({'flow_classifier': expected_sfc_classifier_params})])
+ self.neutron.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_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.list_sfc_port_pair_groups.return_value = \
+ {'port_pair_groups': [{'id': 'id_ppg1'}, {'id': 'id_ppg2'}]}
+ self.neutron.list_sfc_port_pairs.return_value = \
+ {'port_pairs': [{'id': 'id_pp1'}, {'id': 'id_pp2'}]}
+ self.os_sfc.delete_port_groups()
+
+ self.neutron.delete_sfc_port_pair_group.assert_has_calls(
+ [call('id_ppg1'), call('id_ppg2')])
+ self.neutron.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.list_sfc_port_chains.return_value = \
+ {'port_chains': [{'id': 'id_pc1'}]}
+ self.neutron.list_sfc_flow_classifiers.return_value = \
+ {'flow_classifiers': [{'id': 'id_fc1'}]}
+ self.os_sfc.delete_chain()
+ self.neutron.delete_sfc_port_chain.assert_has_calls([call('id_pc1')])
+ self.neutron.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