gps-denied-onboard/tests/conftest.py

"""Shared test fixtures for GPS-denied-onboard test suite."""

from unittest.mock import AsyncMock, MagicMock

import numpy as np
import pytest

from gps_denied.core.coordinates import CoordinateTransformer
from gps_denied.core.models import ModelManager
from gps_denied.schemas import CameraParameters, GPSPoint

# ---------------------------------------------------------------
# Phase 2 / TEST-03: AC traceability plugin
#
# Registers categorical markers (defensive — pyproject.toml is primary), validates
# @pytest.mark.ac() arguments against the canonical AC-ID regex at collection time,
# and (when --ac-dump=<path> is supplied) writes a {ac_id: [test_nodeid, ...]} JSON
# at session end for `scripts/gen_ac_traceability.py` to consume.
#
# See RESEARCH.md §2.1 for the canonical implementation and rationale.
# ---------------------------------------------------------------
import json
import re
from collections import defaultdict
from pathlib import Path

_AC_ID_RE = re.compile(r"^AC-(?:\d+\.\d+[a-z]?|NEW-\d+)$")


def pytest_configure(config):
    """Defensive marker registration. Primary registration lives in pyproject.toml,
    but doing it here too means a future maintainer who drops the pyproject markers
    list does not silently break --strict-markers."""
    for line in (
        "unit: pure-math or single-class test; no I/O",
        "integration: cross-subsystem test; in-memory SQLite / ASGI / full wiring",
        "blackbox: validates external contract without a live producer",
        "sitl: requires ARDUPILOT_SITL_HOST — nightly only",
        "e2e: full-pipeline run against a real dataset — nightly only",
        "ac(ac_id): link test to one or more Acceptance Criteria (e.g. AC-1.1, AC-NEW-3)",
    ):
        config.addinivalue_line("markers", line)


def pytest_addoption(parser):
    parser.addoption(
        "--ac-dump",
        action="store",
        default=None,
        help="Path to write the {ac_id: [test_nodeid, ...]} JSON at session end. "
             "Consumed by scripts/gen_ac_traceability.py.",
    )


def pytest_collection_modifyitems(config, items):
    """Validate @pytest.mark.ac(...) arguments against the canonical AC-ID regex.

    Fail collection (rather than emit a runtime error) so AC-ID typos surface immediately.
    The traceability script's --check mode catches forward orphans (AC without test);
    this hook catches backward orphans (test references non-existent AC) by enforcing
    the syntactic AC-ID shape. Semantic existence (AC ID is declared in the AC doc) is
    enforced by the script in Plan 02-04.
    """
    errors = []
    for item in items:
        for mark in item.iter_markers(name="ac"):
            if not mark.args:
                errors.append(
                    f"{item.nodeid}: @pytest.mark.ac() requires at least one AC ID arg"
                )
                continue
            for arg in mark.args:
                if not isinstance(arg, str) or not _AC_ID_RE.match(arg):
                    errors.append(
                        f"{item.nodeid}: @pytest.mark.ac({arg!r}) — must match "
                        f"AC-X.Y or AC-NEW-N (e.g. 'AC-1.1', 'AC-NEW-3')"
                    )
    if errors:
        raise pytest.UsageError(
            "AC marker validation failed:\n  " + "\n  ".join(errors)
        )


def pytest_sessionfinish(session, exitstatus):
    """Dump {ac_id: [test_nodeid, ...]} to --ac-dump path if supplied.

    Runs even when pytest is invoked with --collect-only (session.items is populated
    before tests execute), so scripts/gen_ac_traceability.py can dump in ~seconds on a
    full suite.
    """
    dump_path = session.config.getoption("--ac-dump", default=None)
    if not dump_path:
        return
    mapping: dict[str, list[str]] = defaultdict(list)
    for item in session.items:
        for mark in item.iter_markers(name="ac"):
            for ac_id in mark.args:
                mapping[ac_id].append(item.nodeid)
    Path(dump_path).write_text(json.dumps(dict(sorted(mapping.items())), indent=2))


# ---------------------------------------------------------------
# Common constants
# ---------------------------------------------------------------

TEST_ORIGIN = GPSPoint(lat=49.0, lon=32.0)
TEST_CAMERA = CameraParameters(
    focal_length=4.5,
    sensor_width=6.17,
    sensor_height=4.55,
    resolution_width=640,
    resolution_height=480,
)


# ---------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------

@pytest.fixture
def mock_repo():
    """Mock FlightRepository."""
    return MagicMock()


@pytest.fixture
def mock_streamer():
    """Mock SSEEventStreamer with async push_event."""
    s = MagicMock()
    s.push_event = AsyncMock()
    return s


@pytest.fixture
def model_manager():
    """Singleton ModelManager (mock engines)."""
    return ModelManager()


@pytest.fixture
def coord_transformer():
    """CoordinateTransformer with a preset test ENU origin."""
    ct = CoordinateTransformer()
    ct.set_enu_origin("test_flight", TEST_ORIGIN)
    return ct


@pytest.fixture
def sample_image():
    """Random 200x200 RGB image for pipeline tests."""
    return np.random.randint(0, 255, (200, 200, 3), dtype=np.uint8)