aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--sfc/unit_tests/unit/lib/test_openstack_utils.py346
1 files changed, 227 insertions, 119 deletions
diff --git a/sfc/unit_tests/unit/lib/test_openstack_utils.py b/sfc/unit_tests/unit/lib/test_openstack_utils.py
index 520c8579..2407415e 100644
--- a/sfc/unit_tests/unit/lib/test_openstack_utils.py
+++ b/sfc/unit_tests/unit/lib/test_openstack_utils.py
@@ -465,26 +465,18 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
@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.FloatingIpConfig', autospec=True)
@patch('sfc.lib.openstack_utils.cr_inst.generate_creator', autospec=True)
- def test_assign_floating_ip_vnfs_raised_exception(self,
- mock_generate_creator,
- mock_floating_ip_config,
- mock_get_stack_servers,
- mock_log):
+ 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",
- "The VNF server_name-float does not have any suitable"
- " port for floating IP assignment"]
- log_calls = [call(ErrorMSG[0]),
- call(ErrorMSG[1])]
-
+ 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']
@@ -494,14 +486,12 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
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:
- # ips = mock_ips
self.os_sfc.assign_floating_ip_vnfs('router', mock_ips)
- self.assertEqual(cm.exception.message, ErrorMSG[0])
+ self.assertEqual(cm.exception.message, ErrorMSG)
mock_get_stack_servers.assert_called_once_with(self.heat,
self.nova,
self.neutron,
@@ -512,14 +502,47 @@ class SfcOpenStackUtilsTesting(unittest.TestCase):
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:
- # ips = None
self.os_sfc.assign_floating_ip_vnfs('router')
- self.assertEqual(cm.exception.message, ErrorMSG[1])
+ mock_get_stack_servers.assert_called_once_with(self.heat,
+ self.nova,
+ self.neutron,
+ 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)
- mock_floating_ip_config.assert_not_called()
@patch('sfc.lib.openstack_utils.FloatingIpConfig', autospec=True)
@patch('sfc.lib.openstack_utils.cr_inst.generate_creator',
@@ -1113,39 +1136,49 @@ class SfcTackerSectionTesting(unittest.TestCase):
mock_log.assert_not_called()
self.assertEqual([1, 2], result)
- @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(self,
- mock_log,
- mock_open,
- mock_safe_load):
+ 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("VNFD template:\nmock_vnfd"),
- call("Error [create_vnfd(tacker_client, 'tosca_file')]: "
+ call("Error [create_vnfd(tacker_client, 'None')]: "
"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')
-
- # tosca_file = None
result = os_sfc_utils.create_vnfd(self.mock_tacker_client,
None,
'vnfd_name')
self.assertIsNone(result)
- mock_open.assert_not_called()
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:])
- # tosca_file = 'tosca_file'
+ @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')
@@ -1153,7 +1186,6 @@ class SfcTackerSectionTesting(unittest.TestCase):
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:])
@@ -1248,25 +1280,32 @@ class SfcTackerSectionTesting(unittest.TestCase):
retrieve_all=True)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- def test_create_vnf_returned_none(self, mock_log):
+ 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"),
- call("error [create_vnf(tacker_client,"
- " 'vnf_name', 'None', 'vnfd_name')]: "
- "vim id or vim name is required")]]
-
- # vnfd_name = None
+ 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
+ """
- # vnfd_name = 'vnfd_name'
+ 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,
@@ -1274,8 +1313,8 @@ class SfcTackerSectionTesting(unittest.TestCase):
None,
None)
self.assertIsNone(result)
- mock_log.info.assert_has_calls(log_calls[0] * 2)
- mock_log.error.assert_has_calls(log_calls[2:])
+ 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)
@@ -1283,11 +1322,11 @@ class SfcTackerSectionTesting(unittest.TestCase):
autospec=True, return_value='vim_id')
@patch('sfc.lib.openstack_utils.get_vnfd_id',
autospec=True, return_value='vnfd_id')
- def test_create_vnf(self,
- mock_get_vnfd_id,
- mock_get_vim_id,
- mock_log,
- mock_open):
+ 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
@@ -1296,13 +1335,9 @@ class SfcTackerSectionTesting(unittest.TestCase):
'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'
- # vnfd_name = 'vnfd_name', vim_name = 'vim_name'
- # param_file = 'param_file'
result = os_sfc_utils.create_vnf(self.mock_tacker_client,
'vnf_name',
None,
@@ -1310,6 +1345,7 @@ class SfcTackerSectionTesting(unittest.TestCase):
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')
@@ -1319,8 +1355,21 @@ class SfcTackerSectionTesting(unittest.TestCase):
self.mock_tacker_client.create_vnf.assert_called_once_with(
body=mock_body)
- # vnfd_id = 'vnfd_id', vnfd_name = 'vnfd_name'
- # vim_id = 'vim_id', vim_name = 'vim_name'
+ @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',
@@ -1328,29 +1377,38 @@ class SfcTackerSectionTesting(unittest.TestCase):
'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.get_vnf_id',
- autospec=True, return_value=None)
@patch('sfc.lib.openstack_utils.logger', autospec=True)
- def test_get_vnf_returned_none(self,
- mock_log,
- mock_get_vnf_id):
+ 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"),
- call("Could not retrieve VNF [vnf_id=None, "
- "vnf_name=vnf_name] - Could not retrieve ID from "
- "name [vnf_name]")]
+ " - You must specify vnf_id or vnf_name")]
- # 'vnf_name' = None
result = os_sfc_utils.get_vnf(self.mock_tacker_client)
self.assertIsNone(result)
+ mock_log.error.assert_has_calls(log_calls)
- # 'vnf_name' = 'vnf_name'
+ @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')
@@ -1397,61 +1455,94 @@ class SfcTackerSectionTesting(unittest.TestCase):
None)
mock_json_loads.assert_called_once_with(vnf['mgmt_url'])
- @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(self,
- mock_get_vnf,
- mock_log,
- mock_sleep):
+ 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_vnf_values = [None,
- {'id': 'vnf_id',
- 'status': 'ERROR'},
- {'id': 'vnf_id',
- 'status': 'PEDNING_CREATE'},
- {'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("Waiting for vnf %s" % str(mock_vnf_values[2])),
- call("error [wait_for_vnf(tacker_client, 'vnf_id', "
+ 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'"),
- call("error [wait_for_vnf(tacker_client, 'vnf_id', "
- "'vnf_name')]: Error when booting vnf vnf_id"),
- call("error [wait_for_vnf(tacker_client, 'vnf_id', "
- "'vnf_name')]: Timeout when booting vnf vnf_id")]
+ " name='vnf_name'")]
- # not able to retrieve VNF
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")]
- # not able to boot VNF
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")]
- # timeout booting VNF
result = os_sfc_utils.wait_for_vnf(self.mock_tacker_client,
'vnf_id',
'vnf_name',
0)
self.assertIsNone(result)
- mock_log.info.assert_has_calls(log_calls[:2])
- mock_log.error.assert_has_calls(log_calls[2:])
+ 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)
@@ -1629,39 +1720,29 @@ class SfcTackerSectionTesting(unittest.TestCase):
@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(self,
- mock_get_vnffgd_id,
- mock_log,
- mock_open,
- mock_safe_load):
+ 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
"""
- mock_get_vnffgd_id.return_value = 'mocked_vnffg_id'
- log_calls = [call('Creating the vnffg...')] * 2
-
+ 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}}),
- call(body={
- 'vnffg': {
- 'attributes': {'param_values': {'type': 'dict',
- 'id': 0}},
- 'vnffgd_id': 'vnffgd_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'}
- # vnffgd_id = None
result = os_sfc_utils.create_vnffg(self.mock_tacker_client,
'vnffg_name',
None,
@@ -1673,16 +1754,43 @@ class SfcTackerSectionTesting(unittest.TestCase):
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'}
- # vnffgd_id = 'vnffgd_id'
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)
- mock_log.error.assert_not_called()
self.mock_tacker_client.create_vnffg.assert_has_calls(vnffg_calls)
@patch('sfc.lib.openstack_utils.logger', autospec=True)