diff --git a/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py b/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py index 4451ef92552..4d1fe0eed90 100644 --- a/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py +++ b/neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py @@ -1082,14 +1082,15 @@ def _update_dnat_entry_if_needed(self, port_id, up=True): {ovn_const.OVN_FIP_EXT_MAC_KEY: nat['external_mac']})).execute() - if up and ovn_conf.is_ovn_distributed_floating_ip(): - mac = nat['external_ids'][ovn_const.OVN_FIP_EXT_MAC_KEY] - if nat['external_mac'] != mac: - LOG.debug("Setting external_mac of port %s to %s", - port_id, mac) - self.nb_ovn.db_set( - 'NAT', nat['_uuid'], ('external_mac', mac)).execute( - check_error=True) + if ovn_conf.is_ovn_distributed_floating_ip(): + if up: + mac = nat['external_ids'].get(ovn_const.OVN_FIP_EXT_MAC_KEY) + if mac and nat['external_mac'] != mac: + LOG.debug("Setting external_mac of port %s to %s", + port_id, mac) + self.nb_ovn.db_set( + 'NAT', nat['_uuid'], ('external_mac', mac)).execute( + check_error=True) else: if nat['external_mac']: LOG.debug("Clearing up external_mac of port %s", port_id) diff --git a/neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py b/neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py index df3db11fd78..2e086bbb10a 100644 --- a/neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py +++ b/neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py @@ -1175,14 +1175,14 @@ def test_set_port_status_concurrent_delete(self): side_effect=exc) as apc, \ mock.patch.object(self.mech_driver, '_update_dnat_entry_if_needed') as ude: - self.mech_driver.set_port_status_down(port1['port']['id'], False) + self.mech_driver.set_port_status_down(port1['port']['id']) apc.assert_called_once_with( mock.ANY, port1['port']['id'], resources.PORT, provisioning_blocks.L2_AGENT_ENTITY ) - ude.assert_called_once_with(port1['port']['id']) + ude.assert_called_once_with(port1['port']['id'], False) def test_bind_port_unsupported_vnic_type(self): fake_port = fakes.FakePort.create_one_port( @@ -2382,9 +2382,10 @@ def test_agent_with_nb_cfg_timestamp_not_timeout(self): self.assertTrue(agent.alive, "Agent of type %s alive=%s" % ( agent.agent_type, agent.alive)) - def _test__update_dnat_entry_if_needed(self, up=True): - ovn_conf.cfg.CONF.set_override( - 'enable_distributed_floating_ip', True, group='ovn') + def _test__update_dnat_entry_if_needed(self, up=True, dvr=True): + if dvr: + ovn_conf.cfg.CONF.set_override( + 'enable_distributed_floating_ip', True, group='ovn') port_id = 'fake-port-id' fake_ext_mac_key = 'fake-ext-mac-key' fake_nat_uuid = uuidutils.generate_uuid() @@ -2392,23 +2393,38 @@ def _test__update_dnat_entry_if_needed(self, up=True): attrs={'_uuid': fake_nat_uuid, 'external_ids': { ovn_const.OVN_FIP_EXT_MAC_KEY: fake_ext_mac_key}, 'external_mac': 'aa:aa:aa:aa:aa:aa'}) + fake_db_find = mock.Mock() fake_db_find.execute.return_value = [nat_row] self.nb_ovn.db_find.return_value = fake_db_find + self.mech_driver._update_dnat_entry_if_needed(port_id, up=up) - if up: + + if up and dvr: # Assert that we are setting the external_mac in the NAT table self.nb_ovn.db_set.assert_called_once_with( 'NAT', fake_nat_uuid, ('external_mac', fake_ext_mac_key)) else: - # Assert that we are cleaning the external_mac from the NAT table - self.nb_ovn.db_clear.assert_called_once_with( - 'NAT', fake_nat_uuid, 'external_mac') - def test__update_dnat_entry_if_needed_up(self): + if dvr: + self.nb_ovn.db_set.assert_not_called() + else: + # Assert that we are cleaning the external_mac from the NAT + # table + self.nb_ovn.db_clear.assert_called_once_with( + 'NAT', fake_nat_uuid, 'external_mac') + + def test__update_dnat_entry_if_needed_up_dvr(self): self._test__update_dnat_entry_if_needed() - def test__update_dnat_entry_if_needed_down(self): + + def test__update_dnat_entry_if_needed_up_no_dvr(self): + self._test__update_dnat_entry_if_needed(dvr=False) + + def test__update_dnat_entry_if_needed_down_dvr(self): self._test__update_dnat_entry_if_needed(up=False) + def test__update_dnat_entry_if_needed_down_no_dvr(self): + self._test__update_dnat_entry_if_needed(up=False, dvr=False) + @mock.patch('neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.' 'ovn_client.OVNClient._get_router_ports') def _test_update_network_fragmentation(self, new_mtu, expected_opts, grps):