"""Integration tests for `e2e/fixtures/sitl_replay_builder/build_p01_fixtures.py` (FT-P-01). Strategy-level unit tests for the underlying ``builder.py`` machinery live in ``test_sitl_replay_builder_builder.py``. This file exercises the FT-P-01 scenario composition end-to-end (with all external deps mocked). """ from __future__ import annotations import json import subprocess import types from pathlib import Path from unittest.mock import MagicMock import pytest import e2e.fixtures.sitl_replay_builder.build_p01_fixtures as bp01 def _mk_fake_video_writer() -> MagicMock: w = MagicMock(name="VideoWriter") w.write = MagicMock() w.release = MagicMock() return w def _write_jsonl(path: Path, records: list[dict]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(json.dumps(r) for r in records)) # resolve_p01_image_paths def test_resolve_p01_image_paths_missing_dir_raises(tmp_path: Path): # Assert with pytest.raises(FileNotFoundError, match="input dir not found"): bp01.resolve_p01_image_paths(tmp_path / "missing") def test_resolve_p01_image_paths_sorted(tmp_path: Path): # Arrange for n in (3, 1, 2): (tmp_path / f"AD{n:06d}.jpg").touch() (tmp_path / "ignored.txt").touch() # Act paths = bp01.resolve_p01_image_paths(tmp_path) # Assert — only AD*.jpg, sorted by name assert [p.name for p in paths] == ["AD000001.jpg", "AD000002.jpg", "AD000003.jpg"] # build_p01_fixtures end-to-end def test_build_p01_fixtures_no_images_raises(tmp_path: Path): # Arrange (tmp_path / "empty").mkdir() cfg = bp01.BuilderConfig( input_dir=tmp_path / "empty", output_dir=tmp_path / "out", fc_kind="ardupilot", host="sitl-host", ) # Assert with pytest.raises(FileNotFoundError, match="no AD\\?\\?\\?\\?\\?\\?.jpg images"): bp01.build_p01_fixtures(cfg) def test_build_p01_fixtures_end_to_end_with_mocks(tmp_path: Path): # Arrange — 3 fake AD000NN.jpg files, mocked OpenCV / pymavlink / subprocess input_dir = tmp_path / "in" output_dir = tmp_path / "out" input_dir.mkdir() for n in range(1, 4): (input_dir / f"AD{n:06d}.jpg").touch() writer = _mk_fake_video_writer() frame = types.SimpleNamespace(shape=(480, 640, 3)) mav_writer = MagicMock(write=MagicMock(), close=MagicMock()) def fake_runner(cmd): fdr_path = Path(cmd[cmd.index("--fdr-out") + 1]) _write_jsonl(fdr_path, [ {"kind": "outbound_position_estimate", "payload": {"lat_deg": 1.0, "lon_deg": 2.0}}, {"kind": "outbound_position_estimate", "payload": {"lat_deg": 3.0, "lon_deg": 4.0}}, {"kind": "outbound_position_estimate", "payload": {"lat_deg": 5.0, "lon_deg": 6.0}}, ]) return subprocess.CompletedProcess(cmd, 0) cfg = bp01.BuilderConfig( input_dir=input_dir, output_dir=output_dir, fc_kind="ardupilot", host="sitl-host", ) # Act result_dir = bp01.build_p01_fixtures( cfg, _runner=fake_runner, _video_writer_factory=lambda out, w, h: writer, _imread=lambda p: frame, _mavlink_writer_factory=lambda out: mav_writer, ) # Assert assert result_dir == output_dir outbound_payload = json.loads((output_dir / "outbound_messages_ardupilot_sitl-host.json").read_text()) assert outbound_payload == { "messages": [ {"image_id": "AD000001.jpg", "lat_deg": 1.0, "lon_deg": 2.0}, {"image_id": "AD000002.jpg", "lat_deg": 3.0, "lon_deg": 4.0}, {"image_id": "AD000003.jpg", "lat_deg": 5.0, "lon_deg": 6.0}, ] } assert (output_dir / "observer_ardupilot_sitl-host.json").is_file() def test_build_p01_fixtures_fewer_estimates_than_frames_pads_nulls(tmp_path: Path): # Arrange — 3 frames, FDR yields 1 estimate; expect 2 null entries input_dir = tmp_path / "in" output_dir = tmp_path / "out" input_dir.mkdir() for n in range(1, 4): (input_dir / f"AD{n:06d}.jpg").touch() def fake_runner(cmd): fdr_path = Path(cmd[cmd.index("--fdr-out") + 1]) _write_jsonl(fdr_path, [ {"kind": "outbound_position_estimate", "payload": {"lat_deg": 1.0, "lon_deg": 2.0}}, ]) return subprocess.CompletedProcess(cmd, 0) cfg = bp01.BuilderConfig( input_dir=input_dir, output_dir=output_dir, fc_kind="ardupilot", host="sitl-host", ) # Act bp01.build_p01_fixtures( cfg, _runner=fake_runner, _video_writer_factory=lambda out, w, h: _mk_fake_video_writer(), _imread=lambda p: types.SimpleNamespace(shape=(480, 640, 3)), _mavlink_writer_factory=lambda out: MagicMock(write=MagicMock(), close=MagicMock()), ) # Assert payload = json.loads((output_dir / "outbound_messages_ardupilot_sitl-host.json").read_text()) assert payload["messages"][0]["lat_deg"] == 1.0 assert payload["messages"][1] is None assert payload["messages"][2] is None def test_build_p01_fixtures_more_estimates_than_frames_truncates(tmp_path: Path, caplog): # Arrange — 2 frames, FDR yields 4 estimates; expect 2 retained + WARN input_dir = tmp_path / "in" output_dir = tmp_path / "out" input_dir.mkdir() for n in range(1, 3): (input_dir / f"AD{n:06d}.jpg").touch() def fake_runner(cmd): fdr_path = Path(cmd[cmd.index("--fdr-out") + 1]) _write_jsonl(fdr_path, [ {"kind": "outbound_position_estimate", "payload": {"lat_deg": float(i), "lon_deg": float(i)}} for i in range(4) ]) return subprocess.CompletedProcess(cmd, 0) cfg = bp01.BuilderConfig( input_dir=input_dir, output_dir=output_dir, fc_kind="ardupilot", host="sitl-host", ) # Act with caplog.at_level("WARNING"): bp01.build_p01_fixtures( cfg, _runner=fake_runner, _video_writer_factory=lambda out, w, h: _mk_fake_video_writer(), _imread=lambda p: types.SimpleNamespace(shape=(480, 640, 3)), _mavlink_writer_factory=lambda out: MagicMock(write=MagicMock(), close=MagicMock()), ) # Assert payload = json.loads((output_dir / "outbound_messages_ardupilot_sitl-host.json").read_text()) assert len(payload["messages"]) == 2 assert any("truncating" in rec.message for rec in caplog.records)