mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 21:11:13 +00:00
[AZ-223] [AZ-224] [AZ-225] [AZ-227] Add runtime gateways
Implement the first runtime component boundaries around the shared contracts so downstream batches can consume typed frame, MAVLink, tile, and FDR behavior with focused tests and batch evidence. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
from shared.contracts import FdrEvent
|
||||
|
||||
from fdr_observability import FdrExportRequest, FdrPayload, InMemoryFlightRecorder
|
||||
|
||||
|
||||
def _event(event_type: str = "anchor") -> FdrEvent:
|
||||
return FdrEvent(
|
||||
event_type=event_type,
|
||||
timestamp_ns=1_000,
|
||||
component="anchor_verification",
|
||||
severity="info",
|
||||
payload_ref="pending",
|
||||
mission_id="mission-1",
|
||||
run_id="run-1",
|
||||
)
|
||||
|
||||
|
||||
def test_valid_event_append_indexes_metadata_and_payload_reference() -> None:
|
||||
# Arrange
|
||||
recorder = InMemoryFlightRecorder(segment_limit_bytes=1_000, storage_limit_bytes=2_000)
|
||||
payload = FdrPayload(ref="fdr://segments/1/payloads/anchor-1.cbor", size_bytes=128)
|
||||
|
||||
# Act
|
||||
result = recorder.append_event(_event(), payload)
|
||||
|
||||
# Assert
|
||||
assert result.appended is True
|
||||
assert result.event is not None
|
||||
assert result.event.payload_ref == payload.ref
|
||||
assert result.segment_id == "segment-0001"
|
||||
assert recorder.health.status == "ready"
|
||||
|
||||
|
||||
def test_rollover_threshold_records_explicit_rollover_result() -> None:
|
||||
# Arrange
|
||||
recorder = InMemoryFlightRecorder(segment_limit_bytes=100, storage_limit_bytes=500)
|
||||
recorder.append_event(_event("first"), FdrPayload(ref="fdr://payloads/1", size_bytes=80))
|
||||
|
||||
# Act
|
||||
result = recorder.append_event(
|
||||
_event("second"), FdrPayload(ref="fdr://payloads/2", size_bytes=50)
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert result.appended is True
|
||||
assert result.rollover is True
|
||||
assert result.segment_id == "segment-0002"
|
||||
|
||||
|
||||
def test_export_request_produces_queryable_evidence_artifacts() -> None:
|
||||
# Arrange
|
||||
recorder = InMemoryFlightRecorder(segment_limit_bytes=1_000, storage_limit_bytes=2_000)
|
||||
recorder.append_event(_event(), FdrPayload(ref="fdr://payloads/1", size_bytes=128))
|
||||
|
||||
# Act
|
||||
result = recorder.export(
|
||||
FdrExportRequest(mission_id="mission-1", run_id="run-1", include_analytics=True)
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert result.produced is True
|
||||
assert result.evidence_ref == "fdr://exports/mission-1/run-1/evidence.json"
|
||||
assert result.analytics_ref == "fdr://exports/mission-1/run-1/analytics.parquet"
|
||||
assert result.segments[0].event_count == 1
|
||||
Reference in New Issue
Block a user