Files
autopilot/crates/vlm_client/tests/parser.rs
T
Oleksandr Bezdieniezhnykh e56d428753 [AZ-649] [AZ-674] [AZ-667] telemetry + vlm schema + mapobjects hydrate batch 6
AZ-649 mission_executor telemetry forwarding:
- shared::models::telemetry::UavTelemetry canonical model
- TelemetryForwarder with atomic ArcSwap snapshot + 3 lossy
  tokio::sync::broadcast channels (MissionExecutor, ScanController,
  MavlinkUplink) + per-consumer drop counters
- MavlinkProjection::from_mavlink for HEARTBEAT/GLOBAL_POSITION_INT/
  ATTITUDE/SYS_STATUS
- spawn_mavlink_pump bridges mavlink_layer into the forwarder at the
  binary edge

AZ-674 vlm_client schema validation + model_version tracking:
- AssessmentParser owns schema validation + model-version state
- wire::read_response_raw splits raw bytes from parsing so invalid
  payloads can be logged size-capped
- VlmStatus gains an Inconclusive variant; exhaustive-match test
  guards downstream consumers
- VlmPipelineStatus mirrors the new variant in shared::models::poi

AZ-667 mapobjects_store hydrate + pending logs + cascade:
- SyncState enum aligned with description.md (FreshBoot, Synced,
  CachedFallback, Degraded, Failed)
- Store::hydrate(MapObjectsBundle) replaces in-memory map atomically;
  freshness=Stale -> CachedFallback
- classify() + end_of_pass append MapObjectObservation events to
  pending_observations (New/Moved/Existing/RemovedCandidate)
- apply_decline + LocalAppended ignored items append to pending_ignored
- drain_pending() returns and clears both logs
- cascade_mission(id) purges by_cell + IgnoredSet + pending logs
- Health surface reports sync_state, pending_obs, pending_ign

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-19 17:40:43 +03:00

204 lines
6.7 KiB
Rust

//! AZ-674 acceptance criteria.
//!
//! AC-1 — valid response parses successfully (round-trip through the
//! UDS fixture, verifying schema fields all survive).
//! AC-2 — schema-invalid response returns `status: SchemaInvalid` and
//! the schema-invalid counter increments.
//! AC-3 — model_version change logged once; subsequent identical
//! versions do NOT re-log (observed via the parser's `changes`
//! counter, which is incremented exactly once per change).
//! AC-4 — `VlmStatus` is exhaustive (compile-time check: this file
//! contains a `match` over every variant with no `_` arm).
#![cfg(feature = "vlm")]
use std::path::PathBuf;
use shared::contracts::VlmProvider;
use shared::models::vlm::{VlmLabel, VlmStatus};
use tempfile::tempdir;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixListener;
use vlm_client::VlmClient;
async fn fixture_emitting(
path: PathBuf,
bodies: Vec<serde_json::Value>,
) -> tokio::task::JoinHandle<()> {
let listener = UnixListener::bind(&path).unwrap();
tokio::spawn(async move {
for body in bodies {
let (mut s, _) = listener.accept().await.unwrap();
let mut lenbuf = [0u8; 4];
if s.read_exact(&mut lenbuf).await.is_err() {
return;
}
let len = u32::from_be_bytes(lenbuf) as usize;
let mut req = vec![0u8; len];
if s.read_exact(&mut req).await.is_err() {
return;
}
let bytes = serde_json::to_vec(&body).unwrap();
let _ = s.write_all(&(bytes.len() as u32).to_be_bytes()).await;
let _ = s.write_all(&bytes).await;
let _ = s.flush().await;
}
})
}
#[tokio::test]
async fn ac1_valid_response_parses_successfully() {
// Arrange
let dir = tempdir().unwrap();
let path = dir.path().join("nanollm.sock");
let body = serde_json::json!({
"label": "confirmed_concealed_position",
"confidence": 0.91,
"evidence_spans": ["thicket"],
"reason": "match",
"status": "ok",
"latency_ms": 42,
"model_version": "VILA1.5-3B-int4"
});
let fixture = fixture_emitting(path.clone(), vec![body]).await;
let client = VlmClient::open(&path).await.expect("connect");
// Act
let a = client
.assess(b"\xff\xd8\xff".to_vec(), "describe".into())
.await
.unwrap();
// Assert
assert_eq!(a.status, VlmStatus::Ok);
assert_eq!(a.label, VlmLabel::ConfirmedConcealedPosition);
assert_eq!(a.model_version, "VILA1.5-3B-int4");
assert_eq!(a.latency_ms, 42);
assert_eq!(a.evidence_spans, vec!["thicket".to_string()]);
// Parser counters reflect the success path.
let parser = client.inner().unwrap().parser();
assert_eq!(parser.schema_invalid_count(), 0);
assert_eq!(parser.model_version_changes(), 1);
fixture.await.unwrap();
}
#[tokio::test]
async fn ac2_schema_invalid_response_returns_schema_invalid_and_increments_counter() {
// Arrange — fixture responds with valid JSON missing `model_version`.
let dir = tempdir().unwrap();
let path = dir.path().join("nanollm.sock");
let bad_body = serde_json::json!({
"label": "rejected",
"confidence": 0.4,
"status": "ok",
"latency_ms": 5
// model_version intentionally missing
});
let fixture = fixture_emitting(path.clone(), vec![bad_body]).await;
let client = VlmClient::open(&path).await.expect("connect");
// Act
let a = client.assess(b"r".to_vec(), "p".into()).await.unwrap();
// Assert
assert_eq!(a.status, VlmStatus::SchemaInvalid);
assert!(a.reason.starts_with("json:"), "got reason={}", a.reason);
let parser = client.inner().unwrap().parser();
assert_eq!(parser.schema_invalid_count(), 1);
assert_eq!(parser.model_version_changes(), 0);
fixture.await.unwrap();
}
/// AC-3 is exercised at the parser level — the model-version tracker
/// is a pure-state component that does not depend on the UDS layer.
/// The integration path is verified by AC-1 (one happy-path round
/// trip → parser sees one change event).
#[test]
fn ac3_model_version_change_logged_once_at_parser_level() {
use vlm_client::AssessmentParser;
// Arrange
let parser = AssessmentParser::new();
let mk = |v: &str| {
serde_json::to_vec(&serde_json::json!({
"label": "rejected",
"confidence": 0.5,
"status": "ok",
"latency_ms": 1,
"model_version": v
}))
.unwrap()
};
// Act — three responses: v1.0, v1.0 (no change), v1.1 (change).
let _ = parser.parse(&mk("v1.0"));
let _ = parser.parse(&mk("v1.0"));
let _ = parser.parse(&mk("v1.1"));
// Assert — exactly 2 change events: None→v1.0 and v1.0→v1.1.
assert_eq!(parser.model_version_changes(), 2);
assert_eq!(parser.current_model_version().as_deref(), Some("v1.1"));
}
/// Compile-time AC-4: this match must cover every `VlmStatus` variant
/// without a `_` arm. Adding a new variant breaks the build until
/// the consumer is updated.
#[test]
fn ac4_vlm_status_match_is_exhaustive() {
// Arrange — synthesise one of each variant.
let cases = [
VlmStatus::Ok,
VlmStatus::Inconclusive,
VlmStatus::Timeout,
VlmStatus::SchemaInvalid,
VlmStatus::IpcError,
VlmStatus::Disabled,
];
// Act / Assert — every variant must produce a labelled outcome.
for s in cases {
let label: &'static str = match s {
VlmStatus::Ok => "ok",
VlmStatus::Inconclusive => "inconclusive",
VlmStatus::Timeout => "timeout",
VlmStatus::SchemaInvalid => "schema_invalid",
VlmStatus::IpcError => "ipc_error",
VlmStatus::Disabled => "disabled",
};
assert!(!label.is_empty());
}
}
#[test]
fn schema_invalid_does_not_pollute_model_version_tracker() {
use vlm_client::AssessmentParser;
// Arrange — one valid body followed by one truncated/invalid body.
// The tracker must not regress to None on the second call.
let parser = AssessmentParser::new();
let good = serde_json::to_vec(&serde_json::json!({
"label": "rejected",
"confidence": 0.5,
"status": "ok",
"latency_ms": 1,
"model_version": "v1.0"
}))
.unwrap();
let bad = good[..good.len() - 10].to_vec();
// Act
let r1 = parser.parse(&good);
let r2 = parser.parse(&bad);
// Assert
assert_eq!(r1.status, VlmStatus::Ok);
assert_eq!(r2.status, VlmStatus::SchemaInvalid);
assert_eq!(parser.model_version_changes(), 1);
assert_eq!(parser.current_model_version().as_deref(), Some("v1.0"));
assert_eq!(parser.schema_invalid_count(), 1);
}