refactor(svc-db): gate the Tor credential tests on one shared lookup

The three svc-db-* credential tests each hand-rolled the same transport
gate in their test.env.j2:

    (lookup('config', application_id, 'services.tor.enabled') | bool
     and lookup('config', application_id, 'services.<entity>.exposed') | bool)

Three copies of one predicate, each naming its own entity key, and each
free to drift from the onion mapping that actually decides reachability.

Extract it into plugins/lookup/tor_reachable.py, split the same way as
its sibling tor_ports.py: a plain is_reachable() for callers holding the
merged applications view, plus a thin LookupModule resolving it through
lookup('applications'). Because collect_exposed_ports() scans
services.values(), the entity name disappears from the templates
entirely -- they collapse to

    TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }}

Rework the integration guard accordingly: it now asserts that every
svc-db-* template delegates to the shared lookup, and drives its
base/variant agreement table through the real is_reachable() instead of
re-implementing the predicate. Add unit coverage for the new lookup.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

plugins/lookup/tor_reachable.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from typing import Any
+
+from ansible.errors import AnsibleError
+from ansible.plugins.loader import lookup_loader
+from ansible.plugins.lookup import LookupBase
+
+from plugins.lookup.tor_ports import _as_bool, collect_exposed_ports
+
+
+def is_reachable(applications: dict[str, Any], application_id: str) -> bool:
+    """True when *application_id* is reachable over the node onion, i.e. when
+    svc-net-tor is deployed for it (``services.tor.enabled``) AND at least one
+    of its services opts in with ``exposed: true``, which is what earns a
+    ``HiddenServicePort`` (``collect_exposed_ports``).
+
+    Both halves are required: ``tor.enabled`` is the dependency edge, not the
+    port forward, so a client picking the Tor transport from that flag alone
+    dials a port with no HiddenServicePort and no host publish.
+    """
+    config = (applications or {}).get(application_id) or {}
+    services = config.get("services") if isinstance(config, dict) else None
+    if not isinstance(services, dict):
+        return False
+    tor = services.get("tor")
+    if not (isinstance(tor, dict) and _as_bool(tor.get("enabled"))):
+        return False
+    return bool(collect_exposed_ports(applications, [application_id]))
+
+
+class LookupModule(LookupBase):
+    """
+    Usage:
+        {{ lookup('tor_reachable', application_id) }}
+
+    Returns whether *application_id* is reachable over the node onion in the
+    current (variant-merged) applications view -- see ``is_reachable``.
+
+    - parameters:
+        1) application_id
+    """
+
+    def run(
+        self,
+        terms: list[Any] | None,
+        variables: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> list[bool]:
+        if not terms or len(terms) != 1:
+            raise AnsibleError(
+                "lookup('tor_reachable', application_id) expects exactly 1 term."
+            )
+        application_id = str(terms[0]).strip()
+        if not application_id:
+            raise AnsibleError(
+                "lookup('tor_reachable', application_id): application_id is empty."
+            )
+
+        variables = variables or getattr(self._templar, "available_variables", {}) or {}
+        applications = lookup_loader.get(
+            "applications",
+            loader=getattr(self, "_loader", None),
+            templar=getattr(self, "_templar", None),
+        ).run([], variables=variables, roles_dir=kwargs.get("roles_dir"))[0]
+
+        return [is_reachable(applications, application_id)]

roles/svc-db-mariadb/files/test/test.sh

@@ -11,7 +11,7 @@
 #                container over TCP (forces password auth).
 #
 # Env (rendered into test.env from templates/test.env.j2):
-#   TOR_ENABLED     true|false        (services.tor.enabled, variant-aware)
+#   TOR_ENABLED     true|false        (lookup('tor_reachable', application_id))
 #   TOR_SOCKS       SOCKS proxy       (set in test.env.j2)
 #   ONION_HOST      node onion host   (svc-net-tor services.tor.node)
 #   DB_PORT         onion-forwarded database port

roles/svc-db-mariadb/templates/test.env.j2

@@ -3,5 +3,5 @@ DB_PASSWORD_B64={{ MARIADB_ROOT_PWD | b64encode }}
 DB_PORT={{ MARIADB_PORT }}
 DB_USER=root
 ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }}
-TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }}
+TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }}
 TOR_SOCKS={{ lookup('tor_socks') }}

roles/svc-db-openldap/files/test/test.sh

@@ -11,7 +11,7 @@
 #                (ldapwhoami over TCP).
 #
 # Env (rendered into test.env from templates/test.env.j2):
-#   TOR_ENABLED     true|false        (services.tor.enabled, variant-aware)
+#   TOR_ENABLED     true|false        (lookup('tor_reachable', application_id))
 #   TOR_SOCKS       SOCKS proxy       (set in test.env.j2)
 #   ONION_HOST      node onion host   (svc-net-tor services.tor.node)
 #   DB_PORT         onion-forwarded LDAP port

roles/svc-db-openldap/templates/test.env.j2

@@ -3,5 +3,5 @@ DB_CONTAINER={{ OPENLDAP_CONTAINER_ADDRESS }}
 DB_PASSWORD_B64={{ OPENLDAP_BIND_PW | b64encode }}
 DB_PORT={{ OPENLDAP_PORT }}
 ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }}
-TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }}
+TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }}
 TOR_SOCKS={{ lookup('tor_socks') }}

roles/svc-db-postgres/files/test/test.sh

@@ -11,7 +11,7 @@
 #                container over TCP (forces password auth via pg_hba).
 #
 # Env (rendered into test.env from templates/test.env.j2):
-#   TOR_ENABLED     true|false        (services.tor.enabled, variant-aware)
+#   TOR_ENABLED     true|false        (lookup('tor_reachable', application_id))
 #   TOR_SOCKS       SOCKS proxy       (set in test.env.j2)
 #   ONION_HOST      node onion host   (svc-net-tor services.tor.node)
 #   DB_PORT         onion-forwarded database port

roles/svc-db-postgres/templates/test.env.j2

@@ -4,5 +4,5 @@ DB_PASSWORD_B64={{ POSTGRES_PASSWORD | b64encode }}
 DB_PORT={{ POSTGRES_PORT }}
 DB_USER=postgres
 ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }}
-TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }}
+TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }}
 TOR_SOCKS={{ lookup('tor_socks') }}

tests/integration/roles/meta/services/test_db_tor_transport_gate.py

@@ -0,0 +1,76 @@
+"""Integration guard: for every ``svc-db-*`` role shipping an over-Tor
+credential test, assert that the transport its ``test.env.j2`` selects agrees
+with the onion mapping ``plugins.lookup.tor_ports.collect_exposed_ports``
+produces, in the base config and in every declared variant.
+"""
+
+from __future__ import annotations
+
+import re
+import unittest
+from pathlib import Path
+
+from plugins.lookup.tor_ports import collect_exposed_ports
+from plugins.lookup.tor_reachable import is_reachable
+from utils import PROJECT_ROOT
+from utils.cache.applications import get_merged_applications, get_variants
+from utils.cache.files import read_text
+
+ROLES_DIR = PROJECT_ROOT / "roles"
+
+_GATE_RE = re.compile(
+    r"TOR_ENABLED=\{\{\s*"
+    r"lookup\('tor_reachable',\s*application_id\)"
+    r"\s*\|\s*lower\s*\}\}"
+)
+
+
+def _roles_with_tor_credential_test() -> dict[str, Path]:
+    found = {}
+    for template in sorted(ROLES_DIR.glob("svc-db-*/templates/test.env.j2")):
+        if "TOR_ENABLED=" in read_text(str(template)):
+            found[template.parent.parent.name] = template
+    return found
+
+
+class TestDbTorTransportGate(unittest.TestCase):
+    def setUp(self) -> None:
+        self.templates = _roles_with_tor_credential_test()
+        self.assertTrue(self.templates, "no svc-db-* role renders TOR_ENABLED")
+
+    def test_template_delegates_to_the_shared_lookup(self) -> None:
+        for role, template in self.templates.items():
+            with self.subTest(role=role):
+                self.assertRegex(
+                    read_text(str(template)),
+                    _GATE_RE,
+                    f"{role} must select its transport with "
+                    f"lookup('tor_reachable', application_id) rather than a "
+                    f"hand-rolled flag expression, so the credential test and "
+                    f"the onion mapping (plugins/lookup/tor_ports.py) cannot "
+                    f"drift apart.",
+                )
+
+    def test_gate_agrees_with_the_onion_mapping(self) -> None:
+        base = get_merged_applications(roles_dir=str(ROLES_DIR))
+        variants = get_variants(roles_dir=str(ROLES_DIR))
+
+        for role in self.templates:
+            configs = [("base", base[role])]
+            configs += [(f"variant {i}", v) for i, v in enumerate(variants.get(role) or [])]
+
+            for label, config in configs:
+                with self.subTest(role=role, config=label):
+                    gate = is_reachable({role: config}, role)
+                    ports = collect_exposed_ports({role: config}, [role])
+                    self.assertEqual(
+                        gate,
+                        bool(ports),
+                        f"{role} [{label}]: the credential test would use "
+                        f"{'Tor' if gate else 'local'} transport while the deploy "
+                        f"forwards {ports or 'no'} port(s) over the onion",
+                    )
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/unit/plugins/lookup/test_tor_reachable.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from ansible.errors import AnsibleError
+
+from plugins.lookup.tor_reachable import LookupModule, is_reachable
+from utils.cache import _reset_cache_for_tests
+from utils.cache.yaml import dump_yaml_str
+from utils.roles.mapping import ROLE_FILE_META_SERVICES
+
+
+def _apps(tor_enabled: object, exposed: object) -> dict:
+    return {
+        "svc-db-mariadb": {
+            "services": {
+                "tor": {"enabled": tor_enabled},
+                "mariadb": {"exposed": exposed, "ports": {"local": {"mariadb": 3306}}},
+            }
+        }
+    }
+
+
+class TestIsReachable(unittest.TestCase):
+    def test_tor_enabled_and_exposed(self) -> None:
+        self.assertTrue(is_reachable(_apps(True, True), "svc-db-mariadb"))
+
+    def test_exposed_without_tor(self) -> None:
+        self.assertFalse(is_reachable(_apps(False, True), "svc-db-mariadb"))
+
+    def test_tor_without_exposed(self) -> None:
+        self.assertFalse(is_reachable(_apps(True, False), "svc-db-mariadb"))
+
+    def test_string_truthiness_matches_tor_ports(self) -> None:
+        self.assertTrue(is_reachable(_apps("true", "yes"), "svc-db-mariadb"))
+        self.assertFalse(is_reachable(_apps("true", "off"), "svc-db-mariadb"))
+
+    def test_exposed_service_without_local_port_has_no_onion_target(self) -> None:
+        apps = {
+            "svc-db-mariadb": {
+                "services": {
+                    "tor": {"enabled": True},
+                    "mariadb": {"exposed": True, "ports": {"public": {"tls": 3307}}},
+                }
+            }
+        }
+        self.assertFalse(is_reachable(apps, "svc-db-mariadb"))
+
+    def test_unknown_or_malformed_application(self) -> None:
+        self.assertFalse(is_reachable(_apps(True, True), "svc-db-nope"))
+        self.assertFalse(is_reachable({"a": None}, "a"))
+        self.assertFalse(is_reachable({}, "a"))
+
+
+class TestTorReachableLookup(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.roles_dir = Path(self._tmpdir.name) / "roles"
+        _reset_cache_for_tests()
+        for role, exposed in (("svc-db-mariadb", True), ("svc-db-postgres", False)):
+            path = self.roles_dir / role / ROLE_FILE_META_SERVICES
+            path.parent.mkdir(parents=True, exist_ok=True)
+            entity = role.rsplit("-", 1)[-1]
+            path.write_text(
+                dump_yaml_str(
+                    {
+                        "tor": {"enabled": True},
+                        entity: {
+                            "exposed": exposed,
+                            "ports": {"local": {entity: 3306}},
+                        },
+                    }
+                ),
+                encoding="utf-8",
+            )
+        self.lookup = LookupModule()
+        self.lookup._templar = None
+
+    def tearDown(self) -> None:
+        _reset_cache_for_tests()
+        self._tmpdir.cleanup()
+
+    def _run(self, application_id: str) -> bool:
+        return self.lookup.run(
+            [application_id], variables={}, roles_dir=str(self.roles_dir)
+        )[0]
+
+    def test_run_resolves_against_the_merged_applications_view(self) -> None:
+        self.assertTrue(self._run("svc-db-mariadb"))
+        self.assertFalse(self._run("svc-db-postgres"))
+
+    def test_arity_and_empty_term_raise(self) -> None:
+        for terms in ([], ["a", "b"], [""], ["   "]):
+            with self.subTest(terms=terms), self.assertRaises(AnsibleError):
+                self.lookup.run(terms, variables={}, roles_dir=str(self.roles_dir))
+
+
+if __name__ == "__main__":
+    unittest.main()