gps-denied-onboard/tests/unit/c7_inference/test_engine_gate.py

"""AZ-301 — ``EngineGate`` acceptance tests.

AC-1..AC-7 + NFR-reliability-no-write cover the five-step deterministic
refusal pipeline plus the happy path. AC-8 (``read_host_tuple`` on a
real Jetson) is Tier-2-only and lives in the integration suite — it
needs an actual NVML-readable GPU + ``/etc/nv_tegra_release``.
"""

from __future__ import annotations

import hashlib
import json
import logging
import os
from pathlib import Path

import pytest

from gps_denied_onboard._types.inference import EngineCacheEntry, PrecisionMode
from gps_denied_onboard.components.c7_inference import (
    DeploymentManifest,
    EngineGate,
    EngineHashMismatchError,
    EngineSchemaMismatchError,
    EngineSidecarMissingError,
    HostTuple,
    ManifestReader,
)

_TIER2_HOST = HostTuple(sm=87, jp="6.2", trt="10.3", precision=PrecisionMode.FP16)


def _write_engine(tmp_path: Path, name: str, payload: bytes = b"fake-engine") -> Path:
    engine_path = tmp_path / name
    engine_path.write_bytes(payload)
    return engine_path


def _write_sidecar(engine_path: Path, hex_digest: str | None = None) -> str:
    if hex_digest is None:
        hex_digest = hashlib.sha256(engine_path.read_bytes()).hexdigest()
    sidecar = Path(str(engine_path) + ".sha256")
    sidecar.write_text(hex_digest, encoding="utf-8")
    return hex_digest


def _entry_for(engine_path: Path) -> EngineCacheEntry:
    return EngineCacheEntry(
        engine_path=engine_path,
        sha256_hex=hashlib.sha256(engine_path.read_bytes()).hexdigest(),
        sm=_TIER2_HOST.sm,
        jp=_TIER2_HOST.jp,
        trt=_TIER2_HOST.trt,
        precision=PrecisionMode.FP16,
        extras={},
    )


def _manifest_with(engine_path: Path, sha_hex: str) -> DeploymentManifest:
    relative = engine_path.relative_to(engine_path.parent)
    return DeploymentManifest(
        root=engine_path.parent,
        entries={relative.as_posix(): sha_hex},
    )


# ----------------------------------------------------------------------
# AC-1: filename-schema parse failure refused at parse time.


def test_ac1_parse_failure_refused_at_parse_time(tmp_path: Path) -> None:
    # Arrange
    bogus = _write_engine(tmp_path, "bogus_name.engine")
    entry = _entry_for(bogus)
    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineSchemaMismatchError, match="AZ-281 schema"):
        gate.validate(entry, _TIER2_HOST, _manifest_with(bogus, "0" * 64))
    assert not Path(str(bogus) + ".sha256").exists(), (
        "Gate must NOT have created a sidecar during refusal"
    )


# ----------------------------------------------------------------------
# AC-2: filename-schema tuple mismatch refused at parse time.


def test_ac2_schema_tuple_mismatch(tmp_path: Path) -> None:
    # Arrange — engine compiled for sm=86 but host is sm=87.
    engine = _write_engine(tmp_path, "ultravpr__sm86_jp6.2_trt10.3_fp16.engine")
    entry = _entry_for(engine)
    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineSchemaMismatchError) as exc_info:
        gate.validate(entry, _TIER2_HOST, _manifest_with(engine, "0" * 64))
    msg = str(exc_info.value)
    assert "sm=86" in msg
    assert "sm=87" in msg


# ----------------------------------------------------------------------
# AC-3: missing sidecar refused before manifest lookup.


def test_ac3_missing_sidecar_refused_before_manifest(tmp_path: Path) -> None:
    # Arrange
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    entry = _entry_for(engine)
    manifest_calls: list[Path] = []

    class _TrackingManifest:
        def read(self) -> DeploymentManifest:
            manifest_calls.append(engine)
            return _manifest_with(engine, "0" * 64)

    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineSidecarMissingError, match=r"\.sha256"):
        gate.validate(entry, _TIER2_HOST, _TrackingManifest())
    # Manifest may be read at step 0 (validate prologue) but the missing-sidecar
    # check refuses BEFORE manifest lookup; no manifest *__getitem__* call.


# ----------------------------------------------------------------------
# AC-4: sidecar trust failure.


def test_ac4_sidecar_hash_mismatches_file(tmp_path: Path) -> None:
    # Arrange
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    _write_sidecar(engine, hex_digest="d" * 64)  # wrong hash
    entry = _entry_for(engine)
    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineHashMismatchError, match="sidecar"):
        gate.validate(entry, _TIER2_HOST, _manifest_with(engine, "d" * 64))


# ----------------------------------------------------------------------
# AC-5: manifest mismatch.


def test_ac5_manifest_hash_mismatches_sidecar(tmp_path: Path) -> None:
    # Arrange
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    true_hash = _write_sidecar(engine)
    bogus_manifest = _manifest_with(engine, "f" * 64)
    entry = _entry_for(engine)
    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineHashMismatchError, match="manifest hash"):
        gate.validate(entry, _TIER2_HOST, bogus_manifest)
    assert "f" * 64 in str(true_hash) or true_hash != "f" * 64


# ----------------------------------------------------------------------
# AC-6: full-success path returns silently and logs INFO.


def test_ac6_full_success_returns_silently_and_logs_pass(
    tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    # Arrange
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    sha = _write_sidecar(engine)
    manifest = _manifest_with(engine, sha)
    entry = _entry_for(engine)
    gate = EngineGate()
    # Act
    with caplog.at_level(logging.INFO, logger="c7_inference.gate"):
        result = gate.validate(entry, _TIER2_HOST, manifest)
    # Assert
    assert result is None
    pass_records = [
        r for r in caplog.records if getattr(r, "kind", None) == "c7.gate.pass"
    ]
    assert len(pass_records) == 1, (
        f"expected exactly one c7.gate.pass record; got {len(pass_records)} "
        f"of {[r.message for r in caplog.records]}"
    )


# ----------------------------------------------------------------------
# AC-7: refusal order is deterministic — schema wins over sidecar-missing.


def test_ac7_schema_error_wins_over_sidecar_missing(tmp_path: Path) -> None:
    # Arrange — both broken: schema mismatch AND no sidecar on disk.
    engine = _write_engine(tmp_path, "bogus.engine")
    entry = _entry_for(engine)
    gate = EngineGate()
    # Act / Assert
    with pytest.raises(EngineSchemaMismatchError):
        gate.validate(entry, _TIER2_HOST, _manifest_with(engine, "0" * 64))


# ----------------------------------------------------------------------
# Manifest reader coverage.


def test_manifest_reader_round_trip(tmp_path: Path) -> None:
    manifest_path = tmp_path / "manifest.json"
    payload = {
        "root": str(tmp_path),
        "entries": {
            "ultravpr__sm87_jp6.2_trt10.3_fp16.engine": "a" * 64,
            "lightglue__sm87_jp6.2_trt10.3_int8.engine": "b" * 64,
        },
    }
    manifest_path.write_text(json.dumps(payload), encoding="utf-8")
    manifest = ManifestReader(manifest_path).read()
    assert manifest.root == tmp_path
    assert manifest["ultravpr__sm87_jp6.2_trt10.3_fp16.engine"] == "a" * 64
    assert "lightglue__sm87_jp6.2_trt10.3_int8.engine" in manifest


def test_manifest_missing_entry_raises_hash_mismatch() -> None:
    """Risk 2 — missing manifest entry MUST raise EngineHashMismatchError, NOT KeyError."""
    manifest = DeploymentManifest(root=Path("/var/lib/x"), entries={})
    with pytest.raises(EngineHashMismatchError, match="missing manifest entry"):
        _ = manifest["unknown.engine"]


def test_manifest_reader_rejects_malformed_json(tmp_path: Path) -> None:
    bad = tmp_path / "manifest.json"
    bad.write_text("not json {", encoding="utf-8")
    with pytest.raises(EngineHashMismatchError, match="not valid JSON"):
        ManifestReader(bad).read()


def test_manifest_reader_rejects_missing_entries_key(tmp_path: Path) -> None:
    bad = tmp_path / "manifest.json"
    bad.write_text(json.dumps({"root": "/x"}), encoding="utf-8")
    with pytest.raises(EngineHashMismatchError, match="entries"):
        ManifestReader(bad).read()


# ----------------------------------------------------------------------
# Engine-outside-manifest-root path.


def test_engine_outside_manifest_root_refused(tmp_path: Path) -> None:
    # Arrange — engine lives under tmp_path but manifest root is a sibling dir.
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    sha = _write_sidecar(engine)
    sibling = tmp_path / "elsewhere"
    sibling.mkdir()
    manifest = DeploymentManifest(
        root=sibling,
        entries={"ultravpr__sm87_jp6.2_trt10.3_fp16.engine": sha},
    )
    gate = EngineGate()
    with pytest.raises(EngineHashMismatchError, match="not under manifest root"):
        gate.validate(_entry_for(engine), _TIER2_HOST, manifest)


# ----------------------------------------------------------------------
# NFR-reliability — no writes by the gate.


def test_nfr_reliability_no_writes(tmp_path: Path) -> None:
    """The gate is read-only — it must not touch the engine, sidecar, or manifest."""
    engine = _write_engine(tmp_path, "ultravpr__sm87_jp6.2_trt10.3_fp16.engine")
    sha = _write_sidecar(engine)
    sidecar = Path(str(engine) + ".sha256")
    snapshot = {
        "engine_mtime": engine.stat().st_mtime,
        "engine_bytes": engine.read_bytes(),
        "sidecar_mtime": sidecar.stat().st_mtime,
        "sidecar_text": sidecar.read_text(encoding="utf-8"),
    }
    manifest = _manifest_with(engine, sha)
    EngineGate().validate(_entry_for(engine), _TIER2_HOST, manifest)
    assert engine.stat().st_mtime == snapshot["engine_mtime"]
    assert engine.read_bytes() == snapshot["engine_bytes"]
    assert sidecar.stat().st_mtime == snapshot["sidecar_mtime"]
    assert sidecar.read_text(encoding="utf-8") == snapshot["sidecar_text"]


# ----------------------------------------------------------------------
# Refusal log on error path.


def test_refusal_emits_error_log(
    tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
    engine = _write_engine(tmp_path, "bogus.engine")
    gate = EngineGate()
    with caplog.at_level(logging.ERROR, logger="c7_inference.gate"):
        with pytest.raises(EngineSchemaMismatchError):
            gate.validate(_entry_for(engine), _TIER2_HOST, _manifest_with(engine, "0" * 64))
    refusals = [
        r for r in caplog.records if getattr(r, "kind", None) == "c7.gate.refuse"
    ]
    assert len(refusals) == 1
    assert refusals[0].kv["step"] == "schema_parse"  # type: ignore[attr-defined]


# ----------------------------------------------------------------------
# AC-8 host-tuple read (Tier-2 hardware only).


@pytest.mark.tier2
@pytest.mark.skipif(
    os.environ.get("GPS_DENIED_TIER") != "2",
    reason="read_host_tuple needs real Jetson hardware + NVML + L4T release file",
)
def test_ac8_read_host_tuple_on_jetson() -> None:
    from gps_denied_onboard.components.c7_inference.engine_gate import read_host_tuple

    host = read_host_tuple(precision=PrecisionMode.FP16)
    assert host.sm == 87
    assert host.jp == "6.2"
    assert host.trt == "10.3"
    assert host.precision is PrecisionMode.FP16