aboutsummaryrefslogtreecommitdiffstats
path: root/sfc/unit_tests
diff options
context:
space:
mode:
authorVenkata Harshavardhan Reddy Allu <venkataharshavardhan_ven@srmuniv.edu.in>2018-08-11 21:59:01 +0530
committerVenkata Harshavardhan Reddy Allu <venkataharshavardhan_ven@srmuniv.edu.in>2018-08-22 18:22:48 +0530
commitdeceb71f71bff41d281ca4fff1d2a381d1daea93 (patch)
treeebfe608e3cedec16ee01ee7cac6069746e7d5d19 /sfc/unit_tests
parent47bb560d4939933ed86fbfab1983da4079d2cf64 (diff)
Added unit tests for 'odl_utils.py' module
JIRA: SFC-123 Change-Id: I49b8dae157816ef04a45dea188cc7dbe3aa72d4a Signed-off-by: Venkata Harshavardhan Reddy Allu <venkataharshavardhan_ven@srmuniv.edu.in>
Diffstat (limited to 'sfc/unit_tests')
-rw-r--r--sfc/unit_tests/unit/lib/test_odl_utils.py775
1 files changed, 775 insertions, 0 deletions
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..d408a814
--- /dev/null
+++ b/sfc/unit_tests/unit/lib/test_odl_utils.py
@@ -0,0 +1,775 @@
+#!/usr/bin/env python
+
+###############################################################################
+# Copyright (c) 2018 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(expected, 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.actual_rsps_in_compute', 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(self,
+ mock_get_tacker_items,
+ mock_promised_rsps_in_compute,
+ mock_actual_rsps_in_compute,
+ mock_get_odl_items,
+ mock_find_compute,
+ mock_log,
+ mock_sleep):
+ """
+ Checks the proper functionality of wait_for_classification_rules
+ function
+ """
+
+ log_calls = [call("Error when waiting for classification rules: "
+ "RSPs not configured in ODL"),
+ 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("['compute|rsps']"),
+ call("Classification rules were updated")]
+
+ mock_promised_rsps_in_compute.return_value = None
+
+ # rsps not configured in odl
+ 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')
+ self.assertEqual(mock_promised_rsps_in_compute.call_count, 10)
+ 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[:1])
+
+ 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']
+ # timeout but classification rules are not updated
+ odl_utils.wait_for_classification_rules('ovs_logger',
+ 'compute_nodes',
+ 'odl_ip',
+ 'odl_port',
+ 'compute_name',
+ 'neutron_ports',
+ timeout=2)
+ mock_log.error.assert_has_calls(log_calls[1:2])
+
+ # classification rules were updated
+ mock_actual_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[2:])
+
+ @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_once_with(
+ 'ml2_odl', 'url')
+ mock_search.assert_called_once_with(r'[0-9]+(?:\.[0-9]+){3}\:[0-9]+',
+ 'config')
+
+ 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.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(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 conditions
+ """
+
+ log_calls = [call('RSPs are still active in the MD-SAL'),
+ call('There was an error getting the compute: ErrorMSG'),
+ call('Classification flows still in the compute')]
+
+ mock_compute = Mock()
+ mock_compute.ssh_client = 'mock_ssh_client'
+
+ mock_actual_rsps.side_effect = [True, True]
+
+ mock_active_rsps_on_ports.side_effect = [True,
+ True, False,
+ True, False]
+
+ mock_find_compute.side_effect = [Exception('ErrorMSG'),
+ mock_compute]
+
+ # RSPs are still active in the MD-SAL
+ 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_log.debug.assert_has_calls(log_calls[0:1])
+ mock_sleep.assert_called_once_with(3)
+
+ # error getting the compute
+ 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_log.debug.assert_has_calls(log_calls[1:2])
+ mock_find_compute.assert_called_once_with('compute_client_name',
+ 'compute_nodes')
+ assert mock_sleep.call_count == 2
+
+ # classification flows still in the compute
+ 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_log.debug.assert_has_calls(log_calls[2:3])
+ mock_actual_rsps.assert_called_with('ovs_logger', 'mock_ssh_client')
+ assert mock_sleep.call_count == 5
+
+ @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()
+
+ @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')