From 59ab482f4b339b63103f505dfc48d464112e797c Mon Sep 17 00:00:00 2001 From: Nisar Khan Date: Mon, 10 Aug 2026 04:42:42 +0000 Subject: [PATCH] Add distributed SNAT unit coverage for RPC, agent, and SNAT manager Expand distributed SNAT test coverage across callback RPCs, agent update handling, and SNAT manager behavior. --- opflexagent/test/test_gbp_ovs_agent.py | 22 +++++++++ opflexagent/test/test_rpc.py | 48 +++++++++++++++++++ opflexagent/test/test_snat_mgr.py | 66 +++++++++++++++++++++++++- 3 files changed, 135 insertions(+), 1 deletion(-) diff --git a/opflexagent/test/test_gbp_ovs_agent.py b/opflexagent/test/test_gbp_ovs_agent.py index 11629d6f..ed57bb4c 100644 --- a/opflexagent/test/test_gbp_ovs_agent.py +++ b/opflexagent/test/test_gbp_ovs_agent.py @@ -266,6 +266,28 @@ def test_process_snat_update_deletes_missing_snat_detail(self): assert_called_once_with({'snat_uuid': snat_uuid}, {}, keep=False) + def test_process_snat_update_syncs_present_snat_details(self): + snat_uuid_keep = '00000000-0000-0000-0000-ffff980a0111' + snat_uuid_delete = '00000000-0000-0000-0000-ffff980a0222' + details = [ + {'snat_uuid': snat_uuid_keep, + 'service_mac': 'aa:bb:cc:dd:ee:ff'}, + {'snat_uuid': snat_uuid_delete} + ] + self.agent.of_rpc.get_snat_details_list = mock.Mock( + return_value=details) + self.agent.ep_manager.dist_snat_manager.sync_host_snat_ip = ( + mock.Mock()) + + self.agent.process_snat_update( + set([snat_uuid_keep, snat_uuid_delete])) + + (self.agent.ep_manager.dist_snat_manager.sync_host_snat_ip. + assert_has_calls([ + mock.call(details[0], {}, keep=True), + mock.call(details[1], {}, keep=False) + ], any_order=True)) + def test_subnet_has_updates(self): fake_sub = {'tenant_id': 'tenant-id', 'id': 'someid'} polling_manager = mock.Mock() diff --git a/opflexagent/test/test_rpc.py b/opflexagent/test/test_rpc.py index 7a067554..5de6457d 100644 --- a/opflexagent/test/test_rpc.py +++ b/opflexagent/test/test_rpc.py @@ -97,3 +97,51 @@ def test_request_vrf_details_list(self): mock.ANY, host='h1', requests=list(range(3))) self.assertFalse( self.callback.agent_notifier.opflex_vrf_update.called) + + def test_request_snat_details(self): + result = {'snat_uuid': 'snat-1'} + self.callback.gbp_driver.request_snat_details = mock.Mock( + return_value=result) + self.callback.request_snat_details(mock.ANY, host='h1') + (self.callback.agent_notifier.opflex_snat_update. + assert_called_once_with(mock.ANY, [result], host='h1')) + + # Test None return + self.callback.agent_notifier.opflex_snat_update.reset_mock() + result = None + self.callback.gbp_driver.request_snat_details = mock.Mock( + return_value=result) + self.callback.request_snat_details(mock.ANY, host='h1') + self.assertFalse( + self.callback.agent_notifier.opflex_snat_update.called) + + def test_request_snat_details_list(self): + result = {'snat_uuid': 'snat-1'} + self.callback.gbp_driver.request_snat_details = mock.Mock( + return_value=result) + self.callback.request_snat_details_list( + mock.ANY, host='h1', requests=list(range(3))) + (self.callback.agent_notifier.opflex_snat_update. + assert_called_once_with(mock.ANY, [result] * 3, host='h1')) + + # Test None return + self.callback.agent_notifier.opflex_snat_update.reset_mock() + result = None + self.callback.gbp_driver.request_snat_details = mock.Mock( + return_value=result) + self.callback.request_snat_details_list( + mock.ANY, host='h1', requests=list(range(3))) + self.assertFalse( + self.callback.agent_notifier.opflex_snat_update.called) + + def test_get_snat_details_list(self): + self.callback.gbp_driver.get_snat_details = mock.Mock( + side_effect=lambda _ctx, snat_id=None, **_kwargs: { + 'snat_uuid': snat_id}) + result = self.callback.get_snat_details_list( + mock.ANY, host='h1', snat_ids=['s1', 's2']) + + self.assertEqual([ + {'snat_uuid': 's1'}, + {'snat_uuid': 's2'} + ], result) diff --git a/opflexagent/test/test_snat_mgr.py b/opflexagent/test/test_snat_mgr.py index ee0b3c9a..4ced172e 100644 --- a/opflexagent/test/test_snat_mgr.py +++ b/opflexagent/test/test_snat_mgr.py @@ -13,6 +13,8 @@ # License for the specific language governing permissions and limitations # under the License. +from unittest import mock + from neutron.tests import base from opflexagent import snat_iptables_manager @@ -25,8 +27,70 @@ class TestSnatManager(base.BaseTestCase): def setUp(self): super(TestSnatManager, self).setUp() - self.mgr = snat_iptables_manager.SnatIptablesManager(None) + self.bridge_manager = mock.Mock() + self.mgr = snat_iptables_manager.SnatIptablesManager( + self.bridge_manager) def test_hash_for_es(self): hash = self.mgr._get_hash_for_es(TEST_HASH_STRING) self.assertEqual(hash, HASH_RESULT) + + def test_setup_snat_for_es_skips_without_any_ip(self): + with mock.patch.object(self.mgr, '_cleanup') as cleanup, \ + mock.patch.object(self.mgr, '_add_port_and_netns') as add_ns, \ + mock.patch.object(self.mgr, '_setup_routes') as setup_routes, \ + mock.patch.object(self.mgr, '_setup_iptables') as setup_ipt: + result = self.mgr.setup_snat_for_es('EXT-1', next_hop_mac='aa:bb') + + self.assertEqual((None, 'aa:bb'), result) + cleanup.assert_not_called() + add_ns.assert_not_called() + setup_routes.assert_not_called() + setup_ipt.assert_not_called() + + def test_setup_snat_for_es_ipv4_programs_namespace_routes_iptables(self): + fake_if_dev = mock.Mock() + fake_if_dev.link.address = 'de:ad:be:ef:00:01' + + with mock.patch.object(self.mgr, '_get_hash_for_es', + return_value='of-hash-if'), \ + mock.patch.object(self.mgr, '_cleanup') as cleanup, \ + mock.patch.object(self.mgr, '_add_port_and_netns', + return_value=fake_if_dev) as add_ns, \ + mock.patch.object(self.mgr, '_setup_routes') as setup_routes, \ + mock.patch.object(self.mgr, '_setup_iptables') as setup_ipt: + result = self.mgr.setup_snat_for_es( + 'EXT-1', ip_start='10.0.0.2', ip_gw='10.0.0.1/24', mtu=1400) + + self.assertEqual(('of-hash-if', 'de:ad:be:ef:00:01'), result) + cleanup.assert_called_once_with('of-hash-if', 'of-hash-if') + add_ns.assert_called_once_with( + 'of-hash-if', 'of-hash-if', if_mac=None, mtu=1400) + setup_routes.assert_called_once_with( + fake_if_dev, 4, '10.0.0.2', '10.0.0.2', '10.0.0.1/24') + setup_ipt.assert_called_once_with( + 'of-hash-if', 'of-hash-if', '10.0.0.2', '10.0.0.2', None, None) + + def test_cleanup_snat_all_respects_exclusions(self): + self.bridge_manager.get_port_name_list.return_value = [ + 'of-a', 'of-b', 'other-port'] + with mock.patch.object(self.mgr, '_get_hash_for_es', + return_value='of-b'), \ + mock.patch.object(self.mgr, '_cleanup') as cleanup: + self.mgr.cleanup_snat_all(exclude_es=['EXT-2']) + + cleanup.assert_called_once_with('of-a', 'of-a') + + def test_check_if_exists_uses_hashed_namespace_name(self): + ipw = mock.Mock() + ipw.netns.exists.return_value = True + with mock.patch.object(self.mgr, '_get_hash_for_es', + return_value='of-check'), \ + mock.patch( + 'opflexagent.snat_iptables_manager.' + 'ip_lib.IPWrapper', + return_value=ipw): + exists = self.mgr.check_if_exists('EXT-1') + + self.assertTrue(exists) + ipw.netns.exists.assert_called_once_with('of-check')