NODEDC_MISSION_CORE/tests/test_lidar_preparation.py

from __future__ import annotations

import hashlib
import json
from pathlib import Path

import pytest
from test_lidar_replay import _capture

from k1link.compute import lidar_replay
from k1link.compute.lidar_preparation import prepare_lidar_replay_pack_v2
from k1link.compute.lidar_replay import LidarReplayError


def test_warm_preparation_reuses_legacy_identity_without_source_decode(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = _capture(tmp_path)
    producer = Path(lidar_replay.__file__)
    producer_digest = hashlib.sha256(producer.read_bytes()).hexdigest()
    original = lidar_replay.build_lidar_replay_pack_v2(source, tmp_path / "packs")
    manifest_before = (original / "manifest.json").read_bytes()
    closed = []
    original_close = lidar_replay.LidarReplayPackV2.close

    def no_decode(*args: object, **kwargs: object) -> None:
        pytest.fail("cache hit must not decode the original capture")

    def close(pack: lidar_replay.LidarReplayPackV2) -> None:
        original_close(pack)
        closed.append(not pack.arrays._arrays)

    monkeypatch.setattr(lidar_replay, "_capture_arrays", no_decode)
    monkeypatch.setattr(lidar_replay.LidarReplayPackV2, "close", close)
    assert prepare_lidar_replay_pack_v2(source, tmp_path / "packs") == original
    assert (original / "manifest.json").read_bytes() == manifest_before
    assert json.loads(manifest_before)["identity"]["producer_sha256"] == producer_digest
    assert hashlib.sha256(producer.read_bytes()).hexdigest() == producer_digest
    assert closed == [True]


def test_cold_preparation_calls_existing_builder_once(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = _capture(tmp_path)
    calls = []
    original = lidar_replay._capture_arrays

    def decode(path: Path):
        calls.append(path)
        return original(path)

    monkeypatch.setattr(lidar_replay, "_capture_arrays", decode)
    result = prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    assert result.is_dir()
    assert calls == [source]


@pytest.mark.parametrize("changed", ["metadata", "origin", "remove-origin", "raw", "session"])
def test_changed_source_never_reuses_previous_pack(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    changed: str,
) -> None:
    source = _capture(tmp_path)
    output = tmp_path / "packs"
    old = prepare_lidar_replay_pack_v2(source, output)
    if changed == "metadata":
        path = source.with_name("mqtt.metadata.jsonl")
        path.write_bytes(path.read_bytes().replace(b"5010000000", b"5009000000"))
    elif changed == "origin":
        path = source.with_name("mqtt.timeline.origin.json")
        path.write_bytes(path.read_bytes().replace(b"5000000000", b"4999999999"))
    elif changed == "remove-origin":
        source.with_name("mqtt.timeline.origin.json").unlink()
    elif changed == "raw":
        # The cache boundary checks bytes before trying to decode invalid data.
        source.write_bytes(source.read_bytes()[:-1] + b"x")
    called = []

    def build(*args: object, **kwargs: object) -> Path:
        called.append((args, kwargs))
        return output / "new-test-pack"

    monkeypatch.setattr(lidar_replay, "build_lidar_replay_pack_v2", build)
    result = prepare_lidar_replay_pack_v2(
        source,
        output,
        session_id="other-session" if changed == "session" else None,
    )
    assert result != old
    assert len(called) == 1
    assert old.is_dir()


@pytest.mark.parametrize(
    "artifact", ["lidar-replay.npz", "quality-report.json", "equivalence-report.json"]
)
def test_corrupt_matching_pack_fails_without_rebuild_or_overwrite(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    artifact: str,
) -> None:
    source = _capture(tmp_path)
    root = prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    path = root / artifact
    damaged = b"damaged-evidence"
    path.write_bytes(damaged)

    def forbidden(*args: object, **kwargs: object) -> None:
        pytest.fail("must preserve corrupt evidence instead of overwriting")

    monkeypatch.setattr(lidar_replay, "build_lidar_replay_pack_v2", forbidden)
    with pytest.raises(LidarReplayError, match="artifact identity changed"):
        prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    assert path.read_bytes() == damaged


def test_report_threshold_collision_is_explicit_and_releases_arrays(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = _capture(tmp_path)
    root = prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    closed = []
    original = lidar_replay.LidarReplayPackV2.close

    def close(pack: lidar_replay.LidarReplayPackV2) -> None:
        original(pack)
        closed.append(not pack.arrays._arrays)

    monkeypatch.setattr(lidar_replay.LidarReplayPackV2, "close", close)
    with pytest.raises(LidarReplayError, match="threshold differs"):
        prepare_lidar_replay_pack_v2(source, tmp_path / "packs", pose_coverage_threshold_ms=50)
    assert closed == [True]
    assert root.is_dir()


def test_source_mutation_during_cache_validation_is_rejected(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    source = _capture(tmp_path)
    prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    original = lidar_replay.LidarReplayPackV2.__init__

    def changed(pack: lidar_replay.LidarReplayPackV2, root: Path) -> None:
        original(pack, root)
        source.with_name("mqtt.metadata.jsonl").touch()

    monkeypatch.setattr(lidar_replay.LidarReplayPackV2, "__init__", changed)
    with pytest.raises(LidarReplayError, match="changed during preparation"):
        prepare_lidar_replay_pack_v2(source, tmp_path / "packs")


def test_missing_metadata_never_uses_cache(tmp_path: Path) -> None:
    source = _capture(tmp_path)
    prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    source.with_name("mqtt.metadata.jsonl").unlink()
    with pytest.raises(LidarReplayError, match="exact host timing"):
        prepare_lidar_replay_pack_v2(source, tmp_path / "packs")


def test_symlink_manifest_is_not_followed(tmp_path: Path) -> None:
    source = _capture(tmp_path)
    root = prepare_lidar_replay_pack_v2(source, tmp_path / "packs")
    manifest = root / "manifest.json"
    saved = root / "saved-manifest.json"
    manifest.rename(saved)
    manifest.symlink_to(saved)
    with pytest.raises(LidarReplayError, match="manifest cannot be a symlink"):
        prepare_lidar_replay_pack_v2(source, tmp_path / "packs")