mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 10:21:13 +00:00
start over again
This commit is contained in:
Oleksandr Bezdieniezhnykh
@@ -1 +0,0 @@
|
||||
|
||||
@@ -1,65 +0,0 @@
|
||||
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"
|
||||
@@ -1,41 +0,0 @@
|
||||
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"
|
||||
@@ -1,157 +0,0 @@
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from shared.contracts import (
|
||||
AnchorDecision,
|
||||
CacheTileRecord,
|
||||
FdrEvent,
|
||||
FramePacket,
|
||||
PositionEstimate,
|
||||
TelemetrySample,
|
||||
VioStatePacket,
|
||||
VprCandidate,
|
||||
)
|
||||
|
||||
|
||||
def test_runtime_dtos_accept_valid_minimal_values() -> None:
|
||||
# Arrange
|
||||
timestamp_ns = 1_000_000
|
||||
|
||||
# Act
|
||||
contracts = [
|
||||
FramePacket(
|
||||
frame_id="frame-1",
|
||||
timestamp_ns=timestamp_ns,
|
||||
image_ref="frames/frame-1",
|
||||
calibration_id="calib-1",
|
||||
occlusion="clear",
|
||||
quality=0.95,
|
||||
),
|
||||
TelemetrySample(
|
||||
timestamp_ns=timestamp_ns,
|
||||
imu={"accel_x": 0.1},
|
||||
attitude={"roll": 0.0},
|
||||
altitude_m=950.0,
|
||||
airspeed_mps=16.0,
|
||||
gps_health="lost",
|
||||
),
|
||||
VioStatePacket(
|
||||
timestamp_ns=timestamp_ns,
|
||||
relative_pose={"x": 1.0},
|
||||
velocity_mps=(1.0, 0.0, 0.0),
|
||||
tracking_quality=0.8,
|
||||
),
|
||||
PositionEstimate(
|
||||
timestamp_ns=timestamp_ns,
|
||||
latitude_deg=49.9,
|
||||
longitude_deg=36.2,
|
||||
altitude_m=950.0,
|
||||
covariance_semimajor_m=12.0,
|
||||
source_label="satellite_anchored",
|
||||
fix_type=3,
|
||||
horizontal_accuracy_m=12.0,
|
||||
anchor_age_ms=200,
|
||||
),
|
||||
VprCandidate(
|
||||
chunk_id="chunk-1",
|
||||
tile_id="tile-1",
|
||||
score=0.87,
|
||||
footprint={"min_lat": 49.0},
|
||||
freshness_status="fresh",
|
||||
),
|
||||
AnchorDecision(
|
||||
candidate_id="candidate-1",
|
||||
accepted=True,
|
||||
estimated_pose={"x": 1.0},
|
||||
inliers=42,
|
||||
mean_reprojection_error_px=0.8,
|
||||
),
|
||||
CacheTileRecord(
|
||||
tile_id="tile-1",
|
||||
crs="EPSG:3857",
|
||||
meters_per_pixel=0.3,
|
||||
capture_date="2026-05-03",
|
||||
signature_hash="sha256:abc",
|
||||
trust_level="trusted",
|
||||
freshness_status="fresh",
|
||||
provenance="suite-satellite-service",
|
||||
),
|
||||
FdrEvent(
|
||||
event_type="health",
|
||||
timestamp_ns=timestamp_ns,
|
||||
component="shared.contracts",
|
||||
severity="info",
|
||||
payload_ref="fdr://segment/1",
|
||||
mission_id="mission-1",
|
||||
run_id="run-1",
|
||||
),
|
||||
]
|
||||
|
||||
# Assert
|
||||
assert len(contracts) == 8
|
||||
|
||||
|
||||
def test_missing_required_timestamp_is_rejected_with_structured_error() -> None:
|
||||
# Act
|
||||
with pytest.raises(ValidationError) as error:
|
||||
FramePacket(
|
||||
frame_id="frame-1",
|
||||
image_ref="frames/frame-1",
|
||||
calibration_id="calib-1",
|
||||
occlusion="clear",
|
||||
quality=0.95,
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert error.value.errors()[0]["loc"] == ("timestamp_ns",)
|
||||
assert error.value.errors()[0]["type"] == "missing"
|
||||
|
||||
|
||||
def test_raw_frame_retention_is_rejected() -> None:
|
||||
# Act
|
||||
with pytest.raises(ValidationError) as error:
|
||||
FramePacket(
|
||||
frame_id="frame-1",
|
||||
timestamp_ns=1,
|
||||
image_ref="frames/frame-1",
|
||||
calibration_id="calib-1",
|
||||
occlusion="clear",
|
||||
quality=0.95,
|
||||
raw_frame_retained=True,
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert "raw frame payloads must be referenced" in str(error.value)
|
||||
|
||||
|
||||
def test_position_accuracy_cannot_under_report_covariance() -> None:
|
||||
# Act
|
||||
with pytest.raises(ValidationError) as error:
|
||||
PositionEstimate(
|
||||
timestamp_ns=1,
|
||||
latitude_deg=49.9,
|
||||
longitude_deg=36.2,
|
||||
altitude_m=950.0,
|
||||
covariance_semimajor_m=50.0,
|
||||
source_label="satellite_anchored",
|
||||
fix_type=3,
|
||||
horizontal_accuracy_m=10.0,
|
||||
anchor_age_ms=200,
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert "must not under-report" in str(error.value)
|
||||
|
||||
|
||||
def test_accepted_anchor_requires_estimated_pose() -> None:
|
||||
# Act
|
||||
with pytest.raises(ValidationError) as error:
|
||||
AnchorDecision(
|
||||
candidate_id="candidate-1",
|
||||
accepted=True,
|
||||
inliers=42,
|
||||
mean_reprojection_error_px=0.8,
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert "accepted anchor decisions require estimated_pose" in str(error.value)
|
||||
Reference in New Issue
Block a user