irtt-rs/crates/irtt-server/tests/runtime.rs

use std::{net::Ipv4Addr, time::Duration};

use irtt_proto::{
    decode_echo_reply, decode_open_reply, encode_request, Clock, Params, ReceivedStats,
    RequestToEncode, StampAt,
};
use tokio::{net::UdpSocket, sync::oneshot, time::timeout};

use irtt_server::{Server, ServerConfig, ServerRuntimeError};
use std::{io, net::SocketAddr};

#[tokio::test(flavor = "current_thread")]
async fn bind_reports_local_address_and_shutdown_stops_run() {
    let requested = SocketAddr::from((Ipv4Addr::LOCALHOST, 0));
    let mut server = Server::bind(requested, ServerConfig::default())
        .await
        .unwrap();
    let bound = server.local_addr().unwrap();
    assert_eq!(bound.ip(), requested.ip());
    assert_ne!(bound.port(), 0);

    let (shutdown_tx, shutdown_rx) = oneshot::channel();
    shutdown_tx.send(()).unwrap();
    server
        .run(async {
            let _ = shutdown_rx.await;
        })
        .await
        .unwrap();
}

#[tokio::test(flavor = "current_thread")]
async fn ipv4_open_echo_close_crosses_a_real_udp_socket() {
    exercise_udp_path(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).await;
}

#[tokio::test(flavor = "current_thread")]
async fn ipv6_open_echo_close_when_loopback_is_available() {
    let addr = "[::1]:0".parse().unwrap();
    let server = match Server::bind(addr, unthrottled()).await {
        Ok(server) => server,
        Err(ServerRuntimeError::Bind { source, .. })
            if matches!(
                source.kind(),
                io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
            ) =>
        {
            return;
        }
        Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"),
    };
    exercise_bound_server(server).await;
}

async fn exercise_udp_path(addr: SocketAddr) {
    let server = Server::bind(addr, unthrottled()).await.unwrap();
    exercise_bound_server(server).await;
}

async fn exercise_bound_server(mut server: Server) {
    let server_addr = server.local_addr().unwrap();
    let client_bind = match server_addr {
        SocketAddr::V4(_) => "127.0.0.1:0",
        SocketAddr::V6(_) => "[::1]:0",
    };
    let client = UdpSocket::bind(client_bind).await.unwrap();
    let (shutdown_tx, shutdown_rx) = oneshot::channel();
    let server_task = tokio::spawn(async move {
        server
            .run(async {
                let _ = shutdown_rx.await;
            })
            .await
    });

    let requested = Params {
        protocol_version: 1,
        duration_ns: 1_000_000_000,
        interval_ns: 100_000_000,
        received_stats: ReceivedStats::Both,
        stamp_at: StampAt::Both,
        clock: Clock::Both,
        ..Params::default()
    };
    let open = encode_request(
        RequestToEncode::Open {
            params: &requested,
            no_test: false,
        },
        None,
    )
    .unwrap();
    client.send_to(&open, server_addr).await.unwrap();

    let mut buffer = vec![0; 65_536];
    let (len, reply_source) = timeout(Duration::from_secs(1), client.recv_from(&mut buffer))
        .await
        .expect("OPEN reply timeout")
        .unwrap();
    assert_eq!(reply_source, server_addr);
    let open_reply = decode_open_reply(&buffer[..len], None).unwrap();
    assert_ne!(open_reply.token, 0);

    let echo = encode_request(
        RequestToEncode::Echo {
            token: open_reply.token,
            sequence: 0,
            params: &open_reply.params,
            payload: &[],
        },
        None,
    )
    .unwrap();
    client.send_to(&echo, server_addr).await.unwrap();
    let (len, reply_source) = timeout(Duration::from_secs(1), client.recv_from(&mut buffer))
        .await
        .expect("ECHO reply timeout")
        .unwrap();
    assert_eq!(reply_source, server_addr);
    let echo_reply = decode_echo_reply(&buffer[..len], &open_reply.params, None).unwrap();
    assert_eq!(echo_reply.token, open_reply.token);
    assert_eq!(echo_reply.sequence, 0);
    assert_eq!(echo_reply.recv_count, Some(1));
    assert_eq!(echo_reply.recv_window, Some(1));

    let close = encode_request(
        RequestToEncode::Close {
            token: open_reply.token,
        },
        None,
    )
    .unwrap();
    client.send_to(&close, server_addr).await.unwrap();
    client.send_to(&echo, server_addr).await.unwrap();
    assert!(
        timeout(Duration::from_millis(100), client.recv_from(&mut buffer))
            .await
            .is_err()
    );

    shutdown_tx.send(()).unwrap();
    server_task.await.unwrap().unwrap();
}

fn unthrottled() -> ServerConfig {
    ServerConfig::default().with_min_send_interval(Duration::ZERO)
}

#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "freebsd")))]
#[tokio::test(flavor = "current_thread")]
async fn wildcard_construction_is_refused_without_reply_source_selection() {
    let addr = "0.0.0.0:0".parse().unwrap();
    assert!(matches!(
        Server::bind(addr, unthrottled()).await,
        Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { .. })
    ));
}