[AZ-221] [AZ-222] Add shared runtime helpers

Provide deterministic geometry/time-sync helpers and structured config, error, health, and telemetry primitives for downstream runtime components.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-03 14:01:04 +03:00
parent 5156453224
commit c3650d979d
17 changed files with 495 additions and 1 deletions
@@ -0,0 +1,65 @@
from shared.config import validate_runtime_profile
from shared.errors import ErrorEnvelope, ResultEnvelope
from shared.telemetry import HealthEvent, MetricsLabels
def test_missing_production_cache_dir_returns_readiness_failure() -> None:
# Arrange
payload = {
"environment": "production",
"config_dir": "/etc/gps-denied-onboard",
"fdr_dir": "/var/lib/gps-denied/fdr",
"database_url": "postgresql://localhost/gpsd",
"mavlink_url": "serial:/dev/ttyTHS1:921600",
"camera_source": "hardware",
"signing_key_ref": "secret-ref",
}
# Act
result = validate_runtime_profile("runtime", payload)
# Assert
assert result.ok is False
assert result.error is not None
assert result.error.component == "runtime"
assert result.error.category == "configuration"
assert result.error.severity == "critical"
assert result.error.retryable is False
def test_dependency_error_envelope_has_required_structured_fields() -> None:
# Act
result = ResultEnvelope.failure(
ErrorEnvelope(
component="tile_manager",
category="dependency",
message="postgis unavailable",
severity="error",
retryable=True,
cause="connection refused",
)
)
# Assert
assert result.ok is False
assert result.error is not None
assert result.error.component == "tile_manager"
assert result.error.category == "dependency"
assert result.error.severity == "error"
assert result.error.retryable is True
def test_health_event_and_metrics_labels_are_fdr_safe_metadata() -> None:
# Act
health = HealthEvent(
component="runtime",
timestamp_ns=1,
liveness="alive",
readiness="ready",
dependency_state={"postgis": "ready"},
)
labels = MetricsLabels(component="runtime", action="startup", status="ok")
# Assert
assert health.dependency_state["postgis"] == "ready"
assert labels.status == "ok"
@@ -0,0 +1,41 @@
from shared.geo_geometry import Wgs84Coordinate, distance_m, local_to_wgs84, wgs84_to_local
from shared.time_sync import check_monotonic_timestamps, select_time_window
def test_wgs84_local_round_trip_is_deterministic() -> None:
# Arrange
origin = Wgs84Coordinate(latitude_deg=49.9808, longitude_deg=36.2527, altitude_m=120.0)
point = Wgs84Coordinate(latitude_deg=49.9811, longitude_deg=36.2531, altitude_m=118.0)
# Act
local = wgs84_to_local(origin, point)
round_trip = local_to_wgs84(origin, local)
# Assert
assert round(round_trip.latitude_deg, 7) == round(point.latitude_deg, 7)
assert round(round_trip.longitude_deg, 7) == round(point.longitude_deg, 7)
assert round(round_trip.altitude_m, 7) == round(point.altitude_m, 7)
assert distance_m(origin, point) > 0.0
def test_non_monotonic_timestamps_return_explicit_violation() -> None:
# Act
violations = check_monotonic_timestamps([100, 200, 150])
# Assert
assert len(violations) == 1
assert violations[0].category == "timestamp_mismatch"
def test_time_window_reports_gap_instead_of_dropping_silently() -> None:
# Act
result = select_time_window(
frame_timestamp_ns=1_000,
sample_timestamps_ns=[100, 200, 300],
tolerance_ns=50,
)
# Assert
assert result.ok is False
assert result.sample_timestamps_ns == ()
assert result.violations[0].category == "gap_exceeded"