gps-denied-onboard/e2e/_unit_tests/fixtures/test_age_injector.py

"""Tests for the AZ-407 age-injector.

Covers AC-3 (capture_date shifted, pixels bit-identical) and AC-7
(provenance docs present).
"""

from __future__ import annotations

import csv
import datetime as _dt
import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[3]
INPUT_DIR = REPO_ROOT / "_docs" / "00_problem" / "input_data"
BUILDER_PY = REPO_ROOT / "e2e" / "fixtures" / "tile-cache-builder" / "builder.py"
INJECTOR_PY = REPO_ROOT / "e2e" / "fixtures" / "age-injector" / "age_injector.py"
INJECTOR_DIR = REPO_ROOT / "e2e" / "fixtures" / "age-injector"


def _run(cmd: list[str]) -> str:
    """Run a subprocess, return stdout (raises on failure)."""

    env = dict(os.environ, PYTHONHASHSEED="0")
    result = subprocess.run(cmd, check=True, capture_output=True, text=True, env=env)
    return result.stdout


def _build_source_cache(out_dir: Path) -> Path:
    """Run the tile-cache builder; return the populated dir."""

    _run(
        [
            sys.executable,
            str(BUILDER_PY),
            "--input-dir",
            str(INPUT_DIR),
            "--output-dir",
            str(out_dir),
            "--quiet",
        ]
    )
    return out_dir


def _file_hashes(root: Path, suffix: str) -> dict[str, str]:
    return {
        p.relative_to(root).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest()
        for p in sorted(root.rglob(f"*{suffix}"))
    }


@pytest.fixture(scope="module")
def source_cache(tmp_path_factory: pytest.TempPathFactory) -> Path:
    """One-shot module-scoped tile-cache build (~1s)."""

    return _build_source_cache(tmp_path_factory.mktemp("source-cache"))


@pytest.mark.parametrize("age_months,threshold_days", [(7, 6 * 30), (13, 12 * 30)])
def test_age_injector_shifts_capture_date(
    tmp_path: Path,
    source_cache: Path,
    age_months: int,
    threshold_days: int,
) -> None:
    """AC-3: every manifest row's capture_date is now - age_months ±1 day."""

    # Arrange
    out = tmp_path / f"out-{age_months}mo"
    today = _dt.datetime.now(tz=_dt.timezone.utc).date()

    # Act
    _run(
        [
            sys.executable,
            str(INJECTOR_PY),
            "--source-dir",
            str(source_cache),
            "--output-dir",
            str(out),
            "--age-months",
            str(age_months),
        ]
    )

    # Assert
    with (out / "manifest.csv").open() as fp:
        rows = list(csv.DictReader(fp))
    assert rows, "aged manifest is empty"
    for r in rows:
        shifted = _dt.date.fromisoformat(r["capture_date"])
        delta_days = (today - shifted).days
        target_days = int(round(age_months * 30.44))
        assert abs(delta_days - target_days) <= 1, (
            f"row {r['tile_x']},{r['tile_y']}: capture_date offset is "
            f"{delta_days} days, expected {target_days} ±1"
        )
        assert delta_days > threshold_days, (
            f"aged capture_date {r['capture_date']} did not exceed the "
            f"{threshold_days}-day threshold"
        )


def test_age_injector_preserves_tile_bytes(tmp_path: Path, source_cache: Path) -> None:
    """AC-3: tile JPEG bodies copy bit-identical."""

    # Arrange
    out = tmp_path / "out-7mo"

    # Act
    _run(
        [
            sys.executable,
            str(INJECTOR_PY),
            "--source-dir",
            str(source_cache),
            "--output-dir",
            str(out),
            "--age-months",
            "7",
        ]
    )

    # Assert
    src_hashes = _file_hashes(source_cache / "tiles", ".jpg")
    out_hashes = _file_hashes(out / "tiles", ".jpg")
    assert src_hashes == out_hashes, "tile JPEG bytes drifted across age injection"


def test_age_injector_updates_sidecar_dates(tmp_path: Path, source_cache: Path) -> None:
    """AC-3: per-tile sidecar JSON also reflects the aged date."""

    # Arrange
    out = tmp_path / "out-13mo"

    # Act
    _run(
        [
            sys.executable,
            str(INJECTOR_PY),
            "--source-dir",
            str(source_cache),
            "--output-dir",
            str(out),
            "--age-months",
            "13",
        ]
    )

    # Assert
    today = _dt.datetime.now(tz=_dt.timezone.utc).date()
    target_days = int(round(13 * 30.44))
    for sidecar in sorted((out / "tiles").rglob("*.json")):
        data = json.loads(sidecar.read_text())
        shifted = _dt.date.fromisoformat(data["capture_date"])
        delta = (today - shifted).days
        assert abs(delta - target_days) <= 1, (
            f"sidecar {sidecar}: capture_date offset {delta}d, expected {target_days}d ±1"
        )


def test_age_injector_rejects_non_positive_months(tmp_path: Path, source_cache: Path) -> None:
    """Defensive: zero or negative age_months must error out, not silently no-op."""

    # Arrange
    out = tmp_path / "rejected"

    # Act + Assert
    with pytest.raises(subprocess.CalledProcessError) as excinfo:
        _run(
            [
                sys.executable,
                str(INJECTOR_PY),
                "--source-dir",
                str(source_cache),
                "--output-dir",
                str(out),
                "--age-months",
                "0",
            ]
        )
    assert "must be positive" in (excinfo.value.stderr or "")


def test_age_injector_provenance_readme_exists() -> None:
    """AC-7: README documents the injector."""

    # Arrange / Act
    readme = INJECTOR_DIR / "README.md"

    # Assert
    assert readme.exists()
    content = readme.read_text()
    assert "Provenance" in content
    assert "Reproducibility" in content