gps-denied-onboard/e2e/runner/helpers/injector_fixtures.py

"""pytest fixtures wrapping the AZ-408 runtime synthetic-injection injectors.

Per-scenario tests (FT-N-01, FT-N-04, FT-P-08, NFT-RES-04, NFT-PERF-04)
opt into an injector by requesting one of the fixtures below. Each
fixture:

1. Builds the injector output under the pytest ``tmp_path_factory`` root
   (so unit-test runs never touch ``/tmp``).
2. Yields a typed handle the test asserts against (out_root, schedule,
   summary).
3. Tears down the scratch directory at fixture exit per AC-6 (≤2 s).

The fixtures are intentionally **session-scoped per parameter set** —
within one parametrize variant the same injector tree is reused across
multiple test methods so we don't pay the ~3 s build cost per assertion.
"""

from __future__ import annotations

from collections.abc import Iterator
from pathlib import Path

import pytest

from fixtures.injectors import blackout_spoof, multi_segment, outlier
from fixtures.injectors._common import cleanup_tmpfs


# ---------------------------------------------------------------------------
# Source data discovery
# ---------------------------------------------------------------------------


@pytest.fixture(scope="session")
def derkachi_source_frames() -> Path:
    """Path to the AD*.jpg frames the injectors operate on.

    Looks up the project's ``_docs/00_problem/input_data/`` (the test
    container mounts this read-only) and asserts the AD-stills exist.
    """
    # Walk up from this file: e2e/runner/helpers/injector_fixtures.py
    repo_root = Path(__file__).resolve().parents[3]
    candidates = [
        repo_root / "_docs/00_problem/input_data",
        Path("/test-data"),  # docker-compose bind-mount target
    ]
    for c in candidates:
        if (c / "AD000001.jpg").is_file():
            return c
    raise FileNotFoundError(
        "Derkachi source frames not found in any of: "
        + ", ".join(str(c) for c in candidates)
    )


@pytest.fixture(scope="session")
def tile_cache_fixture(pytestconfig: pytest.Config) -> Path:
    """Path to the AZ-407 tile-cache fixture tree.

    Two strategies:

    1. ``--tile-cache-fixture=<path>`` CLI flag (added by tests/fixtures
       that explicitly need to point at a pre-built cache).
    2. Default Docker mount at ``/tile-cache`` inside the runner image.

    Skips the consuming test when the cache is missing — the injector
    unit tests use a synthetic mini-cache (see ``test_outlier.py``) and
    don't need this fixture.
    """
    explicit = pytestconfig.getoption("--tile-cache-fixture", default=None)
    if explicit is not None:
        p = Path(str(explicit))
        if p.is_dir():
            return p
    default = Path("/tile-cache")
    if default.is_dir():
        return default
    pytest.skip("tile-cache fixture not available (build with `make fixtures`)")


# ---------------------------------------------------------------------------
# Per-injector fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def outlier_injection_derkachi(
    request: pytest.FixtureRequest,
    derkachi_source_frames: Path,
    tile_cache_fixture: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[outlier.OutlierInjectionReport]:
    """Build the outlier-injection-derkachi fixture for a single test.

    Density is read from the parametrize ID (e.g.
    ``@pytest.mark.parametrize("density", ["medium"], indirect=True)``)
    or defaults to ``"medium"``. Seed defaults to ``0`` — override via
    ``request.param["seed"]`` when a test needs a different stream.
    """
    params = request.param if hasattr(request, "param") else {}
    density = params.get("density", "medium")
    seed = params.get("seed", 0)
    out_root = tmp_path_factory.mktemp(f"outlier-{density}-{seed}")
    report = outlier.build(
        outlier.OutlierInjectionPlan(
            source_frames_dir=derkachi_source_frames,
            tile_cache_dir=tile_cache_fixture,
            density=density,
            seed=seed,
        ),
        out_root,
    )
    yield report
    cleanup_tmpfs(out_root)


@pytest.fixture
def blackout_spoof_derkachi(
    request: pytest.FixtureRequest,
    derkachi_source_frames: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[blackout_spoof.BlackoutSpoofReport]:
    """Build the blackout-spoof-derkachi fixture for a single test."""
    params = request.param if hasattr(request, "param") else {}
    window_seconds = params.get("window_seconds", 15.0)
    seed = params.get("seed", 0)
    out_root = tmp_path_factory.mktemp(f"blackout-spoof-{int(window_seconds)}s-{seed}")
    report = blackout_spoof.build(
        blackout_spoof.BlackoutSpoofPlan(
            source_frames_dir=derkachi_source_frames,
            blackout_seconds=window_seconds,
            seed=seed,
        ),
        out_root,
    )
    yield report
    cleanup_tmpfs(out_root)


@pytest.fixture
def multi_segment_derkachi(
    request: pytest.FixtureRequest,
    derkachi_source_frames: Path,
    tmp_path_factory: pytest.TempPathFactory,
) -> Iterator[multi_segment.MultiSegmentReport]:
    """Build the multi-segment-derkachi fixture for a single test."""
    params = request.param if hasattr(request, "param") else {}
    n_segments = params.get("n_segments", 3)
    segment_seconds = params.get("segment_seconds", 12.0)
    out_root = tmp_path_factory.mktemp(f"multi-segment-{n_segments}x{int(segment_seconds)}s")
    report = multi_segment.build(
        multi_segment.MultiSegmentPlan(
            source_frames_dir=derkachi_source_frames,
            n_segments=n_segments,
            segment_seconds=segment_seconds,
        ),
        out_root,
    )
    yield report
    cleanup_tmpfs(out_root)


# ---------------------------------------------------------------------------
# Tile-cache CLI flag registration
# ---------------------------------------------------------------------------


def pytest_addoption(parser: pytest.Parser) -> None:
    """Register the ``--tile-cache-fixture`` flag at plugin load time.

    Imported by the runner's ``conftest.py`` via ``pytest_plugins`` so it
    runs once per session before fixture resolution.
    """
    group = parser.getgroup("e2e-runner")
    group.addoption(
        "--tile-cache-fixture",
        action="store",
        default=None,
        help="Path to a pre-built tile-cache fixture tree. Default: /tile-cache (Docker mount).",
    )