rutster/crates/rutster-trunk/tests/ws_ping.rs

//! 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::<RegisterTrunkInboundChannel>(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");
}