openstack/neutron
Code Issues Proposed changes

Merge "Fix bug/2115776 where Neutron deletes A/AAAA records"

Browse Source
This commit is contained in:
Zuul
2025-09-19 18:07:15 +00:00
committed by Gerrit Code Review
parent 5b91772ac0 9320cd392f
commit 10b0973551
4 changed files with 36 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
neutron/services/externaldns/drivers/designate/driver.py
View File

@@ -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]
ips = [str(ip) for rec in recordsets for ip in rec['records']]
if set(ips) != set(records):
raise dns_exc.DuplicateRecordSet(dns_name=name)
return ids
return [rec['id'] for rec in recordsets
for ip in rec['records']
if ip in records]

8
neutron/tests/unit/plugins/ml2/extensions/test_dns_domain_keywords.py
View File

@@ -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'])
for ip in port['fixed_ips']]
records_v4 = [ip for ip in ip_addresses if ip.version == 4]
records_v6 = [ip for ip in ip_addresses if ip.version == 6]
records_v4 = [ip['ip_address'] for ip in port['fixed_ips']
if netaddr.IPAddress(ip['ip_address']).version == 4]
records_v6 = [ip['ip_address'] for ip in port['fixed_ips']
if netaddr.IPAddress(ip['ip_address']).version == 6]
recordsets = []
if records_v4:
recordsets.append({'id': test_dns_integration.V4UUID,

8
neutron/tests/unit/plugins/ml2/extensions/test_dns_integration.py
View File

@@ -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'])
for ip in port['fixed_ips']]
records_v4 = [ip for ip in ip_addresses if ip.version == 4]
records_v6 = [ip for ip in ip_addresses if ip.version == 6]
records_v4 = [ip['ip_address'] for ip in port['fixed_ips']
if netaddr.IPAddress(ip['ip_address']).version == 4]
records_v6 = [ip['ip_address'] for ip in port['fixed_ips']
if netaddr.IPAddress(ip['ip_address']).version == 6]
recordsets = []
if records_v4:
recordsets.append({'id': V4UUID, 'records': records_v4})

25
neutron/tests/unit/services/externaldns/drivers/designate/test_driver.py
View File

@@ -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']},