mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 08:41:12 +00:00
Oleksandr Bezdieniezhnykh
43fdef1aac
Implement all 11 `sitl_observer` public surfaces as an offline
FDR-replay strategy (reads JSON fixtures under `${E2E_SITL_REPLAY_DIR}`
instead of live pymavlink/yamspy). Replace 12 per-scenario
`_harness_helpers_implemented` probes with one shared session-scoped
`sitl_replay_ready` fixture in `e2e/tests/conftest.py`.
Net: -636 LoC of duplicated scenario gating, +17 LoC shared fixture,
+38 new unit tests (596 total, up from 558). Includes K=3 cumulative
review for batches 73-75 (PASS).
Co-authored-by: Cursor <cursoragent@cursor.com>
432 lines
12 KiB
Python
432 lines
12 KiB
Python
"""Unit tests for `e2e/runner/helpers/sitl_observer.py` (AZ-595)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from e2e.runner.helpers import sitl_observer as so
|
|
|
|
|
|
@pytest.fixture
|
|
def replay_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
|
"""Sets `${E2E_SITL_REPLAY_DIR}` to a tmp dir for the duration of the test."""
|
|
monkeypatch.setenv("E2E_SITL_REPLAY_DIR", str(tmp_path))
|
|
return tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def unset_replay_dir(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("E2E_SITL_REPLAY_DIR", raising=False)
|
|
|
|
|
|
def _write_json(path: Path, content) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(content))
|
|
|
|
|
|
# replay_dir / replay_dir_available
|
|
|
|
|
|
def test_replay_dir_unset_returns_none(unset_replay_dir):
|
|
# Assert
|
|
assert so.replay_dir() is None
|
|
assert so.replay_dir_available() is False
|
|
|
|
|
|
def test_replay_dir_set_but_missing_returns_false(monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
|
|
# Arrange
|
|
monkeypatch.setenv("E2E_SITL_REPLAY_DIR", str(tmp_path / "nope"))
|
|
|
|
# Assert
|
|
assert so.replay_dir_available() is False
|
|
|
|
|
|
def test_replay_dir_set_and_exists_returns_true(replay_dir: Path):
|
|
# Assert
|
|
assert so.replay_dir_available() is True
|
|
|
|
|
|
def test_replay_dir_whitespace_env_treated_as_unset(monkeypatch: pytest.MonkeyPatch):
|
|
# Arrange
|
|
monkeypatch.setenv("E2E_SITL_REPLAY_DIR", " ")
|
|
|
|
# Assert
|
|
assert so.replay_dir() is None
|
|
|
|
|
|
# read_ekf_divergence_events
|
|
|
|
|
|
def test_read_ekf_events_empty_without_env(unset_replay_dir):
|
|
# Assert
|
|
assert so.read_ekf_divergence_events() == []
|
|
|
|
|
|
def test_read_ekf_events_empty_when_file_missing(replay_dir: Path):
|
|
# Assert
|
|
assert so.read_ekf_divergence_events() == []
|
|
|
|
|
|
def test_read_ekf_events_parses_records(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "ekf_divergence_events.json",
|
|
[
|
|
{"monotonic_ms": 1000, "severity": "WARNING", "message": "EKF_X drift"},
|
|
{"monotonic_ms": 2000, "severity": "CRITICAL", "message": "EKF_X reset"},
|
|
],
|
|
)
|
|
|
|
# Act
|
|
events = so.read_ekf_divergence_events()
|
|
|
|
# Assert
|
|
assert len(events) == 2
|
|
assert events[0] == so.EkfDivergenceEvent(
|
|
monotonic_ms=1000, severity="WARNING", message="EKF_X drift"
|
|
)
|
|
|
|
|
|
def test_read_ekf_events_malformed_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(replay_dir / "ekf_divergence_events.json", [{"monotonic_ms": "bad"}])
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="EKF divergence fixture malformed"):
|
|
so.read_ekf_divergence_events()
|
|
|
|
|
|
def test_read_ekf_events_wrong_top_level_type_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(replay_dir / "ekf_divergence_events.json", {"not": "a list"})
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="must be a JSON list"):
|
|
so.read_ekf_divergence_events()
|
|
|
|
|
|
# read_gps_health_samples
|
|
|
|
|
|
def test_read_gps_health_parses(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "gps_health_samples.json",
|
|
[
|
|
{"monotonic_ms": 0, "healthy": True, "spoofed": False},
|
|
{"monotonic_ms": 1000, "healthy": False, "spoofed": True},
|
|
],
|
|
)
|
|
|
|
# Act
|
|
samples = so.read_gps_health_samples()
|
|
|
|
# Assert
|
|
assert len(samples) == 2
|
|
assert samples[1].spoofed is True
|
|
|
|
|
|
def test_read_gps_health_empty_without_env(unset_replay_dir):
|
|
# Assert
|
|
assert so.read_gps_health_samples() == []
|
|
|
|
|
|
# read_consistency_check_events
|
|
|
|
|
|
def test_read_consistency_check_parses(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "consistency_check_events.json",
|
|
[{"monotonic_ms": 5000, "passed": True}],
|
|
)
|
|
|
|
# Act
|
|
events = so.read_consistency_check_events()
|
|
|
|
# Assert
|
|
assert events == [so.ConsistencyCheckEvent(monotonic_ms=5000, passed=True)]
|
|
|
|
|
|
def test_read_consistency_check_empty_without_env(unset_replay_dir):
|
|
# Assert
|
|
assert so.read_consistency_check_events() == []
|
|
|
|
|
|
# get_observer
|
|
|
|
|
|
def test_get_observer_missing_env_raises(unset_replay_dir):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="env var not set"):
|
|
so.get_observer("ardupilot", "sitl-host")
|
|
|
|
|
|
def test_get_observer_missing_fixture_raises(replay_dir: Path):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="required fixture not found"):
|
|
so.get_observer("ardupilot", "sitl-host")
|
|
|
|
|
|
def test_get_observer_read_gps_state(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "observer_ardupilot_sitl-host.json",
|
|
{
|
|
"gps_state": {
|
|
"primary_source": "MAV",
|
|
"last_position_lat_deg": 50.0,
|
|
"last_position_lon_deg": 30.0,
|
|
"last_position_alt_m": 250.0,
|
|
"fix_quality": 3,
|
|
"horizontal_accuracy_m": 1.5,
|
|
"last_update_age_ms": 100,
|
|
},
|
|
"parameters": {"EK3_SRC1_POSXY": 3},
|
|
},
|
|
)
|
|
|
|
# Act
|
|
obs = so.get_observer("ardupilot", "sitl-host")
|
|
gps = obs.read_gps_state()
|
|
|
|
# Assert
|
|
assert gps.primary_source == "MAV"
|
|
assert gps.fix_quality == 3
|
|
assert obs.read_parameter("EK3_SRC1_POSXY") == 3
|
|
assert obs.read_parameter("MISSING") is None
|
|
|
|
|
|
def test_get_observer_missing_gps_state_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(replay_dir / "observer_inav_h.json", {"parameters": {}})
|
|
|
|
# Act / Assert
|
|
obs = so.get_observer("inav", "h")
|
|
with pytest.raises(RuntimeError, match="fixture missing `gps_state`"):
|
|
obs.read_gps_state()
|
|
|
|
|
|
# prepare_sitl_*
|
|
|
|
|
|
def test_prepare_sitl_cold_boot_no_op(tmp_path: Path):
|
|
# Act — no env var set is fine for the no-op.
|
|
so.prepare_sitl_cold_boot(host="ardupilot-sitl", fixture_path=tmp_path / "cb.json")
|
|
|
|
|
|
def test_prepare_sitl_cold_boot_empty_host_raises(tmp_path: Path):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="host must be non-empty"):
|
|
so.prepare_sitl_cold_boot(host="", fixture_path=tmp_path / "cb.json")
|
|
|
|
|
|
def test_prepare_sitl_cold_boot_none_fixture_path_raises():
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="fixture_path is required"):
|
|
so.prepare_sitl_cold_boot(host="ardupilot-sitl", fixture_path=None) # type: ignore[arg-type]
|
|
|
|
|
|
def test_prepare_sitl_no_gps_no_op():
|
|
# Act
|
|
so.prepare_sitl_no_gps(host="ardupilot-sitl")
|
|
|
|
|
|
def test_prepare_sitl_no_gps_empty_host_raises():
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="host must be non-empty"):
|
|
so.prepare_sitl_no_gps(host="")
|
|
|
|
|
|
# capture_ap_tlog
|
|
|
|
|
|
def test_capture_ap_tlog_missing_env_raises(unset_replay_dir):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="env var not set"):
|
|
so.capture_ap_tlog(host="ardupilot-sitl", duration_s=1.0)
|
|
|
|
|
|
def test_capture_ap_tlog_missing_file_raises(replay_dir: Path):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="fixture not found"):
|
|
so.capture_ap_tlog(host="ardupilot-sitl", duration_s=1.0)
|
|
|
|
|
|
def test_capture_ap_tlog_returns_path(replay_dir: Path):
|
|
# Arrange
|
|
tlog = replay_dir / "ap_tlog_ardupilot-sitl.tlog"
|
|
tlog.write_bytes(b"\x00\x01\x02")
|
|
|
|
# Act
|
|
out = so.capture_ap_tlog(host="ardupilot-sitl", duration_s=1.0)
|
|
|
|
# Assert
|
|
assert out == tlog
|
|
|
|
|
|
def test_capture_ap_tlog_zero_duration_raises():
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="duration_s must be positive"):
|
|
so.capture_ap_tlog(host="x", duration_s=0)
|
|
|
|
|
|
# read_ap_parameter
|
|
|
|
|
|
def test_read_ap_parameter_returns_value(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "ap_parameters_ardupilot-sitl.json",
|
|
{"EK3_SRC1_POSXY": 3, "GPS_TYPE": 14},
|
|
)
|
|
|
|
# Act + Assert
|
|
assert so.read_ap_parameter(host="ardupilot-sitl", name="EK3_SRC1_POSXY") == 3
|
|
assert so.read_ap_parameter(host="ardupilot-sitl", name="UNKNOWN") is None
|
|
|
|
|
|
def test_read_ap_parameter_missing_file_raises(replay_dir: Path):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="required fixture not found"):
|
|
so.read_ap_parameter(host="ardupilot-sitl", name="ANY")
|
|
|
|
|
|
# observe_inav_tcp_handshake
|
|
|
|
|
|
def test_observe_inav_tcp_handshake_returns_record(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_handshake_inav-sitl_5760.json",
|
|
{"established_within_s": 2.1},
|
|
)
|
|
|
|
# Act
|
|
report = so.observe_inav_tcp_handshake(host="inav-sitl", port=5760, timeout_s=5.0)
|
|
|
|
# Assert
|
|
assert report.established_within_s == pytest.approx(2.1)
|
|
|
|
|
|
def test_observe_inav_tcp_handshake_null_established(replay_dir: Path):
|
|
# Arrange — handshake did NOT establish within window.
|
|
_write_json(
|
|
replay_dir / "inav_handshake_inav-sitl_5760.json",
|
|
{"established_within_s": None},
|
|
)
|
|
|
|
# Act
|
|
report = so.observe_inav_tcp_handshake(host="inav-sitl", port=5760, timeout_s=5.0)
|
|
|
|
# Assert
|
|
assert report.established_within_s is None
|
|
|
|
|
|
def test_observe_inav_tcp_handshake_zero_timeout_raises():
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="timeout_s must be positive"):
|
|
so.observe_inav_tcp_handshake(host="x", port=1, timeout_s=0)
|
|
|
|
|
|
def test_observe_inav_tcp_handshake_bad_value_type_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_handshake_inav-sitl_5760.json",
|
|
{"established_within_s": "not-a-number"},
|
|
)
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="must be a number or null"):
|
|
so.observe_inav_tcp_handshake(host="inav-sitl", port=5760, timeout_s=5.0)
|
|
|
|
|
|
# collect_inav_msp_frames
|
|
|
|
|
|
def test_collect_inav_msp_frames_round_trip(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_msp_frames_inav-sitl_5760.json",
|
|
{
|
|
"frames": [
|
|
{"monotonic_ms": 0, "function_id": 0x1F03},
|
|
{"monotonic_ms": 200, "function_id": 0x1F03},
|
|
],
|
|
"expected_num_sat": 12,
|
|
},
|
|
)
|
|
|
|
# Act
|
|
capture = so.collect_inav_msp_frames(host="inav-sitl", port=5760, window_s=60.0)
|
|
|
|
# Assert
|
|
assert capture.expected_num_sat == 12
|
|
assert len(capture.frames) == 2
|
|
assert capture.frames[1].function_id == 0x1F03
|
|
|
|
|
|
def test_collect_inav_msp_frames_missing_expected_num_sat_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_msp_frames_inav-sitl_5760.json",
|
|
{"frames": []},
|
|
)
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="`expected_num_sat` must be an int"):
|
|
so.collect_inav_msp_frames(host="inav-sitl", port=5760, window_s=60.0)
|
|
|
|
|
|
def test_collect_inav_msp_frames_malformed_frame_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_msp_frames_inav-sitl_5760.json",
|
|
{"frames": [{"monotonic_ms": "bad"}], "expected_num_sat": 12},
|
|
)
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="malformed frame"):
|
|
so.collect_inav_msp_frames(host="inav-sitl", port=5760, window_s=60.0)
|
|
|
|
|
|
# query_inav_gps_state
|
|
|
|
|
|
def test_query_inav_gps_state_round_trip(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_gps_state_inav-sitl.json",
|
|
{"fix_type": 3, "num_sat": 14, "provider": "MSP"},
|
|
)
|
|
|
|
# Act
|
|
state = so.query_inav_gps_state(host="inav-sitl")
|
|
|
|
# Assert
|
|
assert state.fix_type == 3
|
|
assert state.num_sat == 14
|
|
assert state.provider == "MSP"
|
|
|
|
|
|
def test_query_inav_gps_state_missing_field_raises(replay_dir: Path):
|
|
# Arrange
|
|
_write_json(
|
|
replay_dir / "inav_gps_state_inav-sitl.json",
|
|
{"fix_type": 3, "num_sat": 14},
|
|
)
|
|
|
|
# Act / Assert
|
|
with pytest.raises(RuntimeError, match="iNav GPS state fixture"):
|
|
so.query_inav_gps_state(host="inav-sitl")
|
|
|
|
|
|
def test_query_inav_gps_state_missing_env_raises(unset_replay_dir):
|
|
# Assert
|
|
with pytest.raises(RuntimeError, match="env var not set"):
|
|
so.query_inav_gps_state(host="inav-sitl")
|