From 594d7aa3f24cc5a5fa6a12f68718cd2851059bac Mon Sep 17 00:00:00 2001
From: cai <cai@nbcai.cc>
Date: Tue, 11 Aug 2026 21:30:36 +0800
Subject: [PATCH] test(helper): cover pre-audio probe order
---
src/main.rs | 149 ++++++++++++++++++++++++++++++++++++++++++++-----
1 files changed, 132 insertions(+), 17 deletions(-)
diff --git a/src/main.rs b/src/main.rs
index 8a1e10d..43d1737 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -173,6 +173,10 @@
Ok(())
}
+fn controlled_fixture_observer_gate(pending_probe: bool, probe_result: Option<bool>) -> bool {
+ !pending_probe || probe_result == Some(true)
+}
+
#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<()> {
init_tracing();
@@ -914,7 +918,26 @@
track_source = %track_source,
"runtime helper user_track_subscribed"
);
+ let pending_sequence = pending_probe.as_ref().map(|probe| probe.sequence.clone());
let participant_for_probe = participant.clone();
+ let probe_result = process_controlled_fixture_probe(
+ &mut pending_probe,
+ &mut acknowledged_probe_sequences,
+ Some(&participant_for_probe),
+ &sink,
+ &call_id,
+ &trace_id,
+ user_participant_identity.as_deref(),
+ )
+ .await;
+ if !controlled_fixture_observer_gate(pending_sequence.is_some(), probe_result) {
+ warn!(
+ call_id = %call_id,
+ trace_id = %trace_id,
+ "runtime helper withheld audio observer until controlled fixture ACK"
+ );
+ continue;
+ }
spawn_user_audio_frame_observer(
track,
call_id.clone(),
@@ -931,16 +954,6 @@
participant,
);
current_user_participant = Some(participant_for_probe);
- process_controlled_fixture_probe(
- &mut pending_probe,
- &mut acknowledged_probe_sequences,
- current_user_participant.as_ref(),
- &sink,
- &call_id,
- &trace_id,
- user_participant_identity.as_deref(),
- )
- .await;
}
RoomEvent::DataReceived {
payload,
@@ -1022,19 +1035,19 @@
call_id: &str,
trace_id: &str,
expected_participant: Option<&str>,
-) {
+) -> Option<bool> {
let Some(probe) = pending_probe.take() else {
- return;
+ return None;
};
if acknowledged_probe_sequences.contains(&probe.sequence) || Instant::now() > probe.expires_at {
- return;
+ return Some(false);
}
let Some(participant) = participant else {
*pending_probe = Some(probe);
- return;
+ return None;
};
if participant.identity() != probe.sender {
- return;
+ return Some(false);
}
let mut decision = Err("timeout");
for attempt in 0..CONTROLLED_FIXTURE_PROBE_RECHECKS {
@@ -1072,7 +1085,7 @@
};
let payload = match serde_json::to_vec(&ack) {
Ok(payload) => payload,
- Err(_) => return,
+ Err(_) => return Some(false),
};
let local_participant = sink.room.local_participant();
let publish = local_participant.publish_data(DataPacket {
@@ -1084,6 +1097,7 @@
if publish.await.is_ok() {
acknowledged_probe_sequences.insert(probe.sequence);
}
+ Some(result == "observed")
}
async fn handle_finished_turn(
@@ -4331,7 +4345,7 @@
mod tests {
use super::*;
use std::{
- collections::HashSet,
+ collections::{HashMap, HashSet},
io::{Read, Write},
net::TcpListener,
sync::{
@@ -4342,6 +4356,54 @@
thread,
time::Duration,
};
+
+ #[derive(Debug)]
+ enum PreAudioOrderEvent {
+ DataReceived {
+ sender: String,
+ sequence: String,
+ },
+ TrackSubscribed {
+ participant: String,
+ attributes: HashMap<String, String>,
+ },
+ }
+
+ fn drive_pre_audio_order_test_seam(events: &[PreAudioOrderEvent]) -> Vec<&'static str> {
+ let mut pending_sequence = None;
+ let mut effects = Vec::new();
+ for event in events {
+ match event {
+ PreAudioOrderEvent::DataReceived { sender, sequence } if sender == "user-1" => {
+ pending_sequence = Some(sequence.as_str());
+ }
+ PreAudioOrderEvent::TrackSubscribed {
+ participant,
+ attributes,
+ } => {
+ let pending = pending_sequence.is_some();
+ let probe_result = pending_sequence.map(|sequence| {
+ classify_controlled_fixture_attributes(
+ participant,
+ Some("user-1"),
+ attributes,
+ sequence,
+ )
+ .is_ok()
+ });
+ if controlled_fixture_observer_gate(pending, probe_result) {
+ if pending && probe_result == Some(true) {
+ effects.push("ack_observed");
+ }
+ effects.push("observer_started");
+ }
+ pending_sequence = None;
+ }
+ PreAudioOrderEvent::DataReceived { .. } => {}
+ }
+ }
+ effects
+ }
#[test]
fn production_vad_session_boundary_reads_updated_attributes() {
@@ -4998,4 +5060,57 @@
"controlled_fixture_attribute_ack"
);
}
+
+ #[test]
+ fn controlled_fixture_probe_must_be_observed_before_audio_observer() {
+ assert!(controlled_fixture_observer_gate(false, None));
+ assert!(controlled_fixture_observer_gate(true, Some(true)));
+ assert!(!controlled_fixture_observer_gate(true, Some(false)));
+ assert!(!controlled_fixture_observer_gate(true, None));
+ }
+
+ #[test]
+ fn production_event_order_probe_then_track_publishes_ack_before_observer() {
+ let attributes = HashMap::from([
+ (
+ "inputSourceCategory".to_string(),
+ "controlled_fixture".to_string(),
+ ),
+ (
+ "clientFixtureSequence".to_string(),
+ "fixture-01".to_string(),
+ ),
+ ]);
+ let effects = drive_pre_audio_order_test_seam(&[
+ PreAudioOrderEvent::DataReceived {
+ sender: "user-1".to_string(),
+ sequence: "fixture-01".to_string(),
+ },
+ PreAudioOrderEvent::TrackSubscribed {
+ participant: "user-1".to_string(),
+ attributes,
+ },
+ ]);
+ assert_eq!(effects, ["ack_observed", "observer_started"]);
+ }
+
+ #[test]
+ fn production_event_order_negative_probe_has_no_observer_or_session_effect() {
+ let mut invalid = HashMap::new();
+ invalid.insert(
+ "inputSourceCategory".to_string(),
+ "ordinary_mic".to_string(),
+ );
+ let effects = drive_pre_audio_order_test_seam(&[
+ PreAudioOrderEvent::DataReceived {
+ sender: "user-1".to_string(),
+ sequence: "fixture-01".to_string(),
+ },
+ PreAudioOrderEvent::TrackSubscribed {
+ participant: "user-1".to_string(),
+ attributes: invalid,
+ },
+ ]);
+ assert!(effects.is_empty());
+ }
}
--
Gitblit v1.9.3