azaion/gps-denied-onboard
1
0
Fork 0
mirror of https://github.com/azaion/gps-denied-onboard.git synced 2026-06-21 21:11:12 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
e367b07e3b0ff3f9443070efabdce04b3900c6fa
gps-denied-onboard/tests/unit/test_az402_replay_cli.py
T
Oleksandr Bezdieniezhnykh 007aa36fbf [AZ-895] Deprecate replay auto-sync surface; file AZ-908 follow-up 
Option A (minimum-deprecation, 2 SP) per user complexity-budget
decision. Auto-sync stays importable as a raising stub for one cycle
so external callers see a clean ReplayInputAdapterError instead of an
ImportError. Full physical removal is filed as AZ-908 (cycle-5+ backlog).

Production:
- auto_sync.py: 700+ LOC -> 56-line no-op stub raising
  "auto-sync removed; supply --imu CSV instead"
- tlog_video_adapter.py: 700+ LOC -> 105-line deprecated stub;
  ReplayInputAdapter.open() raises immediately, close() is a no-op
- _replay_branch.py: dropped legacy auto-sync branch +
  _build_auto_sync_config; _validate_replay_paths now requires
  imu_csv_path; replay_input_adapter_factory parameter removed
- cli/replay.py: --time-offset-ms / --skip-auto-sync / --auto-trim
  emit DeprecationWarning + stderr line; values ignored
- tlog_replay_adapter.py + tlog_ground_truth.py docstrings: AUDIT-ONLY

Tests:
- DELETED test_az405_auto_sync, test_az405_replay_input_adapter,
  test_az698_window_alignment (covered code no longer runs)
- ADDED test_az895_auto_sync_deprecated_stub (5 parametrised, pins AC-1)
- test_az402_replay_cli: deprecation warnings + ignored-value asserts
- test_az401_compose_root_replay: new imu_csv_path-required gate;
  deleted the calibration-loading test that relied on the removed
  replay_input_adapter_factory injection point
- test_derkachi_real_tlog: xfail reason refreshed to AZ-848 + AZ-883
  (AC-4 "AZ-848-scoped reason")

Docs:
- module-layout.md: replay_input file list flags deprecated modules,
  adds csv_ground_truth.py
- _dependencies_table.md: +AZ-908 row, preamble + totals updated
  (179 -> 180 tasks, 567 -> 570 SP)
- AZ-908 backlog spec added; AZ-895 spec moved todo -> done
- batch_03_cycle4_report.md written

Touched-module tests green (111 passed, 1 skipped). Full unit suite
green: 2287 passed, 85 skipped, 1 deselected (pre-existing flaky perf
test, unrelated).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-26 22:09:59 +03:00

652 lines
19 KiB
Python
Raw Blame History

"""AZ-402 — `gps-denied-replay` console-script unit tests.
Covers AC-1..AC-10 of the AZ-402 task spec. AC-10 (console-script
registered) ships as both a static pyproject.toml assertion and a
subprocess smoke test gated on the package being installed.
Implements ``_docs/02_document/contracts/replay/replay_protocol.md``
v2.0.0 — CLI surface + Invariant 11.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from collections.abc import Iterator
from pathlib import Path
from typing import Any
from unittest import mock
import pytest
import numpy as np
from gps_denied_onboard.cli import replay as replay_cli
from gps_denied_onboard.cli.replay import (
EXIT_GENERIC_FAILURE,
EXIT_SUCCESS,
EXIT_SYNC_IMPOSSIBLE,
ReplayCliError,
)
from gps_denied_onboard.config import Config
from gps_denied_onboard.replay_input import ReplayInputAdapterError
# ----------------------------------------------------------------------
# Fixtures
@pytest.fixture
def _calib_payload() -> dict[str, Any]:
return {
"camera_id": "test-cam",
"intrinsics_3x3": np.eye(3).tolist(),
"distortion": [0.0, 0.0, 0.0, 0.0],
"body_to_camera_se3": np.eye(4).tolist(),
"acquisition_method": "operator",
"metadata": {},
}
@pytest.fixture
def _required_files(tmp_path: Path, _calib_payload: dict[str, Any]) -> dict[str, Path]:
"""Create real on-disk files for every required CLI arg."""
video = tmp_path / "video.mp4"
video.write_bytes(b"\x00\x00\x00\x18ftypmp42") # placeholder
tlog = tmp_path / "flight.tlog"
tlog.write_bytes(b"\x00")
imu_csv = tmp_path / "data_imu.csv"
# Minimal placeholder — the CLI only validates existence, parsing
# happens later inside the runtime loop.
imu_csv.write_text("placeholder", encoding="utf-8")
output = tmp_path / "out.jsonl"
calib = tmp_path / "calib.json"
calib.write_text(json.dumps(_calib_payload))
config_yaml = tmp_path / "config.yaml"
config_yaml.write_text("# minimal — env supplies the rest\n")
signing_key = tmp_path / "key.bin"
signing_key.write_bytes(b"X" * 32)
return {
"video": video,
"tlog": tlog,
"imu": imu_csv,
"output": output,
"camera_calibration": calib,
"config": config_yaml,
"mavlink_signing_key": signing_key,
}
@pytest.fixture
def _airborne_env(monkeypatch: pytest.MonkeyPatch) -> None:
"""Set the env vars `load_config` needs to validate successfully."""
for name, value in (
("GPS_DENIED_FC_PROFILE", "ardupilot_plane"),
("GPS_DENIED_TIER", "1"),
("DB_URL", "postgresql+psycopg://gps_denied:dev@db:5432/gps_denied"),
("CAMERA_CALIBRATION_PATH", "/will/be/overridden/by/cli.json"),
("LOG_LEVEL", "INFO"),
("LOG_SINK", "console"),
("INFERENCE_BACKEND", "pytorch_fp16"),
("FDR_PATH", "/var/lib/gps-denied/fdr"),
("TILE_CACHE_PATH", "/var/lib/gps-denied/tiles"),
):
monkeypatch.setenv(name, value)
def _argv(files: dict[str, Path], **overrides: Any) -> list[str]:
"""Build a CLI argv from the required-files fixture + overrides."""
base = {
"--video": str(files["video"]),
"--imu": str(files["imu"]),
"--tlog": str(files["tlog"]),
"--output": str(files["output"]),
"--camera-calibration": str(files["camera_calibration"]),
"--config": str(files["config"]),
"--mavlink-signing-key": str(files["mavlink_signing_key"]),
}
if "pace" in overrides:
base["--pace"] = overrides["pace"]
if "time_offset_ms" in overrides and overrides["time_offset_ms"] is not None:
base["--time-offset-ms"] = str(overrides["time_offset_ms"])
argv: list[str] = []
for k, v in base.items():
argv.extend([k, v])
if overrides.get("skip_auto_sync"):
argv.append("--skip-auto-sync")
if overrides.get("auto_trim"):
argv.append("--auto-trim")
return argv
# ----------------------------------------------------------------------
# AC-1: All required args parsed
def test_ac1_all_required_args_parsed(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SUCCESS
cfg = captured["config"]
assert cfg.replay.video_path == str(_required_files["video"])
assert cfg.replay.tlog_path == str(_required_files["tlog"])
assert cfg.replay.output_path == str(_required_files["output"])
assert cfg.runtime.camera_calibration_path == str(
_required_files["camera_calibration"]
)
# ----------------------------------------------------------------------
# AC-2: --pace default ASAP
def test_ac2_pace_default_asap(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.pace == "asap"
# ----------------------------------------------------------------------
# AC-3: --pace realtime
def test_ac3_pace_realtime(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(
_argv(_required_files, pace="realtime"), shared_main=fake_main
)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.pace == "realtime"
# ----------------------------------------------------------------------
# AC-4: --time-offset-ms deprecated (AZ-895) — ignored + warning emitted
def test_ac4_time_offset_ignored_after_az895(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
"""AZ-895: --time-offset-ms is deprecated; value is ignored, warning emitted."""
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
with pytest.warns(DeprecationWarning, match="--time-offset-ms"):
rc = replay_cli.main(
_argv(_required_files, time_offset_ms=5000), shared_main=fake_main
)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.time_offset_ms is None
err = capsys.readouterr().err
assert "--time-offset-ms is deprecated (AZ-895)" in err
def test_ac4_time_offset_none_when_absent(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.time_offset_ms is None
def test_az895_skip_auto_sync_ignored_and_warned(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
"""AZ-895: --skip-auto-sync deprecated; value ignored, warning emitted."""
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
with pytest.warns(DeprecationWarning, match="--skip-auto-sync"):
rc = replay_cli.main(
_argv(_required_files, skip_auto_sync=True), shared_main=fake_main
)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.skip_auto_sync_validation is False
err = capsys.readouterr().err
assert "--skip-auto-sync is deprecated (AZ-895)" in err
def test_az895_auto_trim_ignored_and_warned(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
"""AZ-895: --auto-trim deprecated; value ignored, warning emitted."""
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
with pytest.warns(DeprecationWarning, match="--auto-trim"):
rc = replay_cli.main(
_argv(_required_files, auto_trim=True), shared_main=fake_main
)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].replay.auto_trim is False
err = capsys.readouterr().err
assert "--auto-trim is deprecated (AZ-895)" in err
def test_az895_no_deprecated_flags_no_warning(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
recwarn: pytest.WarningsRecorder,
) -> None:
"""No AZ-895 flag-specific deprecation warning when none of the flags are used."""
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_SUCCESS
err = capsys.readouterr().err
for flag in ("--time-offset-ms", "--skip-auto-sync", "--auto-trim"):
assert f"{flag} is deprecated" not in err, (
f"unexpected deprecation banner for {flag} when it was not passed"
)
az895_flag_warnings = [
w for w in recwarn.list
if issubclass(w.category, DeprecationWarning)
and any(
f"{flag} is deprecated" in str(w.message)
for flag in ("--time-offset-ms", "--skip-auto-sync", "--auto-trim")
)
]
assert az895_flag_warnings == []
# ----------------------------------------------------------------------
# AC-5: --mavlink-signing-key required (argparse exit 2)
def test_ac5_missing_signing_key_exits_2(
_required_files: dict[str, Path],
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange — drop the signing-key arg pair from argv
argv = _argv(_required_files)
idx = argv.index("--mavlink-signing-key")
del argv[idx : idx + 2]
# Act / Assert — argparse raises SystemExit(2) on missing required
with pytest.raises(SystemExit) as excinfo:
replay_cli.main(argv, shared_main=lambda _c: 0)
assert excinfo.value.code == 2
err = capsys.readouterr().err
assert "--mavlink-signing-key" in err
def test_ac5_missing_video_exits_2(_required_files: dict[str, Path]) -> None:
# Arrange
argv = _argv(_required_files)
idx = argv.index("--video")
del argv[idx : idx + 2]
# Act / Assert
with pytest.raises(SystemExit) as excinfo:
replay_cli.main(argv, shared_main=lambda _c: 0)
assert excinfo.value.code == 2
# ----------------------------------------------------------------------
# AC-6: Calibration loader rejects malformed JSON
def test_ac6_malformed_calibration_exits_1(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange — corrupt the calib.json
_required_files["camera_calibration"].write_text("{ this is not json")
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "camera-calibration JSON malformed" in err
# ----------------------------------------------------------------------
# AC-7: Calibration schema validation
def test_ac7_missing_intrinsics_key_rejected(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange — write a calib.json missing the intrinsics key
_required_files["camera_calibration"].write_text(
json.dumps(
{
"distortion": [0.0, 0.0, 0.0, 0.0],
"body_to_camera_se3": np.eye(4).tolist(),
}
)
)
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "missing 'intrinsics'" in err
def test_ac7_top_level_not_object_rejected(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange — JSON parses but top level is a list
_required_files["camera_calibration"].write_text(json.dumps([1, 2, 3]))
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "expected JSON object" in err
# ----------------------------------------------------------------------
# AC-8: Mode set to replay
def test_ac8_mode_set_to_replay(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SUCCESS
assert captured["config"].mode == "replay"
# The CLI MUST NOT call compose_root directly (replay protocol
# Invariant 11). The shared_main fake here proves the dispatch
# boundary: if compose_root were called inside the CLI we would
# not reach the fake at all.
# ----------------------------------------------------------------------
# AC-9: Exit-code pass-through
@pytest.mark.parametrize("rc", [0, 1, 2])
def test_ac9_exit_code_pass_through(
_required_files: dict[str, Path],
_airborne_env: None,
rc: int,
) -> None:
# Arrange
def fake_main(_config: Config) -> int:
return rc
# Act
actual = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert actual == rc
def test_ac9_replay_input_adapter_error_maps_to_2(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
"""A `ReplayInputAdapterError` raised by shared_main → exit 2."""
# Arrange
def fake_main(_config: Config) -> int:
raise ReplayInputAdapterError("auto-sync hard-fail: 42% match")
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SYNC_IMPOSSIBLE
err = capsys.readouterr().err
assert "replay sync impossible" in err
assert "42% match" in err
def test_unhandled_exception_exits_1_with_traceback(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange
def fake_main(_config: Config) -> int:
raise ValueError("boom: contrived crash inside compose_root")
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "Traceback" in err
assert "boom: contrived crash" in err
# ----------------------------------------------------------------------
# Sanitised banner
def test_signing_key_redacted_in_startup_banner(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Act
replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
err = capsys.readouterr().err
assert "<redacted>" in err
assert str(_required_files["mavlink_signing_key"]) not in err
# ----------------------------------------------------------------------
# AC-10: Console script registered
def test_ac10_console_script_registered_in_pyproject() -> None:
"""Static check: pyproject.toml registers the console-script."""
# Arrange
repo_root = Path(__file__).resolve().parents[2]
pyproject = repo_root / "pyproject.toml"
# Act
text = pyproject.read_text(encoding="utf-8")
# Assert
assert (
'gps-denied-replay = "gps_denied_onboard.cli.replay:main"' in text
), "console script not registered under [project.scripts]"
def test_ac10_console_script_runs_help() -> None:
"""Subprocess: the `gps-denied-replay` script runs `--help` cleanly.
Skipped if the package is not installed (or the script is not on
PATH); the static assertion in the previous test suffices in that
environment.
"""
# Arrange
import shutil
binary = shutil.which("gps-denied-replay")
if binary is None:
venv_bin = Path(sys.executable).parent / "gps-denied-replay"
if not venv_bin.exists():
pytest.skip("gps-denied-replay console script not on PATH or in venv bin")
binary = str(venv_bin)
# Act
result = subprocess.run(
[binary, "--help"], capture_output=True, text=True, timeout=15
)
# Assert
assert result.returncode == 0, result.stderr
assert "gps-denied-replay" in result.stdout
# Required-arg surface check
for arg in (
"--video",
"--imu",
"--tlog",
"--output",
"--camera-calibration",
"--config",
"--mavlink-signing-key",
):
assert arg in result.stdout, f"{arg} missing from --help output"
# ----------------------------------------------------------------------
# File validation
def test_missing_video_file_exits_1(
_required_files: dict[str, Path],
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange
_required_files["video"].unlink()
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "video" in err
assert "does not exist" in err
def test_signing_key_path_must_be_file_not_dir(
_required_files: dict[str, Path],
tmp_path: Path,
_airborne_env: None,
capsys: pytest.CaptureFixture[str],
) -> None:
# Arrange — pass a directory where a key file is expected
fake_dir = tmp_path / "not_a_key_file"
fake_dir.mkdir()
argv = _argv(_required_files)
idx = argv.index("--mavlink-signing-key")
argv[idx + 1] = str(fake_dir)
# Act
rc = replay_cli.main(argv, shared_main=lambda _c: 0)
# Assert
assert rc == EXIT_GENERIC_FAILURE
err = capsys.readouterr().err
assert "is not a file" in err
# ----------------------------------------------------------------------
# Signing key plumbing
def test_signing_key_propagates_to_dev_static_field(
_required_files: dict[str, Path], _airborne_env: None
) -> None:
# Arrange
captured: dict[str, Config] = {}
def fake_main(config: Config) -> int:
captured["config"] = config
return 0
# Act
rc = replay_cli.main(_argv(_required_files), shared_main=fake_main)
# Assert
assert rc == EXIT_SUCCESS
expected_hex = (b"X" * 32).hex()
assert captured["config"].fc.dev_static_signing_key == expected_hex
assert captured["config"].fc.signing_key_source == "dev_static"
Reference in New Issue View Git Blame Copy Permalink