//! App-level WS ping emission on the trunk media-stream WS (deploy //! slice A §5.2). //! //! Spins the real router on an ephemeral listener, performs the Twilio //! connected/start handshake as a tokio-tungstenite client, then asserts //! an engine-originated Ping frame arrives well within CI slack of the //! configured interval. Twilio documents no keepalive of its own — this //! test is the contract that WE originate the pings. use futures_util::{SinkExt, StreamExt}; use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer}; use tokio::sync::mpsc; use tokio_tungstenite::tungstenite::Message; #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn engine_originates_ws_ping_at_configured_interval() { let (register_tx, mut register_rx) = mpsc::channel::(4); let app = TwilioMediaStreamsServer::router(register_tx, std::time::Duration::from_millis(100)); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); // Stub media thread: ack the registration and hold the channel ends // alive so the pump loop stays up for the whole test. let registered = tokio::spawn(async move { let register = register_rx.recv().await.expect("pump registers"); let RegisterTrunkInboundChannel { inbound_from_twilio_rx, outbound_to_twilio_tx, reply, .. } = register; let _ = reply.send(rutster_call_model::ChannelId::new()); (inbound_from_twilio_rx, outbound_to_twilio_tx) }); let (mut ws, _resp) = tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream")) .await .expect("WS connect"); ws.send(Message::Text( r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(), )) .await .unwrap(); ws.send(Message::Text( r#"{"event":"start","start":{"streamSid":"MZping","callSid":"CAping"}}"#.into(), )) .await .unwrap(); let _register = registered.await.unwrap(); // 100 ms interval → a Ping must land within 2 s (20 intervals of // slack for a loaded CI runner; the point is emission, not cadence // precision). let got_ping = tokio::time::timeout(std::time::Duration::from_secs(2), async { while let Some(msg) = ws.next().await { if matches!(msg.expect("WS frame"), Message::Ping(_)) { return true; } } false }) .await .expect("a Ping frame must arrive before the deadline"); assert!(got_ping, "expected an engine-originated Ping frame"); }