Zed proper fix for lb fips

neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py

@@ -1077,14 +1077,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)

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(
@@ -2376,33 +2376,49 @@ 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()
         nat_row = fakes.FakeOvsdbRow.create_one_ovsdb_row(
             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):