fix: refine interactive dns search results

Author: MNametissa

o2switch_cli/cli/interactive.py

@@ -9,12 +9,15 @@
 from o2switch_cli.cli.interactive_support import (
     build_dns_search_suggestions,
     build_domain_suggestions,
+    build_hostname_suggestions,
     build_subdomain_suggestions,
     filter_domains,
+    filter_hostname_results,
     filter_subdomains,
+    paginate_items,
 )
 from o2switch_cli.cli.ui import TerminalUI
-from o2switch_cli.core.models import DomainDescriptor, SubdomainDescriptor
+from o2switch_cli.core.models import DomainDescriptor, HostnameSearchResult, SubdomainDescriptor
 
 MIN_PAGE_SIZE = 6
 MAX_PAGE_SIZE = 12
@@ -54,6 +57,77 @@ def _page_size(ui: TerminalUI) -> int:
     return max(MIN_PAGE_SIZE, min(MAX_PAGE_SIZE, ui.console.size.height - 15))
 
 
+def _browse_hostname_results(
+    ui: TerminalUI,
+    results: list[HostnameSearchResult],
+    *,
+    page_size: int,
+    empty_message: str,
+) -> None:
+    if not results:
+        ui.print_info(empty_message)
+        return
+
+    all_results = list(results)
+    visible_results = list(results)
+    filter_term = ""
+    page = 1
+
+    while True:
+        if not visible_results:
+            ui.console.clear()
+            ui.print_banner()
+            ui.print_info(f'No results matched the filter "{filter_term}".')
+        else:
+            window = paginate_items(visible_results, page=page, page_size=page_size)
+            ui.console.clear()
+            ui.print_banner()
+            if filter_term:
+                ui.print_info(f'Filtered view: "{filter_term}"')
+            ui.print_hostname_search_results(window.items, window)
+            page = window.page
+
+        choices: list[str] = []
+        total_pages = max(1, (len(visible_results) + page_size - 1) // page_size)
+        if page > 1:
+            choices.append("Previous page")
+        if page < total_pages and visible_results:
+            choices.append("Next page")
+        choices.extend(["First page", "Last page", "Filter results"])
+        if filter_term:
+            choices.append("Reset filters")
+        choices.append("Close results")
+
+        action = questionary.select(
+            f"Browse results ({page}/{total_pages})",
+            choices=choices,
+        ).ask()
+        if action == "Previous page":
+            page -= 1
+        elif action == "Next page":
+            page += 1
+        elif action == "First page":
+            page = 1
+        elif action == "Last page":
+            page = total_pages
+        elif action == "Filter results":
+            next_filter = ui.prompt_realtime_search(
+                "Filter current results",
+                suggestions=build_hostname_suggestions(all_results),
+                help_text="Type hosted, dns, hostname, record type, IP, zone, or docroot to narrow the result set",
+            )
+            if next_filter.strip():
+                filter_term = next_filter.strip()
+                visible_results = filter_hostname_results(all_results, filter_term)
+                page = 1
+        elif action == "Reset filters":
+            filter_term = ""
+            visible_results = list(all_results)
+            page = 1
+        else:
+            return
+
+
 def run_interactive_menu(app_context: AppContext) -> None:
     ui = TerminalUI(app_context.console, app_context.output_format)
     cache = InteractiveDataCache()
@@ -115,11 +189,11 @@ def action(active_context: AppContext, selected_choice: str = choice) -> None:
                     return
                 with ui.status("Searching hosted subdomains and DNS zones", spinner="dots12"):
                     matches = active_context.runtime().dns.search(term)
-                ui.browse_pages(
+                _browse_hostname_results(
+                    ui,
                     matches,
                     page_size=_page_size(ui),
                     empty_message="No DNS or hosted results matched the search term.",
-                    render_page=lambda page_items, window: ui.print_hostname_search_results(page_items, window),
                 )
             elif selected_choice == "DNS: upsert A record":
                 host = questionary.text("Hostname or label").ask() or ""

o2switch_cli/core/dns_service.py

@@ -256,6 +256,36 @@ def _hosted_subdomains(self, term: str) -> list[SubdomainDescriptor]:
             )
         return sorted(descriptors, key=lambda item: item.fqdn)
 
+    @staticmethod
+    def _search_rank(item: HostnameSearchResult, needle: str) -> tuple[int, ... | str]:
+        hostname = item.hostname.lower()
+        value = (item.value or "").lower()
+        zone = (item.zone or "").lower()
+        record_type = (item.record_type or "").upper()
+        exact_hostname = 0 if hostname == needle else 1
+        hosted_penalty = 0 if item.category is SearchCategory.HOSTED_SUBDOMAINS else 1
+        exact_zone = 0 if zone == needle else 1
+        exact_value = 0 if value == needle else 1
+        startswith_penalty = 0 if needle and hostname.startswith(needle) else 1
+        endswith_penalty = 0 if needle and hostname.endswith(f".{needle}") else 1
+        service_penalty = 1 if hostname.startswith("_") else 0
+        record_penalty = {"A": 0, "AAAA": 1, "CNAME": 1, "MX": 2, "TXT": 3, "SRV": 4}.get(record_type, 5)
+        length_penalty = abs(len(hostname) - len(needle)) if needle else len(hostname)
+        return (
+            exact_hostname,
+            hosted_penalty,
+            exact_zone,
+            exact_value,
+            startswith_penalty,
+            endswith_penalty,
+            service_penalty,
+            record_penalty,
+            length_penalty,
+            hostname,
+            item.category.value,
+            record_type,
+        )
+
     def search(self, term: str) -> list[HostnameSearchResult]:
         needle = term.strip().lower()
         matches: list[HostnameSearchResult] = []
@@ -287,7 +317,7 @@ def search(self, term: str) -> list[HostnameSearchResult]:
                         )
                     )
         if matches:
-            return sorted(matches, key=lambda item: (item.hostname, item.category.value, item.record_type or ""))
+            return sorted(matches, key=lambda item: self._search_rank(item, needle))
         try:
             hostname = normalize_hostname(term)
         except Exception:

tests/test_dns_service.py

@@ -311,6 +311,21 @@ def test_search_combines_hosted_dns_and_available_states() -> None:
     assert available[0].category is SearchCategory.AVAILABLE
 
 
+def test_search_prioritizes_hosted_and_app_records_over_service_noise() -> None:
+    _, service = build_service(
+        [
+            {"dname": "_autodiscover._tcp.app", "record_type": "SRV", "data": ["0"], "ttl": 300, "line_index": 1},
+            {"dname": "app", "record_type": "A", "address": "203.0.113.25", "ttl": 300, "line_index": 2},
+        ],
+        hosted=[{"domain": "app.ginutech.com", "rootdomain": "ginutech.com", "dir": "/public_html/app"}],
+    )
+
+    results = service.search("ginutech.com")
+    hostnames = [item.hostname for item in results]
+
+    assert hostnames.index("app.ginutech.com") < hostnames.index("_autodiscover._tcp.app.ginutech.com")
+
+
 def test_search_returns_available_when_hostname_is_free() -> None:
     _, service = build_service([])
     results = service.search("free.ginutech.com")

tests/test_interactive.py

@@ -110,8 +110,8 @@ class FakeRuntime:
         subdomains = FakeSubdomains()
         dns = FakeDNS()
 
-    answers = iter(["DNS: search", "Exit"])
-    browsed: list[list[HostnameSearchResult]] = []
+    answers = iter(["DNS: search", "Close results", "Exit"])
+    rendered: list[list[str]] = []
 
     class FakeSelect:
         def ask(self):
@@ -129,12 +129,83 @@ def ask(self):
     )
     monkeypatch.setattr("o2switch_cli.cli.interactive.TerminalUI.print_info", lambda self, message: None)
     monkeypatch.setattr(
-        "o2switch_cli.cli.interactive.TerminalUI.browse_pages",
-        lambda self, items, **kwargs: browsed.append(list(items)),
+        "o2switch_cli.cli.interactive.TerminalUI.print_hostname_search_results",
+        lambda self, items, page_window=None: rendered.append([item.hostname for item in items]),
     )
     monkeypatch.setattr("o2switch_cli.cli.interactive.TerminalUI.status", _noop_status)
+    monkeypatch.setattr("o2switch_cli.cli.interactive._page_size", lambda ui: 12)
 
     run_interactive_menu(_app_context())
 
     assert calls == ["name.ginutech.com"]
-    assert [item.hostname for item in browsed[0]] == ["name.ginutech.com"]
+    assert rendered[0] == ["name.ginutech.com"]
+
+
+def test_interactive_dns_search_can_refine_results_in_place(monkeypatch) -> None:
+    class FakeDomains:
+        def list_domains(self) -> list[DomainDescriptor]:
+            return [DomainDescriptor(domain="ginutech.com", type=DomainType.ADDON)]
+
+    class FakeSubdomains:
+        def search(self, term: str) -> list[SubdomainDescriptor]:
+            assert term == ""
+            return [SubdomainDescriptor(fqdn="app.ginutech.com", label="app", root_domain="ginutech.com")]
+
+    calls: list[str] = []
+
+    class FakeDNS:
+        def search(self, term: str) -> list[HostnameSearchResult]:
+            calls.append(term)
+            return [
+                HostnameSearchResult(
+                    category=SearchCategory.HOSTED_SUBDOMAINS,
+                    hostname="app.ginutech.com",
+                    zone="ginutech.com",
+                    managed_by_cpanel=True,
+                    docroot="/public_html/app",
+                ),
+                HostnameSearchResult(
+                    category=SearchCategory.DNS_RECORDS,
+                    hostname="_autodiscover._tcp.app.ginutech.com",
+                    record_type="SRV",
+                    value="0",
+                    zone="ginutech.com",
+                ),
+            ]
+
+    class FakeRuntime:
+        domains = FakeDomains()
+        subdomains = FakeSubdomains()
+        dns = FakeDNS()
+
+    answers = iter(["DNS: search", "Filter results", "Close results", "Exit"])
+    prompts = iter(["ginutech.com", "hosted"])
+    rendered: list[list[str]] = []
+
+    class FakeSelect:
+        def ask(self):
+            return next(answers)
+
+    monkeypatch.setattr(AppContext, "runtime", lambda self: FakeRuntime())
+    monkeypatch.setattr(
+        "o2switch_cli.cli.interactive.questionary.select",
+        lambda *args, **kwargs: FakeSelect(),
+    )
+    monkeypatch.setattr("o2switch_cli.cli.interactive.TerminalUI.print_banner", lambda self: None)
+    monkeypatch.setattr(
+        "o2switch_cli.cli.interactive.TerminalUI.prompt_realtime_search",
+        lambda self, *args, **kwargs: next(prompts),
+    )
+    monkeypatch.setattr("o2switch_cli.cli.interactive.TerminalUI.print_info", lambda self, message: None)
+    monkeypatch.setattr(
+        "o2switch_cli.cli.interactive.TerminalUI.print_hostname_search_results",
+        lambda self, items, page_window=None: rendered.append([item.hostname for item in items]),
+    )
+    monkeypatch.setattr("o2switch_cli.cli.interactive.TerminalUI.status", _noop_status)
+    monkeypatch.setattr("o2switch_cli.cli.interactive._page_size", lambda ui: 12)
+
+    run_interactive_menu(_app_context())
+
+    assert calls == ["ginutech.com"]
+    assert rendered[0] == ["app.ginutech.com", "_autodiscover._tcp.app.ginutech.com"]
+    assert rendered[1] == ["app.ginutech.com"]