fixing dns server delete record in all roles

requirements.txt

@@ -1,4 +1,4 @@
 jc
 pytest
 python-ldap
-pytest-mh >= 1.0.29
+pytest-mh

sssd_test_framework/roles/ad.py

@@ -18,7 +18,6 @@
     attrs_include_value,
     attrs_parse,
     attrs_to_hash,
-    ip_to_ptr,
     ip_version,
     seconds_to_timespan,
 )
@@ -2416,10 +2415,11 @@ def list_zones(self) -> list[str]:
         :return: List of zones.
         :rtype: list[str]
         """
-        result = self.host.conn.run("Get-DnsServerZone | Format-List -Property ZoneName").stdout_lines
-        result = [x for x in result if x not in ["\r", "", None]]
-        result = [y.replace("\r", "").strip() for y in result]
-        result = [z.split(":")[1].strip() for z in result]
+        result = self.host.conn.run(
+            "Get-DnsServerZone | Format-List -Property ZoneName", unified_newlines=True
+        ).stdout_lines
+        result = [x for x in result if x not in ["", None]]
+        result = [y.split(":")[1].strip() for y in result]
 
         return result
 
@@ -2492,41 +2492,57 @@ def add_record(self, name: str, data: str | int) -> ADDNSZone:
 
     def delete_record(self, name: str) -> None:
         """
-        Delete DNS record, both forward and reverse records are deleted.
+        Delete all records for provided dns entry.
+
+        The complexity is because the method takes one parameter. We need to discover the correct
+        value, cmdlet, protocol, and zone to properly delete the entry.
+
+        If dig returns a result, iterate through the zones, if the zone is in the name, it's a forward record.
+        The return may be from the cache, search the zone for the record, if found, delete it. If "in-addr" is
+        in the zone name, it's a reverse record that follows the same steps as a forward record.
 
         :param name: Name or IP of the record.
         :type name: str
         """
         if self.domain not in name:
             name = f"{name}.{self.domain}"
+        short_name = name.split(".")[0]
 
-        records = self.host.conn.run(f"dig +short +norecurse {name} '@{self.server}'").stdout_lines
-        records = [s.rstrip("\r") for s in records]
-
-        if not isinstance(records, list) or records is None:
-            return None
-
-        if len(records) > 1:
-            for record in records:
-                if ip_version(record) == 4:
-                    self.role.host.conn.run(
-                        f"Remove-DnsServerResourceRecord -RRType A -Force -ZoneName {self.zone_name} -Name {name}"
-                    )
-                if ip_version(record) == 6:
-                    self.host.conn.run(
-                        f"Remove-DnsServerResourceRecord -RRType AAAA -Force -ZoneName {self.zone_name} -Name {name}"
-                    )
-
-        for ptr_records in records:
-            ptr_record = self.host.conn.run(f"dig +short -x +norecurse {ptr_records} '@{self.server}'").stdout_lines
-            ptr_record = [r.rstrip("\r") for r in ptr_record]
-            if ptr_record:
-                self.host.conn.run(
-                    f"Remove-DnsServerResourceRecord -RRType PTR "
-                    f"-Force -ZoneName {ip_to_ptr(ptr_record[0])} "
-                    f"-Name {'.'.join(ptr_record).split()[-1]}",
-                )
-        return None
+        records = self.host.conn.run(
+            f"dig +short +norecurse {name} '@{self.server}'", unified_newlines=True
+        ).stdout_lines
+        zones = self.list_zones()
+
+        if records is not None:
+            for zone in zones:
+                if zone in name:
+                    a_records = self.host.conn.run(
+                        f'Get-DnsServerResourceRecord -ZoneName "{zone}" | Where-Object '
+                        f'{{ $_.HostName -eq "{short_name}" }}',
+                        unified_newlines=True,
+                    ).stdout_lines[3:]
+                    if a_records is not None:
+                        for j in a_records:
+                            if ip_version(j.split()[-1].strip()) == 4:
+                                self.host.conn.run(
+                                    f"Remove-DnsServerResourceRecord -ZoneName {zone} "
+                                    f"-Name {short_name} -RRType A -Force"
+                                )
+                            elif ip_version(j.split()[-1].strip()) == 6:
+                                self.host.conn.run(
+                                    f"Remove-DnsServerResourceRecord -ZoneName {zone} "
+                                    f"-Name {short_name} -RRType AAAA -Force"
+                                )
+                if "in-addr" in zone:
+                    ptr_record = self.host.conn.run(
+                        f'Get-DnsServerResourceRecord -ZoneName "{zone}" | Where-Object '
+                        f'{{ $_.RecordData.PtrDomainName -eq "{name}." }}'
+                    ).stdout_lines
+                    if len(ptr_record) == 1:
+                        ptr_name = ptr_record[0].split()[0].strip()
+                        self.host.conn.run(
+                            f"Remove-DnsServerResourceRecord -ZoneName {zone} " f"-Name {ptr_name}. -RRType PTR -Force"
+                        )
 
     def print(self) -> str:
         """

sssd_test_framework/roles/ipa.py

@@ -3089,7 +3089,10 @@ def add_record(self, name: str, data: str | int) -> IPADNSZone:
 
     def delete_record(self, name: str) -> None:
         """
-        Delete DNS record, both forward and reverse records are deleted.
+        Delete all A/AAAA/PTR/NSPTR records for provided record.
+
+        The complexity is because the method takes one parameter and not specifying the record type will
+        remove the entire entry. Instead the record type and data is specified to preserve sshfp records.
 
         Implements :meth:`GenericDNSZone.delete_record`.
 
@@ -3098,28 +3101,58 @@ def delete_record(self, name: str) -> None:
         """
         if self.domain not in name:
             name = f"{name}.{self.domain}"
-
-        records = self.host.conn.run(f"dig +short +norecurse {name} '@{self.server}'").stdout_lines
-        records = [s.rstrip("\r") for s in records]
-
-        if not isinstance(records, list) or records is None:
-            return None
-
-        if len(records) > 1:
-            for record in records:
-                if ip_version(record) == 4:
-                    self.host.conn.run(f"ipa dnsrecord-del {self.zone_name} --a-rec={record}")
-                if ip_version(record) == 6:
-                    self.host.conn.run(f"ipa dnsrecord-del {self.zone_name} --aaaa-rec={record}")
-
-        for ptr_records in records:
-            ptr_record = self.host.conn.run(f"dig +short -x +norecurse {ptr_records} '@{self.server}'").stdout_lines
-            ptr_record = [r.rstrip("\r") for r in ptr_record]
-            if ptr_record:
-                self.host.conn.run(f"ipa dnsrecord-del {self.zone_name} --ptr-rec={record}")
-        return None
-
-        time.sleep(5)  # Wait for the record to be deleted
+        short_name = name.split(".")[0].strip()
+        zones = self.list_zones()
+
+        for zone in zones:
+            zone_records = self.host.conn.run(f"ipa dnsrecord-find {zone}").stdout_lines
+            in_section: bool = False
+
+            for i, line in enumerate(zone_records):
+                _line = line.strip()
+
+                if not _line:
+                    if in_section:
+                        break
+                    continue
+
+                if _line.startswith("Record name:"):
+                    current = _line.split(":", 1)[1].strip()
+                    in_section = current == short_name or current == f"{name}."
+                    continue
+
+                if in_section:
+                    if _line.startswith("AAAA record:"):
+                        ip = _line.split(":", 1)[1].strip()
+                        _ip = [x.strip() for x in ip.split(",")] if "," in ip else [ip.strip()]
+                        for y in _ip:
+                            self.host.conn.run(
+                                f"ipa dnsrecord-del {zone} {short_name} " f"--aaaa-rec={y}",
+                            )
+
+                    elif _line.startswith("A record:"):
+                        ip = _line.split(":", 1)[1].strip()
+                        _ip = [x.strip() for x in ip.split(",")] if "," in ip else [ip.strip()]
+                        for y in _ip:
+                            self.host.conn.run(
+                                f"ipa dnsrecord-del {zone} {short_name} " f"--a-rec={y}",
+                            )
+                    elif _line.startswith("PTR record:"):
+                        if i > 0:
+                            prev = zone_records[i - 1].strip()
+                            if prev.startswith("Record name:"):
+                                ptr = prev.split(":", 1)[1].strip()
+                                self.host.conn.run(
+                                    f"ipa dnsrecord-del {self.zone_name} {zone} --ptr-rec={ptr} {name}."
+                                )
+                    elif _line.startswith("NSPTR record:"):
+                        if i > 0:
+                            prev = zone_records[i - 1].strip()
+                            if prev.startswith("Record name:"):
+                                ptr = prev.split(":", 1)[1].strip()
+                                self.host.conn.run(
+                                    f"ipa dnsrecord-del {self.zone_name} {zone} --naptr-rec={ptr} {name}."
+                                )
 
     def print(self) -> str:
         """

sssd_test_framework/roles/samba.py

@@ -12,7 +12,7 @@
 from pytest_mh.conn import ProcessResult
 
 from ..hosts.samba import SambaHost
-from ..misc import attrs_parse, ip_to_ptr, ip_version, to_list_of_strings
+from ..misc import attrs_parse, ip_version, to_list_of_strings
 from ..utils.ldap import LDAPRecordAttributes
 from .base import BaseLinuxLDAPRole, BaseObject, DeleteAttribute
 from .generic import (
@@ -1635,40 +1635,66 @@ def delete_record(self, name: str) -> None:
         """
         Delete DNS record, both forward and reverse records are deleted.
 
-        Implements :meth:`GenericDNSZone.delete_record`.
+        The complexity is because the method takes one parameter. We need to discover the correct
+        values and type to properly delete the entry.
+
+        The samba-tool output returns multiple lines per record. Two examples,
+
+          Name=client, Records=1, Children=0
+          A: 172.16.100.40 (flags=f0, serial=110, ttl=3600)
+
+          Name=50, Records=1, Children=0
+          PTR: client.samba.test (flags=f0, serial=2, ttl=900)
+
+        Implements: meth:`GenericDNSZone.delete_record`.
 
         :param name: Name of the record.
         :type name: str
         """
         if self.domain not in name:
             name = f"{name}.{self.domain}"
+        short_name = name.split(".")[0]
 
         records = self.host.conn.run(f"dig +short +norecurse {name} '@{self.server}'").stdout_lines
-        records = [s.rstrip("\r") for s in records]
-
-        if not isinstance(records, list) or records is None:
-            return None
-
-        if len(records) > 1:
-            for record in records:
-                if ip_version(record) == 4:
-                    self.role.host.conn.run(
-                        f"samba-tool dns delete {self.server} {self.zone_name} {name} A {record} {self.credentials}"
-                    )
-                if ip_version(record) == 6:
-                    self.host.conn.run(
-                        f"samba-tool dns delete {self.server} {self.zone_name} {name} AAAA {record} {self.credentials}"
-                    )
-
-        for ptr_records in records:
-            ptr_record = self.host.conn.run(f"dig +short -x +norecurse {ptr_records} '@{self.server}'").stdout_lines
-            ptr_record = [r.rstrip("\r") for r in ptr_record]
-            if ptr_record:
-                self.host.conn.run(
-                    f"samba-tool dns delete {self.server} {ip_to_ptr(ptr_record[0])} {name} "
-                    f"PTR {ptr_record} {self.credentials}"
-                )
-        return None
+        zones = self.list_zones()
+
+        if records is not None:
+            for zone in zones:
+                if zone in name:
+                    a_record = self.host.conn.run(
+                        f"samba-tool dns query {self.server} {zone} @ A {self.credentials}",
+                    ).stdout_lines
+                    for i in a_record:
+                        if f"Name={short_name}," in i:
+                            ip = a_record[a_record.index(i) + 1].strip().split()[1].strip()
+                            if ip_version(ip) == 4:
+                                self.host.conn.run(
+                                    f"samba-tool dns delete {self.server} {zone} "
+                                    f"{short_name} A {ip} {self.credentials}",
+                                )
+                            elif ip_version(ip) == 6:
+                                self.host.conn.run(
+                                    f"samba-tool dns delete {self.server} {zone} "
+                                    f"{short_name} AAAA {ip} {self.credentials}",
+                                )
+                if "in-addr" in zone:
+                    ptr_record = self.host.conn.run(
+                        f"samba-tool dns query {self.server} {zone} @ PTR {self.credentials}",
+                    ).stdout_lines
+                    for j in ptr_record:
+                        if name in j:
+                            _name = (
+                                ptr_record[ptr_record.index(j) - 1]
+                                .strip()
+                                .split()[0]
+                                .strip()
+                                .split("=")[-1]
+                                .rstrip(",")
+                            )
+                            self.host.conn.run(
+                                f"samba-tool dns delete {self.server} {zone} "
+                                f"{_name} PTR {name} {self.credentials}",
+                            )
 
     def print(self) -> str:
         """

sssd_test_framework/topology_controllers.py

@@ -233,6 +233,8 @@ class ADTopologyController(ProvisionedBackupTopologyController):
     def topology_setup(self, client: ClientHost, provider: ADHost | SambaHost) -> None:
         short_hostname = client.conn.run("hostname").stdout.split(".")[0].strip()
         hostname = f"{short_hostname}.{provider.domain}"
+        client.fs.backup("/etc/hostname")
+        client.conn.run(f"echo {hostname} > /etc/hostname")
 
         # Change client hostname to match the domain
         self.logger.info(f"Changing hostname to {hostname}")
@@ -272,6 +274,8 @@ class IPATrustADTopologyController(ProvisionedBackupTopologyController):
     def topology_setup(self, client: ClientHost, ipa: IPAHost, trusted: ADHost | SambaHost) -> None:
         short_hostname = client.conn.run("hostname").stdout.split(".")[0].strip()
         hostname = f"{short_hostname}.{ipa.domain}"
+        client.fs.backup("/etc/hostname")
+        client.conn.run(f"echo {hostname} > /etc/hostname")
 
         # Change client hostname to match the domain
         self.logger.info(f"Changing hostname to {hostname}")