"""Every auditor adapter must point at a driver kind that EXISTS. Three adapters (Google / GitHub / Atlassian) shipped with ``KIND = "software"``. No such kind is registered — the registry only ever knew ``software_provisioning`` — and ``_resolve_integration`` matches ``Integration.kind`` exactly, with no alias. So every call from those adapters raised ``IntegrationMissingError``, which the sync services record as an INFO-level "integration not configured". Result: Google Reports sign-in events (including the highest-weighted "logged in after offboarding" detector), GitHub org PAT/2FA and Atlassian public-space scanning produced nothing from day one, while ``/coverage`` — which checks the CORRECT kind — displayed "configured" and made the hole look like health. See #2122. Nothing could have caught it: the adapters' own tests all mock at the method layer, so the constant was never exercised against the registry. This test closes that gap for every adapter at once, and additionally pins the ``/coverage`` declarations to the same registry so the two halves cannot drift apart again. """ from __future__ import annotations import ast import inspect import textwrap from pathlib import Path import pytest from app.domains.auditor import integration_adapters as adapters from app.integrations import registry def _auditor_adapter_classes() -> list[type]: return [ obj for _name, obj in inspect.getmembers(adapters, inspect.isclass) if obj.__module__ == adapters.__name__ and hasattr(obj, "KIND") and hasattr(obj, "PROVIDER") ] def test_there_are_adapters_to_check() -> None: """A passing empty sweep would be worthless.""" assert len(_auditor_adapter_classes()) >= 6 @pytest.mark.parametrize( "cls", _auditor_adapter_classes(), ids=lambda c: c.__name__, ) def test_adapter_kind_provider_is_a_registered_driver(cls: type) -> None: kind, provider = cls.KIND, cls.PROVIDER assert kind in registry.kinds(), ( f"{cls.__name__}.KIND={kind!r} is not a registered driver kind. " f"Known kinds: {registry.kinds()}. _resolve_integration matches " f"Integration.kind exactly, so every read from this adapter would " f"raise IntegrationMissingError and be swallowed as " f"'not configured'." ) assert registry.get_driver_class(kind, provider) is not None, ( f"{cls.__name__} points at {kind}/{provider}, which no driver " f"registers. Providers under {kind}: " f"{sorted(registry.providers_for_kind(kind))}" ) def _coverage_entry_pairs() -> list[tuple[str, str]]: """(kind, provider) pairs the /coverage endpoint declares. Parsed from the source because the values are literals inside ``_entry(...)`` calls in a long request handler — running the endpoint would only tell us what one code path reached. """ src = Path(inspect.getfile( __import__("app.domains.auditor.router", fromlist=["router"]) )).read_text(encoding="utf-8") pairs: list[tuple[str, str]] = [] for node in ast.walk(ast.parse(src)): if not isinstance(node, ast.Call): continue if not (isinstance(node.func, ast.Name) and node.func.id == "_entry"): continue kw = { k.arg: k.value for k in node.keywords if isinstance(k.value, ast.Constant) } kind, provider = kw.get("kind"), kw.get("provider") if isinstance(kind, ast.Constant) and isinstance(provider, ast.Constant): pairs.append((kind.value, provider.value)) return sorted(set(pairs)) def test_coverage_endpoint_declares_only_real_driver_pairs() -> None: pairs = _coverage_entry_pairs() assert len(pairs) >= 8, "coverage entries not found — did _entry rename?" unknown = [ (k, p) for k, p in pairs if registry.get_driver_class(k, p) is None ] assert not unknown, ( f"/coverage claims these integrations gate its controls, but no " f"driver registers them: {unknown}. A pair that cannot exist " f"renders as a permanent '未配置' the operator can never clear." ) def test_every_adapter_pair_is_visible_on_coverage() -> None: """The adapter set and the coverage set must describe one system. #2122's real damage was the ASYMMETRY: the adapter read one kind while /coverage checked another, so the dashboard reported health for a pipeline that had never run. """ coverage = set(_coverage_entry_pairs()) for cls in _auditor_adapter_classes(): pair = (cls.KIND, cls.PROVIDER) assert pair in coverage, ( f"{cls.__name__} reads {pair} but /coverage never mentions it, " f"so an operator cannot tell whether this data source is " f"configured. Add an _entry(...) for it." ) # ───────────────────────────────────────────────────────────────────── # Same failure mode, one layer in: the KIND is right but the CONFIG KEY # the adapter reads is not one the driver ever stores (2026-08-18). # ───────────────────────────────────────────────────────────────────── def _config_get_chains(cls: type) -> list[tuple[int, tuple[str, ...]]]: """``[(lineno, ("admin_email", "subject"))]`` per ``config.get`` chain. An ``a or b or c`` fallback chain counts as ONE requirement: the adapter needs *one* of those keys to exist. A standalone ``config.get("x")`` is a chain of one. Parsed from source because these literals sit inside inner closures that only run against a live integration — which is exactly why the Google mismatch survived every mocked-at-the-method-layer test. """ src = inspect.getsource(cls) tree = ast.parse(textwrap.dedent(src)) def key_of(node: ast.AST) -> str | None: if not isinstance(node, ast.Call): return None func = node.func if not (isinstance(func, ast.Attribute) and func.attr == "get"): return None if not (isinstance(func.value, ast.Name) and func.value.id == "config"): return None if node.args and isinstance(node.args[0], ast.Constant): value = node.args[0].value return value if isinstance(value, str) else None return None chains: list[tuple[int, tuple[str, ...]]] = [] grouped: set[int] = set() for node in ast.walk(tree): if not (isinstance(node, ast.BoolOp) and isinstance(node.op, ast.Or)): continue keys = [] for value in node.values: key = key_of(value) if key is not None: keys.append(key) grouped.add(id(value)) if keys: chains.append((node.lineno, tuple(keys))) for node in ast.walk(tree): key = key_of(node) if key is not None and id(node) not in grouped: chains.append((node.lineno, (key,))) return chains def test_there_are_config_chains_to_check() -> None: """Guard the guard: a parser that silently finds nothing proves nothing.""" found = { cls.__name__: _config_get_chains(cls) for cls in _auditor_adapter_classes() } assert sum(len(v) for v in found.values()) >= 10, found google = dict( (keys, line) for line, keys in found["GoogleAuditorAdapter"] ) assert any("impersonate_subject_email" in keys for keys in google), found @pytest.mark.parametrize( "cls", _auditor_adapter_classes(), ids=lambda c: c.__name__, ) def test_adapter_config_keys_exist_in_the_driver_schema(cls: type) -> None: """Every key an adapter reads must be one its driver actually stores. ``GoogleAuditorAdapter`` looked for ``admin_email`` / ``delegated_subject`` / ``subject`` while the ``google_workspace`` driver declares ``impersonate_subject_email``. The guard clause then raised ``IntegrationMissingError`` — whose message says "no active integration found" — so MFA status, sign-in events and admin-role assignments returned zero rows for the module's whole lifetime while the integration sat there enabled, and every downstream surface read the silence as "0% enrolled". """ driver = registry.get_driver_class(cls.KIND, cls.PROVIDER) assert driver is not None # covered by the test above declared = { field.key for field in (getattr(driver, "config_schema", None) or ()) } if not declared: pytest.skip(f"{cls.KIND}/{cls.PROVIDER} declares no config_schema") unsatisfiable = [ (line, keys) for line, keys in _config_get_chains(cls) if not (set(keys) & declared) ] assert not unsatisfiable, ( f"{cls.__name__} reads config keys that {cls.KIND}/{cls.PROVIDER} " f"never stores: {unsatisfiable}. Declared keys: {sorted(declared)}. " f"Each entry is one `or` fallback chain, so at least ONE key in it " f"must be a declared field — otherwise the read is dead code and " f"the adapter fails as if the integration did not exist." )