gps-denied-onboard/tests/unit/test_az270_compose_root.py

"""AZ-270 — Composition Root AC tests.

Verifies the contract at ``_docs/02_document/contracts/shared_config/composition_root_protocol.md`` v1.0.0.
"""

from __future__ import annotations

import ast
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path

import pytest

from gps_denied_onboard.config import Config
from gps_denied_onboard.runtime_root import (
    RuntimeRoot,
    StrategyNotLinkedError,
    clear_strategy_registry,
    compose_operator,
    compose_root,
    list_registered_strategies,
    register_strategy,
)

_REPO_ROOT = Path(__file__).resolve().parents[2]


# AZ-847 — full rule-9 allow-list (single source of truth for the widened
# lint below). Mirrors `_docs/02_document/module-layout.md` rule 9
# "AZ-507 cross-component contract surface".
_RULE_9_ALLOW_EXACT_MODULES: frozenset[str] = frozenset(
    {
        "gps_denied_onboard._types",
        "gps_denied_onboard.helpers",
        "gps_denied_onboard.clock",
        "gps_denied_onboard.config",
        "gps_denied_onboard.logging",
        "gps_denied_onboard.fdr_client",
        # frame_source: interface-only per rule 9. The package root is allowed
        # because its `__init__.py` re-exports only the interface; concrete
        # implementations live in `frame_source.live_camera` /
        # `frame_source.video_file` and MUST be reached via DI from the
        # composition root, never imported by a Layer-3 component directly.
        "gps_denied_onboard.frame_source",
        "gps_denied_onboard.frame_source.interface",
    }
)
_RULE_9_ALLOW_PACKAGE_PREFIXES: tuple[str, ...] = (
    "gps_denied_onboard._types.",
    "gps_denied_onboard.helpers.",
    "gps_denied_onboard.clock.",
    "gps_denied_onboard.config.",
    "gps_denied_onboard.logging.",
    "gps_denied_onboard.fdr_client.",
)


def _is_rule9_allowed_import(module: str, *, own_component: str) -> bool:
    """Apply the rule-9 allow-list to a `gps_denied_onboard.*` import string.

    Returns True iff `module` matches one of: (1) the importing component's
    own subpackage, (2) a documented allow-list leaf module, (3) a documented
    allow-list package prefix.
    """
    own_pkg = f"gps_denied_onboard.components.{own_component}"
    if module == own_pkg or module.startswith(own_pkg + "."):
        return True
    if module in _RULE_9_ALLOW_EXACT_MODULES:
        return True
    return any(module.startswith(p) for p in _RULE_9_ALLOW_PACKAGE_PREFIXES)


@dataclass(frozen=True)
class _C1Block:
    strategy: str = "okvis2"


@dataclass(frozen=True)
class _C4Block:
    strategy: str = "opencv_gtsam"


@dataclass(frozen=True)
class _C5Block:
    strategy: str = "gtsam_isam2"


@dataclass(frozen=True)
class _C11Block:
    strategy: str = "ardupilot_tile_manager"


@dataclass
class _OrderRecorder:
    constructed: list[str]


@pytest.fixture(autouse=True)
def _isolated_registry() -> Iterator[None]:
    """Reset the strategy registry around every test."""
    clear_strategy_registry()
    yield
    clear_strategy_registry()


@pytest.fixture
def _airborne_env(monkeypatch: pytest.MonkeyPatch) -> None:
    for name, value in (
        ("GPS_DENIED_FC_PROFILE", "ardupilot_plane"),
        ("GPS_DENIED_TIER", "1"),
        ("DB_URL", "postgresql+psycopg://gps_denied:dev@db:5432/gps_denied"),
        ("CAMERA_CALIBRATION_PATH", "/etc/gps-denied/calib.yml"),
        ("LOG_LEVEL", "INFO"),
        ("LOG_SINK", "console"),
        ("INFERENCE_BACKEND", "pytorch_fp16"),
        ("FDR_PATH", "/var/lib/gps-denied/fdr"),
        ("TILE_CACHE_PATH", "/var/lib/gps-denied/tiles"),
        ("MAVLINK_SIGNING_KEY", "ZZZZZZZZ"),
    ):
        monkeypatch.setenv(name, value)


@pytest.fixture
def _operator_env(monkeypatch: pytest.MonkeyPatch) -> None:
    for name, value in (
        ("GPS_DENIED_FC_PROFILE", "ardupilot_plane"),
        ("GPS_DENIED_TIER", "1"),
        ("DB_URL", "postgresql+psycopg://gps_denied:dev@db:5432/gps_denied"),
        ("CAMERA_CALIBRATION_PATH", "/etc/gps-denied/calib.yml"),
        ("LOG_LEVEL", "INFO"),
        ("LOG_SINK", "console"),
        ("INFERENCE_BACKEND", "pytorch_fp16"),
        ("FDR_PATH", "/var/lib/gps-denied/fdr"),
        ("TILE_CACHE_PATH", "/var/lib/gps-denied/tiles"),
        ("SATELLITE_PROVIDER_URL", "http://localhost:8080"),
    ):
        monkeypatch.setenv(name, value)


def _register_okvis2(recorder: _OrderRecorder) -> None:
    def factory(config: Config, components: dict[str, object]) -> object:
        recorder.constructed.append("c1_vio")
        return ("c1_vio", "okvis2")

    register_strategy("c1_vio", "okvis2", factory, tier="airborne")


def _register_c4(recorder: _OrderRecorder) -> None:
    def factory(config: Config, components: dict[str, object]) -> object:
        recorder.constructed.append("c4_pose")
        return ("c4_pose", "opencv_gtsam")

    register_strategy("c4_pose", "opencv_gtsam", factory, tier="airborne")


def _register_c5(recorder: _OrderRecorder) -> None:
    def factory(config: Config, components: dict[str, object]) -> object:
        assert "c1_vio" in components, "c5_state factory ran before c1_vio existed"
        assert "c4_pose" in components, "c5_state factory ran before c4_pose existed"
        recorder.constructed.append("c5_state")
        return ("c5_state", "gtsam_isam2")

    register_strategy(
        "c5_state",
        "gtsam_isam2",
        factory,
        tier="airborne",
        depends_on=("c1_vio", "c4_pose"),
    )


def test_ac1_default_deployment_composes(_airborne_env: None) -> None:
    recorder = _OrderRecorder(constructed=[])
    _register_okvis2(recorder)
    _register_c4(recorder)
    _register_c5(recorder)
    config = Config.with_blocks(
        c1_vio=_C1Block(),
        c4_pose=_C4Block(),
        c5_state=_C5Block(),
    )
    root = compose_root(config)
    assert isinstance(root, RuntimeRoot)
    assert root.binary == "airborne"
    assert set(root.components.keys()) == {"c1_vio", "c4_pose", "c5_state"}


def test_ac2_strategy_not_linked_raises_with_payload(_airborne_env: None) -> None:
    # Only okvis2 is registered; config asks for vins_mono.
    recorder = _OrderRecorder(constructed=[])
    _register_okvis2(recorder)
    config = Config.with_blocks(c1_vio=_C1Block(strategy="vins_mono"))
    with pytest.raises(StrategyNotLinkedError) as info:
        compose_root(config)
    assert info.value.strategy_name == "vins_mono"
    assert info.value.component_slug == "c1_vio"
    assert info.value.available_strategies == ["okvis2"]


def test_ac3_operator_excludes_airborne_only(_operator_env: None) -> None:
    # c1_vio is registered as airborne; an operator config that references it must fail.
    recorder = _OrderRecorder(constructed=[])
    _register_okvis2(recorder)
    config = Config.with_blocks(c1_vio=_C1Block())
    with pytest.raises(StrategyNotLinkedError) as info:
        compose_operator(config)
    assert info.value.component_slug == "c1_vio"
    assert "airborne" in info.value.reason or "tier" in info.value.reason


def test_ac4_runtime_root_smoke_exit_zero(_airborne_env: None) -> None:
    # A Config with no component blocks must compose cleanly (every required
    # component is hard-wired by its bootstrap; no strategy to resolve).
    config = Config()
    root = compose_root(config)
    assert isinstance(root, RuntimeRoot)
    assert root.components == {}


def test_ac5_construction_order_respects_dependencies(_airborne_env: None) -> None:
    recorder = _OrderRecorder(constructed=[])
    # Register in reverse order to make the topological pass non-trivial.
    _register_c5(recorder)
    _register_c4(recorder)
    _register_okvis2(recorder)
    config = Config.with_blocks(
        c1_vio=_C1Block(),
        c4_pose=_C4Block(),
        c5_state=_C5Block(),
    )
    root = compose_root(config)
    # Dependencies must construct strictly before dependents.
    assert recorder.constructed.index("c1_vio") < recorder.constructed.index("c5_state")
    assert recorder.constructed.index("c4_pose") < recorder.constructed.index("c5_state")
    assert root.construction_order[-1] == "c5_state"


def test_ac6_only_compose_root_imports_concrete_strategies() -> None:
    """Architecture lint — full rule-9 allow-list enforcement (AZ-847 widening).

    Source of truth: ``_docs/02_document/module-layout.md`` rule 9
    "AZ-507 cross-component contract surface". For every
    ``components/<X>/*.py`` file, an ``ImportFrom`` or ``Import`` node
    referring to a ``gps_denied_onboard.*`` module is allowed iff it
    matches one of:

    1. The component's own subpackage
       (``gps_denied_onboard.components.<X>`` or ``...<X>.*``).
    2. ``gps_denied_onboard._types`` or any submodule (incl.
       ``_types.inference_errors``, ``_types.route``, etc.).
    3. ``gps_denied_onboard.helpers`` or any submodule.
    4. ``gps_denied_onboard.config`` / ``logging`` / ``fdr_client`` /
       ``clock``, or any submodule of those packages.
    5. ``gps_denied_onboard.frame_source`` or
       ``gps_denied_onboard.frame_source.interface`` only.

    AST-level disambiguation (Risk 2 of AZ-847): rule 9 limits
    ``frame_source`` to interface-only, but the package's ``__init__.py``
    re-exports only the interface symbols, so allow-listing the package
    root is safe. Concrete implementations
    (``frame_source.live_camera`` / ``frame_source.video_file``) and the
    error envelope (``frame_source.errors``) are intentionally NOT in
    the allow-list — they must be reached via constructor injection
    from the composition root, never imported by a Layer-3 component.

    Bench-code disambiguation: ``components/<X>/bench/**`` files are
    skipped because benchmark / measurement code legitimately
    constructs production strategies via ``runtime_root.*`` factories
    (that is its job). Rule 9 governs production runtime code, not
    measurement harnesses. Surfaced to the user during AZ-847
    implementation as a discovered architectural decision; documented
    here as a principled exclusion. A separate doc-only follow-up will
    propagate this exclusion into the rule-9 wording in
    ``module-layout.md``.

    Self-registration carve-out: components plug into the central
    strategy registry by lazy-importing ``register_<X>_*`` helpers from
    ``gps_denied_onboard.runtime_root.<X>_factory`` inside their
    ``register()`` function (e.g.
    ``c5_state.gtsam_isam2_estimator.register`` →
    ``runtime_root.state_factory.register_state_estimator``). This is
    the registry pattern, NOT cross-component coupling. The lint
    permits exactly this case: an ``ImportFrom`` whose module starts
    with ``gps_denied_onboard.runtime_root.`` AND every imported name
    starts with ``register_``. Anything else from ``runtime_root.*``
    (e.g. ``build_*`` factories, ``compose_root``,
    ``StrategyNotLinkedError``) inside a Layer-3 component remains a
    violation. Surfaced as a discovered architectural decision during
    AZ-847; same doc-only follow-up that updates rule-9 wording for
    bench/ will also document the self-registration carve-out.

    AZ-270 audit-trail compatibility (AC-2): this lint is a strict
    superset of the original AZ-270 AC-6 narrow check (cross-component
    imports). Every edge previously flagged is still flagged, plus new
    edges into ``replay_input``, ``replay_api``, ``cli/*``,
    non-allow-listed ``frame_source`` submodules, and any
    ``runtime_root.*`` import that is NOT a ``register_*`` self-
    registration helper.
    """
    components_root = _REPO_ROOT / "src" / "gps_denied_onboard" / "components"
    violations: list[str] = []
    for module_path in components_root.rglob("*.py"):
        relative_parts = module_path.relative_to(components_root).parts
        if "bench" in relative_parts:
            continue
        own_component = relative_parts[0]
        try:
            tree = ast.parse(module_path.read_text(encoding="utf-8"))
        except SyntaxError:
            continue
        for node in ast.walk(tree):
            module_names: tuple[str, ...]
            if isinstance(node, ast.ImportFrom) and node.module:
                if (
                    node.module.startswith("gps_denied_onboard.runtime_root.")
                    and node.names
                    and all(alias.name.startswith("register_") for alias in node.names)
                ):
                    continue
                module_names = (node.module,)
            elif isinstance(node, ast.Import):
                module_names = tuple(alias.name for alias in node.names)
            else:
                continue
            for name in module_names:
                if not name.startswith("gps_denied_onboard."):
                    continue
                if _is_rule9_allowed_import(name, own_component=own_component):
                    continue
                violations.append(
                    f"{module_path.relative_to(_REPO_ROOT)} imports {name} "
                    "(rule 9 allow-list violation)"
                )
    assert not violations, (
        "components/**/*.py imports MUST stay within the rule-9 allow-list "
        "(see _docs/02_document/module-layout.md rule 9 — AZ-507 "
        "cross-component contract surface). Violations:\n  - "
        + "\n  - ".join(violations)
    )


def test_nfr_reliability_partial_construction_closed_on_failure(_airborne_env: None) -> None:
    closed: list[str] = []

    class _Closable:
        def __init__(self, slug: str) -> None:
            self.slug = slug

        def close(self) -> None:
            closed.append(self.slug)

    def good_factory(config: Config, components: dict[str, object]) -> _Closable:
        return _Closable("c1_vio")

    def failing_factory(config: Config, components: dict[str, object]) -> object:
        raise RuntimeError("boom from c5_state factory")

    register_strategy("c1_vio", "okvis2", good_factory, tier="airborne")
    register_strategy(
        "c5_state",
        "gtsam_isam2",
        failing_factory,
        tier="airborne",
        depends_on=("c1_vio",),
    )
    config = Config.with_blocks(c1_vio=_C1Block(), c5_state=_C5Block())
    with pytest.raises(RuntimeError, match=r"boom"):
        compose_root(config)
    assert closed == ["c1_vio"], "prior instances must be .close()d on mid-composition failure"


def test_list_registered_strategies_returns_sorted_names() -> None:
    register_strategy("c1_vio", "okvis2", lambda c, m: None, tier="airborne")
    register_strategy("c1_vio", "vins_mono", lambda c, m: None, tier="airborne")
    register_strategy("c2_vpr", "netvlad", lambda c, m: None, tier="airborne")
    assert list_registered_strategies("c1_vio") == ["okvis2", "vins_mono"]
    assert list_registered_strategies("c2_vpr") == ["netvlad"]
    assert list_registered_strategies("c99_unknown") == []