Updated mock resolution to also include HostAddressManager DNS-related methods

mariadb-AlexanderPresniakov · mariadb-AlexanderPresniakov · commit 50efa91a0f21 · 2025-11-10T20:13:46.000Z
diff --git a/cmapi/cmapi_server/managers/host_identity.py b/cmapi/cmapi_server/managers/host_identity.py
@@ -250,42 +250,85 @@ def get_local_identity(self) -> HostIdentity:
 
     def _resolve_dns(self, hostname: str) -> tuple[list[IPAddress], list[str]]:
         """Resolve the given hostname using DNS and return (addresses, names)."""
-        resolver = dns.resolver.Resolver(configure=True)
-
-        # IPv4
-        a_records: list[IPAddress] = []
+        ipv4_texts: list[str] = []
+        ipv6_texts: list[str] = []
         try:
-            for a_rdata in resolver.resolve(hostname, 'A', raise_on_no_answer=False):
-                a_records.append(ipaddress.ip_address(a_rdata.to_text()))
+            ipv4_texts = self._dns_resolve_ipv4(hostname)
+        except dns.resolver.NoAnswer:
+            logger.debug('IPv4 lookup returned no records for %s', hostname)
+            ipv4_texts = []
         except Exception:
-            logger.exception('A lookup failed for %s', hostname)
-
-        # IPv6
-        aaaa_records: list[IPAddress] = []
+            logger.exception('IPv4 lookup unexpected failure for %s', hostname)
+            raise
         if self._policy.allow_ipv6:
             try:
-                for aaaa_rdata in resolver.resolve(hostname, 'AAAA', raise_on_no_answer=False):
-                    aaaa_records.append(ipaddress.ip_address(aaaa_rdata.to_text()))
+                ipv6_texts = self._dns_resolve_ipv6(hostname)
+            except dns.resolver.NoAnswer:
+                logger.debug('IPv6 lookup returned no records for %s', hostname)
+                ipv6_texts = []
             except Exception:
-                logger.exception('AAAA lookup failed for %s', hostname)
+                logger.exception('IPv6 lookup unexpected failure for %s', hostname)
+                raise
+
+        addrs: list[IPAddress] = []
+        for t in ipv4_texts:
+            try:
+                addrs.append(ipaddress.ip_address(t))
+            except ValueError:
+                continue
+        for t in ipv6_texts:
+            try:
+                addrs.append(ipaddress.ip_address(t))
+            except ValueError:
+                continue
 
         names = [hostname]
-        return a_records + aaaa_records, names
+        return addrs, names
 
     def _reverse_dns_names(self, ip: IPAddress) -> list[str]:
         """Fetch PTR names for an IP via DNS."""
         try:
-            reverse_name = dns.reversename.from_address(str(ip))
-            answer = dns.resolver.resolve(reverse_name, 'PTR', raise_on_no_answer=False)
-            names: list[str] = []
-            for ptr_rdata in answer:
+            return self._dns_reverse(str(ip))
+        except dns.resolver.NoAnswer:
+            logger.debug('PTR lookup returned no records for %s', ip)
+            return []
+        except Exception:
+            logger.exception('PTR lookup unexpected failure for %s', ip)
+            raise
+
+    # DNS abstraction methods for easier mocking
+    def _dns_resolve_ipv4(self, hostname: str) -> list[str]:
+        resolver = dns.resolver.Resolver(configure=True)
+        results: list[str] = []
+        for rdata in resolver.resolve(hostname, 'A', raise_on_no_answer=False):
+            try:
+                results.append(rdata.to_text())
+            except Exception:
+                continue
+        return results
+
+    def _dns_resolve_ipv6(self, hostname: str) -> list[str]:
+        resolver = dns.resolver.Resolver(configure=True)
+        results: list[str] = []
+        for rdata in resolver.resolve(hostname, 'AAAA', raise_on_no_answer=False):
+            try:
+                results.append(rdata.to_text())
+            except Exception:
+                continue
+        return results
+
+    def _dns_reverse(self, ip_text: str) -> list[str]:
+        reverse_name = dns.reversename.from_address(ip_text)
+        answer = dns.resolver.resolve(reverse_name, 'PTR', raise_on_no_answer=False)
+        names: list[str] = []
+        for ptr_rdata in answer:
+            try:
                 fqdn_name = str(ptr_rdata.target).rstrip('.').lower()
                 if _is_fqdn(fqdn_name):
                     names.append(fqdn_name)
-            return names
-        except Exception:
-            logger.exception('PTR lookup failed for %s', ip)
-            return []
+            except Exception:
+                continue
+        return names
 
     def _contains_private(self, addrs: list[str]) -> bool:
         """Return True if any resolvable address string is a private IP."""
diff --git a/cmapi/cmapi_server/test/mock_resolution.py b/cmapi/cmapi_server/test/mock_resolution.py
@@ -73,6 +73,79 @@ def set_default(self, ip: str, hostname: str):
         self._default_hostname = hostname
         return self
 
+    def build(self):
+
+        @contextmanager
+        def _ctx():
+            patches = [
+                # Patch socket-level resolvers (NetworkManager uses these under the hood)
+                patch('socket.getaddrinfo', side_effect=self._fake_getaddrinfo),
+                patch('socket.gethostbyname', side_effect=self._fake_gethostbyname),
+                patch('socket.gethostbyaddr', side_effect=self._fake_gethostbyaddr),
+                # Patch local identity to be synthetic; avoid real system calls
+                patch('socket.gethostname', return_value=CUR_HOST_HOSTNAME),
+                patch('socket.getfqdn', return_value=CUR_HOST_HOSTNAME),
+                # Patch NetworkManager local IP discovery (it uses psutil or system libs,
+                #   proper mocking would be too complex)
+                patch('cmapi_server.managers.network.NetworkManager.get_current_node_ips', return_value=[CUR_HOST_IP, DEFAULT_LOCALHOST_IP]),
+                # Patch HostAddressManager DNS abstraction methods
+                patch('cmapi_server.managers.host_identity.HostAddressManager._dns_resolve_ipv4',
+                      side_effect=self._fake_dns_resolve_ipv4),
+                patch('cmapi_server.managers.host_identity.HostAddressManager._dns_resolve_ipv6',
+                      side_effect=self._fake_dns_resolve_ipv6),
+                patch('cmapi_server.managers.host_identity.HostAddressManager._dns_reverse',
+                      side_effect=self._fake_dns_reverse),
+            ]
+            with ExitStack() as stack:
+                for p in patches:
+                    stack.enter_context(p)
+                yield
+
+        return _ctx()
+
+    def _fake_getaddrinfo(self, host, port, family=socket.AF_UNSPEC, type=0, proto=0, flags=0):
+        # Only handle AF_INET calls; otherwise, simulate failure
+        if family not in (socket.AF_UNSPEC, socket.AF_INET):
+            raise socket.gaierror
+        # For localhost, return loopback first and include CUR_HOST_IP as secondary
+        if host == DEFAULT_LOCALHOST_HOSTNAME:
+            return [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, '', (DEFAULT_LOCALHOST_IP, port)),
+                (socket.AF_INET, socket.SOCK_STREAM, 6, '', (CUR_HOST_IP, port)),
+            ]
+        ip, _ = self._resolve_forward(host)
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (ip, port))]
+
+    def _fake_gethostbyname(self, name: str) -> str:
+        ip, _ = self._resolve_forward(name)
+        return ip
+
+    def _fake_gethostbyaddr(self, addr: str):
+        # If no reverse record was set, simulate reverse lookup failure
+        if addr not in self._reverse:
+            raise socket.herror
+        primary, aliases = self._reverse[addr]
+        return (primary, aliases, [addr])
+
+    # HostIdentityManager DNS abstraction mocks
+    def _fake_dns_resolve_ipv4(self, hostname: str) -> List[str]:
+        # Return mapped IPv4 for provided hostname, or empty list if unknown
+        ip = self._forward.get(hostname)
+        return [ip] if ip else []
+
+    def _fake_dns_resolve_ipv6(self, hostname: str) -> List[str]:
+        # Keep IPv6 disabled by default in tests for determinism
+        return []
+
+    def _fake_dns_reverse(self, ip_text: str) -> List[str]:
+        # Return PTR names (primary + aliases) from reverse map, lowercase
+        rec = self._reverse.get(ip_text)
+        if not rec:
+            return []
+        primary, aliases = rec
+        names = [primary, *aliases]
+        return [n.rstrip('.').lower() for n in names if n]
+
     def _resolve_forward(self, host: str) -> Tuple[str, str]:
         """Resolve hostname or IP to (ip, hostname) using mappings/defaults."""
         # If input looks like an IP, return it with reverse or default hostname
@@ -101,50 +174,6 @@ def _resolve_forward(self, host: str) -> Tuple[str, str]:
         # As a last resort, echo back (host, host)
         return host, host
 
-    def build(self):
-
-        def _fake_getaddrinfo(host, port, family=socket.AF_UNSPEC, type=0, proto=0, flags=0):
-            # Only handle AF_INET calls; otherwise, simulate failure
-            if family not in (socket.AF_UNSPEC, socket.AF_INET):
-                raise socket.gaierror
-            # For localhost, return loopback first and include CUR_HOST_IP as secondary
-            if host == DEFAULT_LOCALHOST_HOSTNAME:
-                return [
-                    (socket.AF_INET, socket.SOCK_STREAM, 6, '', (DEFAULT_LOCALHOST_IP, port)),
-                    (socket.AF_INET, socket.SOCK_STREAM, 6, '', (CUR_HOST_IP, port)),
-                ]
-            ip, _ = self._resolve_forward(host)
-            return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (ip, port))]
-
-        def _fake_gethostbyname(name: str) -> str:
-            ip, _ = self._resolve_forward(name)
-            return ip
-
-        def _fake_gethostbyaddr(addr: str):
-            # If no reverse record was set, simulate reverse lookup failure
-            if addr not in self._reverse:
-                raise socket.herror
-            primary, aliases = self._reverse[addr]
-            return (primary, aliases, [addr])
-
-        @contextmanager
-        def _ctx():
-            patches = [
-                # Patch socket-level resolvers (NetworkManager uses these under the hood)
-                patch('socket.getaddrinfo', side_effect=_fake_getaddrinfo),
-                patch('socket.gethostbyname', side_effect=_fake_gethostbyname),
-                patch('socket.gethostbyaddr', side_effect=_fake_gethostbyaddr),
-                # Patch local identity to be synthetic; avoid real system calls
-                patch('socket.gethostname', return_value=CUR_HOST_HOSTNAME),
-                patch('socket.getfqdn', return_value=CUR_HOST_HOSTNAME),
-            ]
-            with ExitStack() as stack:
-                for p in patches:
-                    stack.enter_context(p)
-                yield
-
-        return _ctx()
-
 
 def simple_resolution_mock(hostname: str, ip: str):
     """Return a context manager for simple name/IP resolution mocking.
diff --git a/cmapi/cmapi_server/test/test_failover_agent.py b/cmapi/cmapi_server/test/test_failover_agent.py
@@ -1,13 +1,10 @@
 import logging
 
-from mcs_node_control.models.node_config import NodeConfig
-
 from cmapi_server.failover_agent import FailoverAgent
-from cmapi_server.managers.network import NetworkManager
 from cmapi_server.node_manipulation import add_node, remove_node
-from cmapi_server.test.mock_resolution import simple_resolution_mock, make_local_resolution_builder
+from cmapi_server.test.mock_resolution import make_local_resolution_builder
 from cmapi_server.test.unittest_global import BaseNodeManipTestCase, tmp_mcs_config_filename
-
+from mcs_node_control.models.node_config import NodeConfig
 
 logging.basicConfig(level='DEBUG')
 
