Lochnair-Labs/irtt-rs
2
0 Fork 0
Code Issues 1 Pull Requests 4 Actions 4 Packages Projects Releases 6 Wiki Activity

Tests cleanup #14

Merged
Lochnair merged 2 commits from tests-cleanup into main 2026-05-16 16:41:12 +00:00
Conversation 0 Commits 2 Files Changed 4
+19 -729
4 changed files with 19 additions and 729 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/irtt-client/src/client.rs
+5 -9
View File
@@ -457,7 +457,7 @@ impl Client {
.is_some_and(|h| sequence_is_before(wire_seq, h));
let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
update_highest_received(session, wire_seq);
update_highest_received(&mut session.highest_received_seq, wire_seq);
session.completed.insert(wire_seq);
let mut events = Vec::new();
@@ -494,7 +494,7 @@ impl Client {
}
Ok(events)
} else if session.completed.contains(wire_seq) {
update_highest_received(session, wire_seq);
update_highest_received(&mut session.highest_received_seq, wire_seq);
Ok(vec![ClientEvent::DuplicateReply {
seq: wire_seq,
remote: self.remote,
@@ -507,7 +507,7 @@ impl Client {
let one_way = compute_one_way(&timed_out.sent_at, &now, &reply.timestamps);
let received_stats = build_received_stats(&reply);
let highest_seen = session.highest_received_seq.unwrap_or(wire_seq);
update_highest_received(session, wire_seq);
update_highest_received(&mut session.highest_received_seq, wire_seq);
session.completed.insert(wire_seq);
let mut events = vec![ClientEvent::LateReply {
@@ -688,14 +688,10 @@ impl Client {
fn send_probe_at(&mut self, ts: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> {
self.send_probe_inner(Some(ts))
}
fn recv_once_at(&mut self, ts: ClientTimestamp) -> Result<Vec<ClientEvent>, ClientError> {
self.recv_once_inner(Some(ts))
}
}
fn update_highest_received(session: &mut ActiveSession, wire_seq: u32) {
session.highest_received_seq = Some(session.highest_received_seq.map_or(wire_seq, |h| {
fn update_highest_received(highest_received_seq: &mut Option<u32>, wire_seq: u32) {
*highest_received_seq = Some(highest_received_seq.map_or(wire_seq, |h| {
if sequence_is_after(wire_seq, h) {
wire_seq
} else {
crates/irtt-client/src/client/tests/close.rs
-25
View File
@@ -68,31 +68,6 @@ fn send_probe_fails_after_close() {
server.join();
}
#[test]
fn close_clears_timed_out_metadata() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
probe_timeout: Duration::from_millis(40),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60));
client.poll_timeouts().unwrap();
assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1);
client.close().unwrap();
assert!(client.session.is_none());
server.join();
}
#[test]
fn close_flagged_echo_reply_emits_reply_then_closes_without_sending_close() {
let params = default_params();
crates/irtt-client/src/client/tests/probes_replies.rs
+9 -673
View File
@@ -86,7 +86,7 @@ fn echo_sent_reports_schedule_and_timer_error() {
wall: SystemTime::now(),
};
assert_open_started(client.open().unwrap());
let session_start = client.session.as_ref().unwrap().start_mono;
let session_start = client.next_send_deadline().unwrap();
assert!(
session_start >= start.mono,
"probe schedule must start after open begins"
@@ -170,8 +170,7 @@ fn send_probe_respects_finite_duration_exclusive_end() {
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let session = client.session.as_ref().unwrap();
let start = session.start_mono;
let start = client.next_send_deadline().unwrap();
let interval = Duration::from_millis(500);
let now0 = ClientTimestamp {
@@ -192,7 +191,6 @@ fn send_probe_respects_finite_duration_exclusive_end() {
};
let events = client.send_probe_at(now2).unwrap();
assert!(events.is_empty());
assert!(client.session.as_ref().unwrap().sending_done);
assert!(client.next_send_deadline().is_none());
client.close().unwrap();
@@ -223,7 +221,7 @@ fn continuous_duration_keeps_generating_send_deadlines() {
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let start = client.session.as_ref().unwrap().start_mono;
let start = client.next_send_deadline().unwrap();
let interval = Duration::from_millis(500);
for seq in 0..4 {
let now = ClientTimestamp {
@@ -233,7 +231,6 @@ fn continuous_duration_keeps_generating_send_deadlines() {
let events = client.send_probe_at(now).unwrap();
assert_eq!(events.len(), 1);
assert!(client.next_send_deadline().is_some());
assert!(!client.session.as_ref().unwrap().sending_done);
}
client.close().unwrap();
@@ -245,92 +242,20 @@ fn next_send_deadline_advances_by_interval() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
let session = client.session.as_ref().unwrap();
let start = session.start_mono;
let deadline0 = client.next_send_deadline().unwrap();
assert_eq!(deadline0, start);
client.send_probe().unwrap();
let deadline1 = client.next_send_deadline().unwrap();
assert_eq!(deadline1, start + Duration::from_secs(1));
assert_eq!(deadline1, deadline0 + Duration::from_secs(1));
client.send_probe().unwrap();
let deadline2 = client.next_send_deadline().unwrap();
assert_eq!(deadline2, start + Duration::from_secs(2));
assert_eq!(deadline2, deadline0 + Duration::from_secs(2));
client.close().unwrap();
server.join();
}
#[test]
fn next_probe_deadline_reports_overflow() {
let start = Instant::now();
assert_eq!(
next_probe_deadline(start, 1_000_000_000, 2).unwrap(),
start + Duration::from_secs(2)
);
assert!(matches!(
next_probe_deadline(start, u64::MAX, 2),
Err(ClientError::DurationOverflow)
));
}
#[test]
fn send_probe_reports_schedule_overflow() {
let params = Params {
protocol_version: 1,
duration_ns: 0,
interval_ns: 1_000_000_000,
received_stats: ReceivedStats::Both,
stamp_at: StampAt::Both,
clock: Clock::Both,
..Params::default()
};
let server = silent_open_server(params);
let config = ClientConfig {
duration: None,
interval: Duration::from_secs(1),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.session.as_mut().unwrap().packets_sent = u64::MAX - 1;
assert!(matches!(
client.send_probe(),
Err(ClientError::DurationOverflow)
));
server.join();
}
#[test]
fn send_probe_reports_packets_sent_counter_overflow() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.session.as_mut().unwrap().packets_sent = u64::MAX;
assert!(matches!(
client.send_probe(),
Err(ClientError::CounterOverflow {
counter: "packets_sent"
})
));
server.join();
}
#[test]
fn recv_once_returns_empty_on_timeout() {
let params = default_params();
@@ -466,28 +391,6 @@ fn recv_once_decodes_only_received_bytes_after_longer_datagram() {
server.join();
}
#[test]
#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
fn echo_reply_metadata_is_unavailable_without_ancillary() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(50));
let events = client.recv_once().unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::EchoReply { packet_meta, .. } => {
assert_packet_meta_unavailable(packet_meta);
}
other => panic!("expected EchoReply, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
#[cfg(all(target_os = "linux", feature = "ancillary"))]
fn echo_reply_metadata_propagates_observed_dscp_with_ancillary() {
@@ -542,38 +445,6 @@ fn echo_reply_metadata_propagates_observed_dscp_with_ancillary() {
server.join();
}
#[test]
#[cfg(all(target_os = "linux", feature = "ancillary"))]
fn echo_reply_metadata_preserves_observed_zero_with_ancillary() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(50));
let events = client.recv_once().unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::EchoReply { packet_meta, .. } => {
let Some(traffic_class) = packet_meta.traffic_class else {
metadata_unavailable_skip(
"echo_reply_metadata_preserves_observed_zero_with_ancillary",
);
client.close().unwrap();
server.join();
return;
};
assert_eq!(traffic_class, 0);
assert_eq!(packet_meta.dscp, Some(0));
assert_eq!(packet_meta.ecn, Some(0));
}
other => panic!("expected EchoReply, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
#[cfg(all(target_os = "linux", feature = "ancillary"))]
fn echo_reply_metadata_propagates_kernel_rx_timestamp_with_ancillary() {
@@ -674,24 +545,6 @@ fn server_processing_greater_than_raw_does_not_underflow() {
assert_eq!(rtt.effective_signed, SignedDuration { ns: -999_999_999 });
}
#[test]
fn received_stats_parsed_into_sample() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(30));
let events = client.recv_once().unwrap();
if let ClientEvent::EchoReply { received_stats, .. } = &events[0] {
let rs = received_stats.as_ref().unwrap();
assert_eq!(rs.count, Some(42));
assert_eq!(rs.window, Some(0x07));
} else {
panic!("expected EchoReply");
}
client.close().unwrap();
server.join();
}
#[test]
fn wrong_token_reply_is_dropped() {
let params = default_params();
@@ -872,31 +725,6 @@ fn poll_timeouts_emits_echo_loss() {
server.join();
}
#[test]
fn poll_timeouts_removes_expired_pending() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
probe_timeout: Duration::from_millis(100),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(150));
client.poll_timeouts().unwrap();
let session = client.session.as_ref().unwrap();
assert_eq!(session.pending.len(), 0);
assert_eq!(session.timed_out.len(), 1);
server.join();
}
#[test]
fn late_reply_after_timeout_preserves_measurement_metadata() {
let params = default_params();
@@ -937,8 +765,6 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
assert_eq!(client.session.as_ref().unwrap().pending.len(), 0);
assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 1);
let late = client.recv_once().unwrap();
match &late[0] {
@@ -964,8 +790,6 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
}
other => panic!("expected stats-eligible LateReply, got {other:?}"),
}
assert_eq!(client.session.as_ref().unwrap().timed_out.len(), 0);
let duplicate = client.recv_once().unwrap();
assert!(matches!(
&duplicate[0],
@@ -980,149 +804,6 @@ fn late_reply_after_timeout_preserves_measurement_metadata() {
server.join();
}
#[test]
#[cfg(not(all(target_os = "linux", feature = "ancillary")))]
fn late_reply_metadata_is_unavailable_without_ancillary() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
let mut buf = [0_u8; 2048];
let (size, _) = socket.recv_from(&mut buf).unwrap();
tx.send(buf[..size].to_vec()).unwrap();
let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap());
thread::sleep(Duration::from_millis(90));
let reply_packet =
echo_reply_packet(TOKEN, seq, &params, &TimestampFields::default(), None);
socket.send_to(&reply_packet, peer).unwrap();
});
let config = ClientConfig {
probe_timeout: Duration::from_millis(40),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let late = client.recv_once().unwrap();
assert_eq!(late.len(), 1);
match &late[0] {
ClientEvent::LateReply { packet_meta, .. } => {
assert_packet_meta_unavailable(packet_meta);
}
other => panic!("expected LateReply, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
#[cfg(all(target_os = "linux", feature = "ancillary"))]
fn late_reply_metadata_propagates_observed_dscp_with_ancillary() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
let mut buf = [0_u8; 2048];
let (size, _) = socket.recv_from(&mut buf).unwrap();
tx.send(buf[..size].to_vec()).unwrap();
let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap());
thread::sleep(Duration::from_millis(90));
crate::socket_options::apply_dscp_to_socket(&socket, peer, 46).unwrap();
let reply_packet =
echo_reply_packet(TOKEN, seq, &params, &TimestampFields::default(), None);
socket.send_to(&reply_packet, peer).unwrap();
});
let config = ClientConfig {
probe_timeout: Duration::from_millis(40),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let events = client.recv_once().unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::LateReply { packet_meta, .. } => {
let Some(traffic_class) = packet_meta.traffic_class else {
metadata_unavailable_skip(
"late_reply_metadata_propagates_observed_dscp_with_ancillary",
);
client.close().unwrap();
server.join();
return;
};
assert_eq!(traffic_class, 184);
assert_eq!(packet_meta.dscp, Some(46));
assert_eq!(packet_meta.ecn, Some(0));
if let Some(timestamp) = packet_meta.kernel_rx_timestamp {
assert!(timestamp.duration_since(SystemTime::UNIX_EPOCH).is_ok());
} else {
kernel_rx_timestamp_unavailable_skip(
"late_reply_metadata_propagates_observed_dscp_with_ancillary",
);
}
}
other => panic!("expected LateReply, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
fn pending_map_bounded() {
let params = Params {
duration_ns: 60_000_000_000,
..default_params()
};
let server = silent_open_server(params);
let config = ClientConfig {
duration: Some(Duration::from_secs(60)),
max_pending_probes: 3,
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
client.send_probe().unwrap();
client.send_probe().unwrap();
assert!(matches!(
client.send_probe(),
Err(ClientError::PendingLimitExceeded { limit: 3 })
));
client.close().unwrap();
server.join();
}
#[test]
fn minimal_negotiated_layout_works() {
let params = Params {
@@ -1180,65 +861,6 @@ fn minimal_negotiated_layout_works() {
server.join();
}
#[test]
fn late_reply_with_pending_preserves_rtt() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
socket
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let mut seqs = Vec::new();
for _ in 0..3 {
let mut buf = [0_u8; 2048];
if let Ok((size, _)) = socket.recv_from(&mut buf) {
tx.send(buf[..size].to_vec()).unwrap();
let seq = u32::from_le_bytes(buf[12..16].try_into().unwrap());
seqs.push(seq);
}
}
let ts = TimestampFields::default();
let reply2 = echo_reply_packet(TOKEN, seqs[2], &params, &ts, None);
socket.send_to(&reply2, peer).unwrap();
thread::sleep(Duration::from_millis(10));
let reply0 = echo_reply_packet(TOKEN, seqs[0], &params, &ts, None);
socket.send_to(&reply0, peer).unwrap();
});
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe().unwrap();
client.send_probe().unwrap();
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(50));
let ev1 = client.recv_once().unwrap();
assert!(matches!(&ev1[0], ClientEvent::EchoReply { seq: 2, .. }));
thread::sleep(Duration::from_millis(30));
let ev2 = client.recv_once().unwrap();
match &ev2[0] {
ClientEvent::LateReply {
seq, rtt, sent_at, ..
} => {
assert_eq!(*seq, 0);
assert!(rtt.is_some());
assert!(sent_at.is_some());
}
other => panic!("expected LateReply, got {other:?}"),
}
server.join();
}
// ---------- Regression tests ----------
#[test]
@@ -1393,12 +1015,6 @@ fn unmatched_future_reply_does_not_update_highest_received_seq() {
let ev_future = client.recv_once().unwrap();
assert!(matches!(&ev_future[0], ClientEvent::Warning { .. }));
assert_eq!(
client.session.as_ref().unwrap().highest_received_seq,
Some(0),
"highest_received_seq should not be updated by unmatched future reply"
);
thread::sleep(Duration::from_millis(30));
let ev1 = client.recv_once().unwrap();
assert!(
@@ -1562,95 +1178,6 @@ fn process_received_packet_uses_supplied_receive_metadata() {
server.join();
}
#[test]
fn process_received_packet_uses_supplied_receive_metadata_for_late_reply() {
let params = default_params();
let server = silent_open_server(params.clone());
let config = ClientConfig {
probe_timeout: Duration::from_millis(40),
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.send_probe_at(ClientTimestamp::now()).unwrap();
thread::sleep(Duration::from_millis(60));
let losses = client.poll_timeouts().unwrap();
assert!(matches!(&losses[0], ClientEvent::EchoLoss { seq: 0, .. }));
let recv_ts = ClientTimestamp::now();
let reply = echo_reply_packet(TOKEN, 0, &params, &TimestampFields::default(), None);
let events = client
.process_received_packet(
&reply,
recv_ts,
ReceiveMeta {
traffic_class: Some(0),
kernel_rx_timestamp: Some(SystemTime::UNIX_EPOCH + Duration::new(56, 78)),
},
)
.unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::LateReply { packet_meta, .. } => {
assert_eq!(packet_meta.traffic_class, Some(0));
assert_eq!(packet_meta.dscp, Some(0));
assert_eq!(packet_meta.ecn, Some(0));
assert_eq!(
packet_meta.kernel_rx_timestamp,
Some(SystemTime::UNIX_EPOCH + Duration::new(56, 78))
);
}
other => panic!("expected LateReply, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
fn receive_metadata_does_not_broaden_malformed_warning() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let events = client
.process_received_packet(
b"not an irtt reply",
ClientTimestamp::now(),
ReceiveMeta {
traffic_class: Some(184),
kernel_rx_timestamp: None,
},
)
.unwrap();
assert_eq!(events.len(), 1);
match &events[0] {
ClientEvent::Warning { kind, message, .. } => {
assert_eq!(*kind, WarningKind::MalformedOrUnrelatedPacket);
assert_eq!(message, "dropped malformed or unrelated packet");
}
other => panic!("expected Warning, got {other:?}"),
}
client.close().unwrap();
server.join();
}
#[test]
fn recv_available_drains_burst_replies() {
let params = default_params();
@@ -1704,129 +1231,6 @@ fn recv_available_respects_packet_budget() {
server.join();
}
#[test]
fn send_probe_wraps_wire_sequence_at_u32_max() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let session = client.session.as_mut().unwrap();
session.next_wire_seq = u32::MAX;
client.send_probe().unwrap();
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(30));
let packets: Vec<_> = server.rx.try_iter().collect();
let seqs: Vec<u32> = packets
.iter()
.filter(|p| p.len() >= 16 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0)
.map(|p| u32::from_le_bytes(p[12..16].try_into().unwrap()))
.collect();
assert_eq!(seqs, vec![u32::MAX, 0]);
client.close().unwrap();
server.join();
}
#[test]
fn wrapped_reply_after_u32_max_is_not_late() {
let params = default_params();
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
socket
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
let mut seqs = Vec::new();
for _ in 0..2 {
let mut buf = [0_u8; 2048];
let (size, _) = socket.recv_from(&mut buf).unwrap();
tx.send(buf[..size].to_vec()).unwrap();
seqs.push(u32::from_le_bytes(buf[12..16].try_into().unwrap()));
}
assert_eq!(seqs, vec![u32::MAX, 0]);
let ts = TimestampFields::default();
let reply_max = echo_reply_packet(TOKEN, u32::MAX, &params, &ts, None);
socket.send_to(&reply_max, peer).unwrap();
thread::sleep(Duration::from_millis(10));
let reply_zero = echo_reply_packet(TOKEN, 0, &params, &ts, None);
socket.send_to(&reply_zero, peer).unwrap();
});
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let session = client.session.as_mut().unwrap();
session.next_wire_seq = u32::MAX;
client.send_probe().unwrap();
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(50));
let first = client.recv_once().unwrap();
assert!(matches!(
&first[0],
ClientEvent::EchoReply { seq: u32::MAX, .. }
));
thread::sleep(Duration::from_millis(30));
let second = client.recv_once().unwrap();
assert!(
matches!(&second[0], ClientEvent::EchoReply { seq: 0, .. }),
"freshly wrapped reply should not be late, got {:?}",
second[0]
);
server.join();
}
#[test]
fn send_probe_after_sending_done_is_noop() {
let params = default_params();
let server = silent_open_server(params);
let config = ClientConfig {
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(50)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
client.session.as_mut().unwrap().sending_done = true;
let events = client.send_probe().unwrap();
assert!(events.is_empty());
assert_eq!(client.session.as_ref().unwrap().packets_sent, 0);
thread::sleep(Duration::from_millis(30));
let packets: Vec<_> = server.rx.try_iter().collect();
let echo_count = packets
.iter()
.filter(|p| p.len() >= 4 && p[3] & FLAG_OPEN == 0 && p[3] & flags::FLAG_CLOSE == 0)
.count();
assert_eq!(echo_count, 0);
client.close().unwrap();
server.join();
}
#[test]
fn compute_one_way_returns_available_direction_samples() {
let sent_at = ClientTimestamp {
@@ -1848,56 +1252,6 @@ fn compute_one_way_returns_available_direction_samples() {
assert_eq!(sample.server_to_client, Some(Duration::from_millis(15)));
}
#[test]
fn recv_buffer_uses_negotiated_packet_length() {
let params = Params {
protocol_version: 1,
duration_ns: 3_000_000_000,
interval_ns: 1_000_000_000,
length: 4096,
received_stats: ReceivedStats::Both,
stamp_at: StampAt::Both,
clock: Clock::Both,
..Params::default()
};
let server = start_fake_server(move |socket, tx| {
let (_, peer) = recv_request(&socket, &tx);
let reply = open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params, None);
socket.send_to(&reply, peer).unwrap();
socket
.set_read_timeout(Some(Duration::from_secs(2)))
.unwrap();
loop {
let mut buf = [0_u8; 8192];
match socket.recv_from(&mut buf) {
Ok((size, _)) => {
tx.send(buf[..size].to_vec()).unwrap();
}
Err(_) => break,
}
}
});
let config = ClientConfig {
length: 4096,
socket_config: crate::SocketConfig {
recv_timeout: Some(Duration::from_millis(200)),
..Default::default()
},
..default_test_config(server.addr)
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
let buf_size = client.recv_buffer_size();
assert!(
buf_size > 4096,
"recv buffer should include truncation-detection byte, got {buf_size}"
);
assert_eq!(buf_size, 4097);
assert_eq!(client.recv_buffer.len(), buf_size);
client.close().unwrap();
server.join();
}
#[test]
fn short_echo_reply_does_not_emit_echo_reply() {
let params = Params {
@@ -1997,6 +1351,10 @@ fn overlong_datagram_detection_uses_extra_receive_byte() {
};
let mut client = Client::connect(config).unwrap();
assert_open_started(client.open().unwrap());
assert_eq!(
client.recv_buffer_size(),
echo_packet_len(false, &params) + 1
);
assert_eq!(
client.recv_buffer.len(),
echo_packet_len(false, &params) + 1
@@ -2054,25 +1412,3 @@ fn exact_length_echo_reply_still_emits_echo_reply() {
client.close().unwrap();
server.join();
}
#[test]
fn recv_once_at_test_helper_provides_deterministic_timestamp() {
let params = default_params();
let (mut client, server) = open_client_with_echo_server(&params);
client.send_probe().unwrap();
thread::sleep(Duration::from_millis(50));
let fixed_ts = ClientTimestamp {
mono: Instant::now(),
wall: SystemTime::now(),
};
let events = client.recv_once_at(fixed_ts).unwrap();
assert_eq!(events.len(), 1);
if let ClientEvent::EchoReply { received_at, .. } = &events[0] {
assert_eq!(*received_at, fixed_ts);
} else {
panic!("expected EchoReply");
}
client.close().unwrap();
server.join();
}
crates/irtt-client/src/client/tests/sequence.rs
+5 -22
View File
@@ -1,10 +1,4 @@
use super::*;
use std::time::Instant;
use crate::{
probe::{CompletedSet, PendingMap, TimedOutMap},
session::ActiveSession,
};
#[test]
fn sequence_ordering_handles_normal_and_wrapped_values() {
@@ -24,21 +18,10 @@ fn sequence_ordering_handles_normal_and_wrapped_values() {
#[test]
fn highest_received_seq_updates_across_wrap() {
let mut session = ActiveSession {
next_wire_seq: 0,
highest_received_seq: Some(u32::MAX),
packets_sent: 0,
start_mono: Instant::now(),
end_mono: None,
next_send_at: Instant::now(),
pending: PendingMap::new(8),
timed_out: TimedOutMap::new(8),
completed: CompletedSet::new(8),
sending_done: false,
};
let mut highest = Some(u32::MAX);
update_highest_received(&mut session, 0);
assert_eq!(session.highest_received_seq, Some(0));
update_highest_received(&mut session, u32::MAX);
assert_eq!(session.highest_received_seq, Some(0));
update_highest_received(&mut highest, 0);
assert_eq!(highest, Some(0));
update_highest_received(&mut highest, u32::MAX);
assert_eq!(highest, Some(0));
}