Files
gps-denied-onboard/tests/blackbox/test_infrastructure.py
T
Oleksandr Bezdieniezhnykh cab7b5d020 [AZ-233] Update Docker Compose and enhance test documentation
- Modified the Docker Compose configuration to include an input root for replay tests and added an environment variable for enabling SITL.
- Enhanced documentation for various testing processes, including the addition of a Runtime Completeness Decomposition Gate and clarifications on internal module testing requirements.
- Updated the implementation completeness report to reflect the current state and added new test cases for performance and resilience scenarios.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-06 05:03:48 +03:00

97 lines
3.3 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"}
security_row = next(row for row in rows if row["Test ID"] == "NFT-SEC-INFRA")
assert security_row["Result"] == "pass"
assert security_row["Source Label"] == "untrusted_cache_rejected"
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_uses_configurable_input_root_for_replay_fixtures(tmp_path: Path) -> None:
# Arrange
input_root = tmp_path / "input"
(input_root / "expected_results").mkdir(parents=True)
(input_root / "coordinates.csv").write_text("image,lat,lon\nAD000001.jpg,48.0,37.0\n")
(input_root / "expected_results" / "results_report.md").write_text("# Expected results\n")
# Act
result = BlackboxReplayRunner(output_root=tmp_path / "output", input_root=input_root).run()
# Assert
reports_by_id = {report.scenario_id: report for report in result.reports}
assert reports_by_id["FT-P-01"].result == ScenarioResult.PASS
assert reports_by_id["NFT-PERF-INFRA"].result == ScenarioResult.PASS
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)