mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 05:41:13 +00:00
Oleksandr Bezdieniezhnykh
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import csv
|
|
from pathlib import Path
|
|
|
|
from e2e.replay.harness import (
|
|
REPORT_COLUMNS,
|
|
BlackboxReplayRunner,
|
|
SatelliteCacheStub,
|
|
ScenarioConfig,
|
|
ScenarioGroup,
|
|
ScenarioResult,
|
|
)
|
|
|
|
|
|
def test_replay_environment_reports_missing_prerequisites_as_blocked(tmp_path: Path) -> None:
|
|
# Arrange
|
|
scenario = ScenarioConfig(
|
|
scenario_id="BLOCKED-INFRA",
|
|
name="Blocked prerequisite smoke",
|
|
group=ScenarioGroup.RESILIENCE,
|
|
input_dataset="sitl_spoofing_scenarios",
|
|
required_paths=(tmp_path / "missing-fixture.csv",),
|
|
required_services=("sitl",),
|
|
)
|
|
|
|
# Act
|
|
result = BlackboxReplayRunner(output_root=tmp_path, scenarios=(scenario,)).run()
|
|
|
|
# Assert
|
|
report = result.reports[0]
|
|
assert report.result == ScenarioResult.BLOCKED
|
|
assert "missing fixture path" in report.error_message
|
|
assert "SITL prerequisite blocked" in report.error_message
|
|
|
|
|
|
def test_satellite_cache_stub_is_deterministic_and_records_interactions() -> None:
|
|
# Arrange
|
|
stub = SatelliteCacheStub()
|
|
|
|
# Act
|
|
first = stub.query_manifest("FT-P-01", "valid")
|
|
second = stub.query_manifest("FT-P-01", "valid")
|
|
|
|
# Assert
|
|
assert first == second
|
|
assert first["network_fetch_attempted"] is False
|
|
assert len(stub.interactions) == 2
|
|
assert stub.interactions[0].service == "satellite-cache-stub"
|
|
|
|
|
|
def test_runner_executes_all_required_groups_and_writes_reports(tmp_path: Path) -> None:
|
|
# Act
|
|
result = BlackboxReplayRunner(output_root=tmp_path).run()
|
|
|
|
# Assert
|
|
assert result.completed_groups == set(ScenarioGroup)
|
|
rows = list(csv.DictReader(result.csv_path.open(encoding="utf-8")))
|
|
assert rows
|
|
assert rows[0].keys() == set(REPORT_COLUMNS)
|
|
assert {row["Result"] for row in rows} <= {"pass", "blocked"}
|
|
|
|
markdown = result.markdown_path.read_text(encoding="utf-8")
|
|
assert "FDR Validation Summary" in markdown
|
|
assert "SITL prerequisite blocked" in markdown
|
|
assert "Jetson prerequisite blocked" in markdown
|
|
|
|
|
|
def test_runner_keeps_generated_artifacts_run_scoped(tmp_path: Path) -> None:
|
|
# Act
|
|
result = BlackboxReplayRunner(output_root=tmp_path).run()
|
|
|
|
# Assert
|
|
assert result.run_dir.parent == tmp_path
|
|
assert result.csv_path.parent == result.run_dir
|
|
assert result.markdown_path.parent == result.run_dir
|
|
for report in result.reports:
|
|
assert report.artifacts
|
|
assert all(artifact.parent.parent == result.run_dir for artifact in report.artifacts)
|