Merge pull request #89 from Lochnair/test/real-server-peers

test: prefer real server for compliant peers

Cargo.lock

@@ -712,6 +712,7 @@ name = "irtt-client"
 version = "0.4.0"
 dependencies = [
  "irtt-proto",
+ "irtt-server",
  "nix 0.31.3",
  "socket2",
  "thiserror 2.0.20",

crates/irtt-cli/Cargo.toml

@@ -55,3 +55,7 @@ irtt-server = { workspace = true, optional = true }
 irtt-stats = { workspace = true, optional = true }
 ratatui = { version = "0.30.2", optional = true }
 tokio = { version = "1.53.1", default-features = false, features = ["rt", "time"], optional = true }
+
+[dev-dependencies]
+irtt-server.workspace = true
+tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }

crates/irtt-cli/tests/client_multi_target.rs

@@ -1,5 +1,8 @@
 #![cfg(feature = "client")]
 
+#[path = "support/in_tree_server.rs"]
+mod in_tree_server;
+
 use std::{
     io::Read,
     net::{SocketAddr, UdpSocket},
@@ -10,6 +13,7 @@ use std::{
     time::{Duration, Instant},
 };
 
+use in_tree_server::InTreeServer;
 use irtt_proto::{
     echo_packet_len, encode_echo_reply, encode_open_reply,
     flags::{FLAG_CLOSE, FLAG_OPEN, FLAG_REPLY},
@@ -69,10 +73,7 @@ fn list_columns_succeeds_without_target() {
 
 #[test]
 fn single_target_default_table_omits_target_and_accepts_pacing() {
-    let server = start_echo_server(test_params(
-        Some(Duration::from_millis(30)),
-        Duration::from_millis(10),
-    ));
+    let server = InTreeServer::start();
 
     let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli"))
         .args([
@@ -89,7 +90,7 @@ fn single_target_default_table_omits_target_and_accepts_pacing() {
         .output()
         .unwrap();
 
-    server.join();
+    drop(server);
 
     assert!(
         output.status.success(),
@@ -104,10 +105,7 @@ fn single_target_default_table_omits_target_and_accepts_pacing() {
 
 #[test]
 fn single_target_custom_target_column_renders_positional_label() {
-    let server = start_echo_server(test_params(
-        Some(Duration::from_millis(30)),
-        Duration::from_millis(10),
-    ));
+    let server = InTreeServer::start();
     let target = server.addr.to_string();
 
     let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli"))
@@ -127,7 +125,7 @@ fn single_target_custom_target_column_renders_positional_label() {
         .output()
         .unwrap();
 
-    server.join();
+    drop(server);
 
     assert!(
         output.status.success(),
@@ -238,9 +236,8 @@ fn continuous_single_target_interruption_drains_final_events_and_succeeds() {
 
 #[test]
 fn multi_target_csv_emits_rows_for_both_labels() {
-    let params = test_params(Some(Duration::from_millis(40)), Duration::from_millis(10));
-    let a = start_echo_server(params.clone());
-    let b = start_echo_server(params);
+    let a = InTreeServer::start();
+    let b = InTreeServer::start();
 
     let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli"))
         .args([
@@ -262,8 +259,8 @@ fn multi_target_csv_emits_rows_for_both_labels() {
         .output()
         .unwrap();
 
-    a.join();
-    b.join();
+    drop(a);
+    drop(b);
 
     assert!(
         output.status.success(),
@@ -643,20 +640,9 @@ fn wait_with_timeout(mut child: std::process::Child, timeout: Duration) -> Outpu
     }
 }
 
-fn start_echo_server(params: Params) -> FakeServer {
-    start_echo_server_inner(params, None)
-}
-
 fn start_echo_server_with_first_reply(
     params: Params,
     first_reply_tx: mpsc::Sender<()>,
-) -> FakeServer {
-    start_echo_server_inner(params, Some(first_reply_tx))
-}
-
-fn start_echo_server_inner(
-    params: Params,
-    mut first_reply_tx: Option<mpsc::Sender<()>>,
 ) -> FakeServer {
     let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
     socket
@@ -664,6 +650,7 @@ fn start_echo_server_inner(
         .unwrap();
     let addr = socket.local_addr().unwrap();
     let done = thread::spawn(move || {
+        let mut first_reply_tx = Some(first_reply_tx);
         let (_, peer) = recv_request(&socket);
         socket
             .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)

crates/irtt-cli/tests/support/in_tree_server.rs

@@ -0,0 +1,74 @@
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::mpsc,
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+
+use irtt_server::{Server, ServerConfig};
+use tokio::sync::oneshot;
+
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(2);
+
+pub struct InTreeServer {
+    pub addr: SocketAddr,
+    shutdown: Option<oneshot::Sender<()>>,
+    done: Option<JoinHandle<Result<(), irtt_server::ServerRuntimeError>>>,
+}
+
+impl InTreeServer {
+    pub fn start() -> Self {
+        let (startup_tx, startup_rx) = mpsc::sync_channel(1);
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let done = thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("failed to build in-tree test server runtime");
+            runtime.block_on(async move {
+                let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
+                let mut server = Server::bind(bind_addr, ServerConfig::default()).await?;
+                startup_tx
+                    .send(server.local_addr()?)
+                    .expect("test stopped waiting for in-tree server startup");
+                server
+                    .run(async {
+                        let _ = shutdown_rx.await;
+                    })
+                    .await
+            })
+        });
+
+        let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) {
+            Ok(addr) => addr,
+            Err(error) => {
+                let _ = shutdown_tx.send(());
+                let result = done.join().expect("in-tree test server thread panicked");
+                panic!("in-tree test server failed before startup ({error}): {result:?}");
+            }
+        };
+
+        Self {
+            addr,
+            shutdown: Some(shutdown_tx),
+            done: Some(done),
+        }
+    }
+
+    fn stop(&mut self) {
+        if let Some(shutdown) = self.shutdown.take() {
+            let _ = shutdown.send(());
+        }
+        if let Some(done) = self.done.take() {
+            done.join()
+                .expect("in-tree test server thread panicked")
+                .expect("in-tree test server stopped with an error");
+        }
+    }
+}
+
+impl Drop for InTreeServer {
+    fn drop(&mut self) {
+        self.stop();
+    }
+}

crates/irtt-client/Cargo.toml

@@ -27,3 +27,7 @@ tokio = ["dep:tokio"]
 
 [package.metadata.docs.rs]
 all-features = true
+
+[dev-dependencies]
+irtt-server.workspace = true
+tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] }

crates/irtt-client/src/async_client/tests.rs

@@ -14,11 +14,16 @@ use irtt_proto::{
 };
 use tokio::runtime::{Builder, Runtime};
 
+#[path = "../../tests/support/in_tree_server.rs"]
+mod in_tree_server;
+
 use super::*;
 use crate::{
     probe::PendingProbe, socket_options::tokio_socket_traffic_class, Client, ClientTimestamp,
     RunMode, SocketConfig,
 };
+use in_tree_server::InTreeServer;
+use irtt_server::ServerConfig;
 
 const TOKEN: u64 = 0x1234_5678_90ab_cdef;
 
@@ -248,24 +253,6 @@ fn start_peer_close_server() -> TestServer {
     })
 }
 
-fn start_echo_close_server(key: Option<Vec<u8>>) -> TestServer {
-    start_server(move |socket, tx| {
-        let (params, peer) = open_session(&socket, &tx, key.as_deref());
-
-        let (probe_packet, _) = recv_packet(&socket, &tx);
-        let sequence = echo_request_sequence(&probe_packet, key.as_deref());
-        socket
-            .send_to(
-                &echo_reply(&params, sequence, TOKEN, flags::FLAG_REPLY, key.as_deref()),
-                peer,
-            )
-            .unwrap();
-
-        let (close_packet, _) = recv_packet(&socket, &tx);
-        assert_eq!(close_request_token(&close_packet, key.as_deref()), TOKEN);
-    })
-}
-
 fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
     let mut context = Context::from_waker(Waker::noop());
     Future::poll(future, &mut context)
@@ -1084,16 +1071,16 @@ fn successful_close_uses_post_acceptance_timestamp() {
 #[test]
 fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() {
     let key = b"async-equivalence-key".to_vec();
-    let blocking_server = start_echo_close_server(Some(key.clone()));
+    let blocking_server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
     let mut blocking =
         Client::connect(config(blocking_server.addr, Some(key.clone()), 46)).unwrap();
     let blocking_open = blocking.open().unwrap();
     let blocking_sent = blocking.send_probe().unwrap();
     let blocking_reply = blocking.recv_once().unwrap();
     let blocking_close = blocking.close().unwrap();
-    assert_eq!(blocking_server.finish().len(), 3);
+    drop(blocking_server);
 
-    let async_server = start_echo_close_server(Some(key.clone()));
+    let async_server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
     let (async_open, async_sent, async_reply, async_close) = runtime().block_on(async {
         let mut client = AsyncClient::connect(config(async_server.addr, Some(key), 46))
             .await
@@ -1105,7 +1092,7 @@ fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() {
         let closed = client.close().await.unwrap();
         (opened, sent, reply, closed)
     });
-    assert_eq!(async_server.finish().len(), 3);
+    drop(async_server);
 
     assert_eq!(
         open_negotiated(&blocking_open),
@@ -1118,20 +1105,28 @@ fn blocking_and_async_hmac_dscp_lifecycle_are_semantically_equivalent() {
     );
     assert_matching_event_shape(&blocking_sent[0], &async_sent[0]);
     assert_matching_event_shape(&blocking_reply[0], &async_reply[0]);
+    assert!(matches!(
+        blocking_close.as_slice(),
+        [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&blocking_open)
+    ));
+    assert!(matches!(
+        async_close.as_slice(),
+        [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&async_open)
+    ));
     assert_matching_event_shape(&blocking_close[0], &async_close[0]);
 }
 
 #[test]
 fn blocking_and_async_no_test_are_semantically_equivalent() {
-    let blocking_no_test_server = start_no_test_server();
+    let blocking_no_test_server = InTreeServer::start(ServerConfig::default());
     let mut blocking_no_test_config = config(blocking_no_test_server.addr, None, 0);
     blocking_no_test_config.run_mode = RunMode::NoTest;
     let mut blocking_no_test = Client::connect(blocking_no_test_config).unwrap();
     let blocking_no_test_open = blocking_no_test.open().unwrap();
     assert!(blocking_no_test.is_run_complete());
-    assert_eq!(blocking_no_test_server.finish().len(), 1);
+    drop(blocking_no_test_server);
 
-    let async_no_test_server = start_no_test_server();
+    let async_no_test_server = InTreeServer::start(ServerConfig::default());
     let mut async_no_test_config = config(async_no_test_server.addr, None, 0);
     async_no_test_config.run_mode = RunMode::NoTest;
     let async_no_test_open = runtime().block_on(async {
@@ -1140,7 +1135,7 @@ fn blocking_and_async_no_test_are_semantically_equivalent() {
         assert!(client.is_run_complete());
         opened
     });
-    assert_eq!(async_no_test_server.finish().len(), 1);
+    drop(async_no_test_server);
     assert!(matches!(
         blocking_no_test_open,
         OpenOutcome::NoTestCompleted { .. }
@@ -1197,6 +1192,13 @@ fn open_negotiated(outcome: &OpenOutcome) -> &crate::NegotiatedParams {
     }
 }
 
+fn open_token(outcome: &OpenOutcome) -> u64 {
+    match outcome {
+        OpenOutcome::Started { token, .. } => *token,
+        OpenOutcome::NoTestCompleted { .. } => panic!("no-test open has no session token"),
+    }
+}
+
 fn assert_matching_event_shape(blocking: &ClientEvent, asynchronous: &ClientEvent) {
     assert_eq!(event_shape(blocking), event_shape(asynchronous));
 }
@@ -1205,7 +1207,7 @@ fn event_shape(event: &ClientEvent) -> (&'static str, u64, usize) {
     match event {
         ClientEvent::EchoSent { seq, bytes, .. } => ("sent", u64::from(*seq), *bytes),
         ClientEvent::EchoReply { seq, bytes, .. } => ("reply", u64::from(*seq), *bytes),
-        ClientEvent::SessionClosed { token, .. } => ("closed", *token, 0),
+        ClientEvent::SessionClosed { .. } => ("closed", 0, 0),
         other => panic!("unexpected equivalence event: {other:?}"),
     }
 }

crates/irtt-client/tests/hmac.rs

@@ -276,7 +276,7 @@ fn assert_no_duplicate_late_or_warning(events: &[ClientEvent]) {
 fn backend_hmac_correct_key_succeeds() {
     let key = b"compat-secret".to_vec();
     let params = default_params();
-    let peer = BackendPeer::start_open_echo(params.clone(), Some(key.clone()));
+    let peer = BackendPeer::start_open_echo(Some(key.clone()));
     let mut config = config_for_params(peer.addr(), &params);
     config.hmac_key = Some(key);
 

crates/irtt-client/tests/large_packets.rs

@@ -83,7 +83,7 @@ fn echo_reply_bytes_consistent_with_echo_packet_len_for_various_lengths() {
 #[test]
 fn backend_large_packet_smoke() {
     let params = params_with_length(1472);
-    let peer = BackendPeer::start_open_echo(params.clone(), None);
+    let peer = BackendPeer::start_open_echo(None);
     let mut client = Client::connect(ClientConfig {
         negotiation_policy: NegotiationPolicy::Loose,
         length: 1472,

crates/irtt-client/tests/protocol_options/backend_smoke.rs

@@ -6,7 +6,7 @@ use crate::support::{config_for_params, default_params, params_for_modes, Backen
 #[test]
 fn backend_basic_open_echo_close() {
     let params = default_params();
-    let peer = BackendPeer::start_open_echo(params.clone(), None);
+    let peer = BackendPeer::start_open_echo(None);
     let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
 
     let outcome = client.open().unwrap();
@@ -31,7 +31,7 @@ fn backend_received_stats_smoke() {
         ReceivedStats::Both,
     ] {
         let params = params_for_modes(mode, StampAt::None, Clock::Both);
-        let peer = BackendPeer::start_open_echo(params.clone(), None);
+        let peer = BackendPeer::start_open_echo(None);
         let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
 
         client.open().unwrap();
@@ -54,7 +54,7 @@ fn backend_timestamp_smoke() {
         StampAt::Midpoint,
     ] {
         let params = params_for_modes(ReceivedStats::None, mode, Clock::Both);
-        let peer = BackendPeer::start_open_echo(params.clone(), None);
+        let peer = BackendPeer::start_open_echo(None);
         let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
 
         client.open().unwrap();
@@ -101,7 +101,7 @@ fn backend_timestamp_smoke() {
 fn backend_clock_smoke() {
     for clock in [Clock::Wall, Clock::Monotonic, Clock::Both] {
         let params = params_for_modes(ReceivedStats::None, StampAt::Both, clock);
-        let peer = BackendPeer::start_open_echo(params.clone(), None);
+        let peer = BackendPeer::start_open_echo(None);
         let mut client = Client::connect(config_for_params(peer.addr(), &params)).unwrap();
 
         client.open().unwrap();

crates/irtt-client/tests/server_fill.rs

@@ -5,7 +5,7 @@ use irtt_proto::{Params, TimestampFields};
 
 use support::{
     config_for_params, default_params, run_one_probe, run_one_probe_with_config, server_fill,
-    start_open_server, OneProbeRun, ServerObservation,
+    start_open_server, InTreeServer, OneProbeRun, ServerObservation,
 };
 
 use crate::support::test_echo_packet_len;
@@ -68,8 +68,21 @@ fn server_fill_empty_and_oversized_values_are_rejected_at_config_boundary() {
 fn server_fill_negotiated_params_reflect_accepted_value() {
     let mut params = default_params();
     params.server_fill = server_fill("rand");
-    let run = run_one_probe(params.clone(), TimestampFields::default());
-    assert_eq!(run.negotiated.params.server_fill, params.server_fill);
+    let server = InTreeServer::start(irtt_server::ServerConfig::default());
+    let mut client = Client::connect(config_for_params(server.addr, &params)).unwrap();
+    let outcome = client.open().unwrap();
+    let negotiated = match outcome {
+        irtt_client::OpenOutcome::Started { negotiated, .. } => negotiated,
+        other => panic!("expected started outcome, got {other:?}"),
+    };
+    assert_eq!(negotiated.params.server_fill, params.server_fill);
+
+    client.send_probe().unwrap();
+    assert!(matches!(
+        client.recv_once().unwrap().as_slice(),
+        [irtt_client::ClientEvent::EchoReply { .. }]
+    ));
+    client.close().unwrap();
 }
 
 #[test]

crates/irtt-client/tests/support/in_tree_server.rs

@@ -0,0 +1,74 @@
+use std::{
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::mpsc,
+    thread::{self, JoinHandle},
+    time::Duration,
+};
+
+use irtt_server::{Server, ServerConfig};
+use tokio::sync::oneshot;
+
+const STARTUP_TIMEOUT: Duration = Duration::from_secs(2);
+
+pub struct InTreeServer {
+    pub addr: SocketAddr,
+    shutdown: Option<oneshot::Sender<()>>,
+    done: Option<JoinHandle<Result<(), irtt_server::ServerRuntimeError>>>,
+}
+
+impl InTreeServer {
+    pub fn start(config: ServerConfig) -> Self {
+        let (startup_tx, startup_rx) = mpsc::sync_channel(1);
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let done = thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("failed to build in-tree test server runtime");
+            runtime.block_on(async move {
+                let bind_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0);
+                let mut server = Server::bind(bind_addr, config).await?;
+                startup_tx
+                    .send(server.local_addr()?)
+                    .expect("test stopped waiting for in-tree server startup");
+                server
+                    .run(async {
+                        let _ = shutdown_rx.await;
+                    })
+                    .await
+            })
+        });
+
+        let addr = match startup_rx.recv_timeout(STARTUP_TIMEOUT) {
+            Ok(addr) => addr,
+            Err(error) => {
+                let _ = shutdown_tx.send(());
+                let result = done.join().expect("in-tree test server thread panicked");
+                panic!("in-tree test server failed before startup ({error}): {result:?}");
+            }
+        };
+
+        Self {
+            addr,
+            shutdown: Some(shutdown_tx),
+            done: Some(done),
+        }
+    }
+
+    fn stop_inner(&mut self) {
+        if let Some(shutdown) = self.shutdown.take() {
+            let _ = shutdown.send(());
+        }
+        if let Some(done) = self.done.take() {
+            done.join()
+                .expect("in-tree test server thread panicked")
+                .expect("in-tree test server stopped with an error");
+        }
+    }
+}
+
+impl Drop for InTreeServer {
+    fn drop(&mut self) {
+        self.stop_inner();
+    }
+}

crates/irtt-client/tests/support/mod.rs

@@ -1,5 +1,6 @@
 #![allow(dead_code)]
 
+mod in_tree_server;
 mod real_irtt;
 
 use std::{
@@ -17,7 +18,9 @@ use irtt_proto::{
     PacketLayout, Params, ProtoError, ReceivedStats, ServerFill, StampAt, TimestampFields,
     HMAC_SIZE, MAGIC, PROTOCOL_VERSION,
 };
+use irtt_server::ServerConfig;
 
+pub use in_tree_server::InTreeServer;
 pub use real_irtt::RealIrtServer;
 
 const HMAC_OFFSET: usize = 4;
@@ -27,7 +30,7 @@ pub const RECV_WINDOW: u64 = 0x07;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum TestBackendKind {
-    Fake,
+    InTree,
     Real,
 }
 
@@ -43,9 +46,9 @@ pub fn selected_backend() -> TestBackendKind {
                 _ => panic!("IRTT_TEST_BACKEND=real but irtt binary not found at '{irtt_bin}'"),
             }
         }
-        Ok("fake") | Err(_) => {
-            debug_backend("[backend] selected backend=fake");
-            TestBackendKind::Fake
+        Ok("in-tree") | Err(_) => {
+            debug_backend("[backend] selected backend=in-tree");
+            TestBackendKind::InTree
         }
         Ok(other) => panic!("unknown IRTT_TEST_BACKEND value: {other}"),
     }
@@ -58,24 +61,25 @@ fn debug_backend(message: &str) {
 }
 
 pub enum BackendPeer {
-    Fake(FakeServer),
+    InTree(InTreeServer),
     Real(RealIrtServer),
 }
 
 impl BackendPeer {
     pub fn addr(&self) -> SocketAddr {
         match self {
-            BackendPeer::Fake(s) => s.addr,
+            BackendPeer::InTree(s) => s.addr,
             BackendPeer::Real(s) => s.addr(),
         }
     }
 
-    pub fn start_open_echo(params: Params, hmac_key: Option<Vec<u8>>) -> Self {
+    pub fn start_open_echo(hmac_key: Option<Vec<u8>>) -> Self {
         match selected_backend() {
-            TestBackendKind::Fake => {
-                let timestamps = standard_timestamps();
-                BackendPeer::Fake(start_one_probe_server(params, timestamps, hmac_key))
-            }
+            TestBackendKind::InTree => BackendPeer::InTree(InTreeServer::start(
+                hmac_key.map_or_else(ServerConfig::default, |key| {
+                    ServerConfig::default().with_hmac_key(key)
+                }),
+            )),
             TestBackendKind::Real => {
                 BackendPeer::Real(RealIrtServer::start(hmac_key.as_deref()).unwrap())
             }
@@ -84,9 +88,8 @@ impl BackendPeer {
 
     pub fn start_hmac_required(key: Vec<u8>) -> Self {
         match selected_backend() {
-            TestBackendKind::Fake => BackendPeer::Fake(start_hmac_required_open_drop_server(
-                key,
-                Duration::from_millis(250),
+            TestBackendKind::InTree => BackendPeer::InTree(InTreeServer::start(
+                ServerConfig::default().with_hmac_key(key),
             )),
             TestBackendKind::Real => BackendPeer::Real(RealIrtServer::start(Some(&key)).unwrap()),
         }