Fix bug/2115776 where Neutron deletes A/AAAA records

Enable the deletion of all the Designate recordssets for the same
dns_name and dns_domain associated with the specified IP
addresses.

This change enables in the Designate driver the separate deletion
of two recordsets associated to the same dns-name and dns-domain
that were created by a sequence of one port create/update and one
floating ip create/update. Before this change, the assumption when
deleting recordsets associated to the same same dns-name and
dns-domain was that they were created in a single operation.
Consequently, the change handles each deletion operation of the
dns_name in the dns_domain based on the IP addresses specified
in the input argument.

Existing unit tests in test_dns_integration.py and
test_dns_domain_keywords.py failed with this change.  These existing
test cases are updated in this commit such that the IP addresses
in the mock recordsets are strings, instead of netaddr.IPAddress()
objects, to match recordset['records'] data type in
_get_ids_ips_to_delete().  The unit tests passed with mismatched
types in the past because the check between recordset and input
argument records compared the size of two sets rather than the
values of the data.

A new unit test test_delete_single_record_from_two_records() is
added to test that the designate driver handles the use case where
only a subset of A or AAAA records are deleted.
test_delete_single_record_from_two_records() was generated with the
assistance of Claude Code, copying from test_delete_record_set()
with a small modification to delete only one of the records.

Assisted-by: Claude Code
Depends-on: https://review.opendev.org/c/openstack/neutron-tempest-plugin/+/960659
Closes-Bug: #2115776
Change-Id: I42f1d504a063f1d8542c861b3b7caffe56c47bf1
Signed-off-by: Helen Chen <ichen@redhat.com>

neutron/services/externaldns/drivers/designate/driver.py

@@ -165,8 +165,6 @@ class Designate(driver.ExternalDNSService):
                 dns_domain, criterion={"name": "%s" % name})
         except (d_exc.NotFound, d_exc.Forbidden):
             raise dns_exc.DNSDomainNotFound(dns_domain=dns_domain)
-        ids = [rec['id'] for rec in recordsets]
+        return [rec['id'] for rec in recordsets
-        ips = [str(ip) for rec in recordsets for ip in rec['records']]
+                for ip in rec['records']
-        if set(ips) != set(records):
+                if ip in records]
-            raise dns_exc.DuplicateRecordSet(dns_name=name)
-        return ids

neutron/tests/unit/plugins/ml2/extensions/test_dns_domain_keywords.py

@@ -84,10 +84,10 @@ class DNSDomainKeywordsTestCase(
                               new_dns_name=test_dns_integration.NEWDNSNAME,
                               new_dns_domain=None, **kwargs):
         test_dns_integration.mock_client.reset_mock()
-        ip_addresses = [netaddr.IPAddress(ip['ip_address'])
+        records_v4 = [ip['ip_address'] for ip in port['fixed_ips']
-                        for ip in port['fixed_ips']]
+                      if netaddr.IPAddress(ip['ip_address']).version == 4]
-        records_v4 = [ip for ip in ip_addresses if ip.version == 4]
+        records_v6 = [ip['ip_address'] for ip in port['fixed_ips']
-        records_v6 = [ip for ip in ip_addresses if ip.version == 6]
+                      if netaddr.IPAddress(ip['ip_address']).version == 6]
         recordsets = []
         if records_v4:
             recordsets.append({'id': test_dns_integration.V4UUID,

neutron/tests/unit/plugins/ml2/extensions/test_dns_integration.py

@@ -126,10 +126,10 @@ class DNSIntegrationTestCase(test_plugin.Ml2PluginV2TestCase):
     def _update_port_for_test(self, port, new_dns_name=NEWDNSNAME,
                               new_dns_domain=None, **kwargs):
         mock_client.reset_mock()
-        ip_addresses = [netaddr.IPAddress(ip['ip_address'])
+        records_v4 = [ip['ip_address'] for ip in port['fixed_ips']
-                        for ip in port['fixed_ips']]
+                      if netaddr.IPAddress(ip['ip_address']).version == 4]
-        records_v4 = [ip for ip in ip_addresses if ip.version == 4]
+        records_v6 = [ip['ip_address'] for ip in port['fixed_ips']
-        records_v6 = [ip for ip in ip_addresses if ip.version == 6]
+                      if netaddr.IPAddress(ip['ip_address']).version == 6]
         recordsets = []
         if records_v4:
             recordsets.append({'id': V4UUID, 'records': records_v4})

neutron/tests/unit/services/externaldns/drivers/designate/test_driver.py

@@ -169,6 +169,31 @@ class TestDesignateDriver(base.BaseTestCase):
         )
         self.admin_client.recordsets.delete.assert_not_called()
 
+    def test_delete_single_record_from_two_records(self):
+        # Set up two records similar to test_delete_record_set
+        self.client.recordsets.list.return_value = [
+            {'id': 123, 'records': ['192.168.0.10']},
+            {'id': 456, 'records': ['2001:db8:0:1::1']}
+        ]
+
+        cfg.CONF.set_override(
+            'allow_reverse_dns_lookup', False, group='designate'
+        )
+
+        # Delete only the first record (IPv4) out of the two
+        self.driver.delete_record_set(
+            self.context, 'example.test.', 'test',
+            ['192.168.0.10']
+        )
+
+        # Verify that only the IPv4 record was deleted
+        self.client.recordsets.delete.assert_called_once_with(
+            'example.test.', 123
+        )
+
+        # Admin client should not be called since reverse DNS is disabled
+        self.admin_client.recordsets.delete.assert_not_called()
+
     def test_delete_record_set_with_reverse_dns(self):
         self.client.recordsets.list.return_value = [
             {'id': 123, 'records': ['192.168.0.10']},