From adfe8e7a6bb21d0e5a60a8ba5a773dc4003fbfa5 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Wed, 13 May 2026 11:59:06 +0700 Subject: [PATCH] test(irtt-stats): move integration tests to dedicated test module --- crates/irtt-stats/src/lib.rs | 400 -------------------------- crates/irtt-stats/tests/common/mod.rs | 114 ++++++++ crates/irtt-stats/tests/stats.rs | 270 +++++++++++++++++ 3 files changed, 384 insertions(+), 400 deletions(-) create mode 100644 crates/irtt-stats/tests/common/mod.rs create mode 100644 crates/irtt-stats/tests/stats.rs diff --git a/crates/irtt-stats/src/lib.rs b/crates/irtt-stats/src/lib.rs index f800068..d8335f2 100644 --- a/crates/irtt-stats/src/lib.rs +++ b/crates/irtt-stats/src/lib.rs @@ -119,16 +119,6 @@ impl StatsCollector { pub fn summary(&self) -> FiniteSummary { self.cumulative() } - - #[cfg(test)] - fn retained_median_samples(&self) -> usize { - self.cumulative.retained_median_samples() - } - - #[cfg(test)] - fn retained_sequence_samples(&self) -> usize { - self.cumulative.retained_sequence_samples() - } } pub type RollingSnapshot = CumulativeSnapshot; @@ -499,23 +489,6 @@ impl CoreStats { }, } } - - #[cfg(test)] - fn retained_median_samples(&self) -> usize { - self.rtt_primary.retained_samples() - + self.rtt_raw.retained_samples() - + self.rtt_adjusted.retained_samples() - + self.ipdv_round_trip.retained_samples() - + self.ipdv_send.retained_samples() - + self.ipdv_receive.retained_samples() - + self.send_delay.retained_samples() - + self.receive_delay.retained_samples() - } - - #[cfg(test)] - fn retained_sequence_samples(&self) -> usize { - self.samples.len() - } } #[derive(Debug, Clone, PartialEq)] @@ -731,11 +704,6 @@ impl MetricU64 { .and_then(|samples| median_u64(samples)), } } - - #[cfg(test)] - fn retained_samples(&self) -> usize { - self.samples.as_ref().map_or(0, Vec::len) - } } #[derive(Debug, Clone, PartialEq)] @@ -772,11 +740,6 @@ impl MetricI128 { .and_then(|samples| median_i128(samples)), } } - - #[cfg(test)] - fn retained_samples(&self) -> usize { - self.samples.as_ref().map_or(0, Vec::len) - } } #[derive(Debug, Clone, PartialEq, Default)] @@ -1041,113 +1004,8 @@ fn abs_i128_to_u64(value: i128) -> u64 { #[cfg(test)] mod tests { use super::*; - use irtt_client::{ - ClientTimestamp, PacketMeta, ReceivedStatsSample, RttSample, ServerTiming, SignedDuration, - }; use std::time::SystemTime; - fn ts(ms: u64) -> ClientTimestamp { - ClientTimestamp { - mono: Instant::now() + Duration::from_millis(ms), - wall: UNIX_EPOCH + Duration::from_millis(ms), - } - } - - fn rtt(raw_ms: u64, effective_ms: i128) -> RttSample { - RttSample { - raw: Duration::from_millis(raw_ms), - adjusted: u64::try_from(effective_ms).ok().map(Duration::from_millis), - effective: u64::try_from(effective_ms) - .ok() - .map(Duration::from_millis) - .unwrap_or_else(|| Duration::from_millis(raw_ms)), - adjusted_signed: Some(SignedDuration { - ns: effective_ms * 1_000_000, - }), - effective_signed: SignedDuration { - ns: effective_ms * 1_000_000, - }, - } - } - - fn sent(seq: u32, sent_at: ClientTimestamp) -> ClientEvent { - ClientEvent::EchoSent { - seq, - remote: "127.0.0.1:2112".parse().unwrap(), - scheduled_at: sent_at.mono, - sent_at, - bytes: 32, - send_call: Duration::from_micros(10), - timer_error: Duration::from_micros(2), - } - } - - fn reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent { - let sent_at = ts(seq as u64 * 10); - let received_at = ClientTimestamp { - mono: sent_at.mono + Duration::from_millis(raw_ms), - wall: sent_at.wall + Duration::from_millis(raw_ms), - }; - ClientEvent::EchoReply { - seq, - remote: "127.0.0.1:2112".parse().unwrap(), - sent_at, - received_at, - rtt: rtt(raw_ms, effective_ms), - server_timing: Some(ServerTiming { - receive_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 1_000_000), - receive_mono_ns: Some(seq as i64 * 10_000_000 + 1_000_000), - send_wall_ns: Some(system_time_ns(sent_at.wall).unwrap() as i64 + 2_000_000), - send_mono_ns: Some(seq as i64 * 10_000_000 + 2_000_000), - midpoint_wall_ns: None, - midpoint_mono_ns: None, - processing: Some(Duration::from_millis(1)), - }), - one_way: Some(OneWayDelaySample { - client_to_server: Some(Duration::from_millis(1)), - server_to_client: Some(Duration::from_millis(2)), - }), - received_stats: Some(ReceivedStatsSample { - count: Some(seq + 1), - window: Some(0xff), - }), - bytes: 64, - packet_meta: PacketMeta::default(), - } - } - - fn late_reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent { - let ClientEvent::EchoReply { - seq, - remote, - sent_at, - received_at, - rtt, - server_timing, - one_way, - received_stats, - bytes, - packet_meta, - .. - } = reply(seq, raw_ms, effective_ms) - else { - unreachable!(); - }; - ClientEvent::LateReply { - seq, - highest_seen: seq + 1, - remote, - sent_at: Some(sent_at), - received_at, - rtt: Some(rtt), - server_timing, - one_way, - received_stats, - bytes, - packet_meta, - } - } - #[test] fn running_duration_stats_use_sample_variance() { let mut metric = MetricU64::new(false); @@ -1172,264 +1030,6 @@ mod tests { assert_eq!(median_i128(&[-5, 1, 3, 7]), Some(2.0)); } - #[test] - fn disabled_median_avoids_finite_retention() { - let mut collector = StatsCollector::new(StatsConfig::continuous()); - collector.process(&reply(0, 10, 9)); - collector.process(&reply(1, 20, 19)); - assert_eq!(collector.retained_median_samples(), 0); - assert_eq!(collector.cumulative().rtt.primary.median_ns, None); - } - - #[test] - fn continuous_mode_bounds_sequence_tracking() { - let mut collector = StatsCollector::new(StatsConfig::continuous()); - for seq in 0..(CONTINUOUS_SEQUENCE_LIMIT as u32 + 8) { - collector.process(&reply(seq, 10, 10)); - } - - assert_eq!(collector.retained_median_samples(), 0); - assert!(collector.retained_sequence_samples() <= CONTINUOUS_SEQUENCE_LIMIT); - assert_eq!( - collector.cumulative().rtt.primary.stats.count, - CONTINUOUS_SEQUENCE_LIMIT as u64 + 8 - ); - } - - #[test] - fn cumulative_rtt_uses_signed_effective_and_tracks_raw() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&reply(0, 1, -2)); - collector.process(&reply(1, 10, 8)); - - let snapshot = collector.cumulative(); - assert_eq!(snapshot.rtt.primary.stats.count, 2); - assert_eq!(snapshot.rtt.primary.stats.min_ns, Some(-2_000_000)); - assert_eq!(snapshot.rtt.primary.median_ns, Some(3_000_000.0)); - assert_eq!(snapshot.rtt.raw.stats.total_ns, 11_000_000); - } - - #[test] - fn late_unique_counts_and_duplicates_do_not_update_measurements() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&sent(0, ts(0))); - collector.process(&sent(1, ts(10))); - collector.process(&reply(1, 10, 9)); - collector.process(&late_reply(0, 20, 19)); - collector.process(&ClientEvent::DuplicateReply { - seq: 0, - remote: "127.0.0.1:2112".parse().unwrap(), - received_at: ts(50), - bytes: 64, - }); - - let snapshot = collector.cumulative(); - assert_eq!(snapshot.packets.packets_sent, 2); - assert_eq!(snapshot.packets.packets_received, 3); - assert_eq!(snapshot.packets.unique_replies, 2); - assert_eq!(snapshot.packets.duplicates, 1); - assert_eq!(snapshot.packets.late_packets, 1); - assert_eq!(snapshot.rtt.primary.stats.count, 2); - assert_eq!(snapshot.loss.lost_packets, 0); - assert_eq!(snapshot.loss.duplicate_percent, 100.0 / 3.0); - } - - #[test] - fn final_loss_uses_sent_minus_unique_replies_not_echo_loss_events() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&sent(0, ts(0))); - collector.process(&sent(1, ts(10))); - collector.process(&ClientEvent::EchoLoss { - seq: 0, - sent_at: ts(0), - timeout_at: Instant::now(), - }); - collector.process(&late_reply(0, 10, 9)); - - let snapshot = collector.summary(); - assert_eq!(snapshot.events.loss_events, 1); - assert_eq!(snapshot.packets.unique_replies, 1); - assert_eq!(snapshot.loss.lost_packets, 1); - assert_eq!(snapshot.loss.packet_loss_percent, 50.0); - } - - fn assert_no_ipdv_pairs(update: &EventStatsUpdate) { - assert!(update.ipdv_pairs.is_empty(), "{update:?}"); - } - - fn assert_one_ipdv_pair( - update: &EventStatsUpdate, - previous_seq: u32, - current_seq: u32, - rtt_ipdv: Duration, - ) -> &IpdvPairUpdate { - assert_eq!(update.ipdv_pairs.len(), 1, "{update:?}"); - let pair = &update.ipdv_pairs[0]; - assert_eq!(pair.previous_seq, previous_seq); - assert_eq!(pair.current_seq, current_seq); - assert_eq!(pair.rtt_ipdv, rtt_ipdv); - pair - } - - #[test] - fn ipdv_is_sequence_adjacent_and_gap_preserving() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - let first = collector.process(&reply(0, 10, 10)); - let gap = collector.process(&reply(2, 15, 15)); - let adjacent = collector.process(&reply(3, 12, 12)); - - let snapshot = collector.cumulative(); - assert!(first.contributed_sample); - assert_no_ipdv_pairs(&first); - - assert!(gap.contributed_sample); - assert_no_ipdv_pairs(&gap); - - assert!(adjacent.contributed_sample); - assert_one_ipdv_pair(&adjacent, 2, 3, Duration::from_millis(3)); - assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); - assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 3_000_000); - } - - #[test] - fn late_reply_can_complete_ipdv_pair() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&reply(1, 20, 20)); - let update = collector.process(&late_reply(0, 10, 10)); - - let snapshot = collector.cumulative(); - - assert!(update.contributed_sample); - assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(10)); - - assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); - assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); - } - - #[test] - fn update_exposes_directional_ipdv_when_available() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&reply(0, 10, 10)); - let update = collector.process(&reply(1, 13, 13)); - - assert!(update.contributed_sample); - - let pair = assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(3)); - assert!(pair.send_ipdv.is_some()); - assert!(pair.receive_ipdv.is_some()); - } - - #[test] - fn gap_fill_update_exposes_both_completed_ipdv_pairs() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - - let first = collector.process(&reply(0, 10, 10)); - let gap = collector.process(&reply(2, 20, 20)); - let fill = collector.process(&reply(1, 13, 13)); - - assert!(first.contributed_sample); - assert!(first.ipdv_pairs.is_empty()); - - assert!(gap.contributed_sample); - assert!(gap.ipdv_pairs.is_empty()); - - assert!(fill.contributed_sample); - assert_eq!(fill.ipdv_pairs.len(), 2); - - assert_eq!(fill.ipdv_pairs[0].previous_seq, 0); - assert_eq!(fill.ipdv_pairs[0].current_seq, 1); - assert_eq!(fill.ipdv_pairs[0].rtt_ipdv, Duration::from_millis(3)); - - assert_eq!(fill.ipdv_pairs[1].previous_seq, 1); - assert_eq!(fill.ipdv_pairs[1].current_seq, 2); - assert_eq!(fill.ipdv_pairs[1].rtt_ipdv, Duration::from_millis(7)); - - let snapshot = collector.cumulative(); - assert_eq!(snapshot.ipdv.round_trip.stats.count, 2); - assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); - } - - #[test] - fn server_processing_and_one_way_require_available_samples() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&reply(0, 10, 9)); - collector.process(&ClientEvent::LateReply { - seq: 9, - highest_seen: 10, - remote: "127.0.0.1:2112".parse().unwrap(), - sent_at: None, - received_at: ts(100), - rtt: None, - server_timing: None, - one_way: None, - received_stats: None, - bytes: 64, - packet_meta: PacketMeta::default(), - }); - - let snapshot = collector.cumulative(); - assert_eq!(snapshot.server_processing.processing.count, 1); - assert_eq!(snapshot.one_way_delay.send_delay.stats.count, 1); - assert_eq!(snapshot.events.untracked_late_replies, 1); - } - - #[test] - fn rolling_count_eviction_recomputes_from_bounded_events() { - let mut collector = StatsCollector::new(StatsConfig { - median: MedianMode::Disabled, - rolling_count: Some(2), - rolling_time: None, - }); - collector.process(&sent(0, ts(0))); - collector.process(&reply(0, 10, 10)); - collector.process(&reply(1, 20, 20)); - - let rolling = collector.rolling_count().unwrap(); - assert_eq!(rolling.packets.packets_sent, 0); - assert_eq!(rolling.packets.unique_replies, 2); - assert_eq!(rolling.rtt.primary.stats.count, 2); - } - - #[test] - fn rolling_time_eviction_uses_event_timestamps() { - let mut collector = StatsCollector::new(StatsConfig { - median: MedianMode::Disabled, - rolling_count: None, - rolling_time: Some(Duration::from_millis(15)), - }); - collector.process(&sent(0, ts(0))); - collector.process(&sent(1, ts(10))); - collector.process(&sent(2, ts(30))); - - let rolling = collector.rolling_time().unwrap(); - assert_eq!(rolling.packets.packets_sent, 1); - } - - #[test] - fn empty_and_all_lost_edges_are_defined() { - let empty = StatsCollector::new(StatsConfig::finite()).summary(); - assert_eq!(empty.loss.packet_loss_percent, 0.0); - - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&sent(0, ts(0))); - let all_lost = collector.summary(); - assert_eq!(all_lost.loss.lost_packets, 1); - assert_eq!(all_lost.loss.packet_loss_percent, 100.0); - } - - #[test] - fn directional_loss_uses_server_received_count_when_available() { - let mut collector = StatsCollector::new(StatsConfig::finite()); - collector.process(&sent(0, ts(0))); - collector.process(&sent(1, ts(10))); - collector.process(&reply(0, 10, 10)); - - let loss = collector.summary().loss; - assert_eq!(loss.upstream_loss_packets, Some(1)); - assert_eq!(loss.downstream_loss_packets, Some(0)); - assert_eq!(loss.upstream_loss_percent, 50.0); - } - #[test] fn single_sample_stddev_is_zero() { let mut metric = MetricU64::new(false); diff --git a/crates/irtt-stats/tests/common/mod.rs b/crates/irtt-stats/tests/common/mod.rs new file mode 100644 index 0000000..3539f43 --- /dev/null +++ b/crates/irtt-stats/tests/common/mod.rs @@ -0,0 +1,114 @@ +use std::time::{Duration, Instant, UNIX_EPOCH}; + +use irtt_client::{ + ClientEvent, ClientTimestamp, OneWayDelaySample, PacketMeta, ReceivedStatsSample, RttSample, + ServerTiming, SignedDuration, +}; + +pub fn ts(ms: u64) -> ClientTimestamp { + ClientTimestamp { + mono: Instant::now() + Duration::from_millis(ms), + wall: UNIX_EPOCH + Duration::from_millis(ms), + } +} + +pub fn rtt(raw_ms: u64, effective_ms: i128) -> RttSample { + RttSample { + raw: Duration::from_millis(raw_ms), + adjusted: u64::try_from(effective_ms).ok().map(Duration::from_millis), + effective: u64::try_from(effective_ms) + .ok() + .map(Duration::from_millis) + .unwrap_or_else(|| Duration::from_millis(raw_ms)), + adjusted_signed: Some(SignedDuration { + ns: effective_ms * 1_000_000, + }), + effective_signed: SignedDuration { + ns: effective_ms * 1_000_000, + }, + } +} + +pub fn sent(seq: u32, sent_at: ClientTimestamp) -> ClientEvent { + ClientEvent::EchoSent { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + scheduled_at: sent_at.mono, + sent_at, + bytes: 32, + send_call: Duration::from_micros(10), + timer_error: Duration::from_micros(2), + } +} + +pub fn reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent { + let sent_at = ts(seq as u64 * 10); + let received_at = ClientTimestamp { + mono: sent_at.mono + Duration::from_millis(raw_ms), + wall: sent_at.wall + Duration::from_millis(raw_ms), + }; + ClientEvent::EchoReply { + seq, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at, + received_at, + rtt: rtt(raw_ms, effective_ms), + server_timing: Some(ServerTiming { + receive_wall_ns: Some(system_time_ns(sent_at.wall) as i64 + 1_000_000), + receive_mono_ns: Some(seq as i64 * 10_000_000 + 1_000_000), + send_wall_ns: Some(system_time_ns(sent_at.wall) as i64 + 2_000_000), + send_mono_ns: Some(seq as i64 * 10_000_000 + 2_000_000), + midpoint_wall_ns: None, + midpoint_mono_ns: None, + processing: Some(Duration::from_millis(1)), + }), + one_way: Some(OneWayDelaySample { + client_to_server: Some(Duration::from_millis(1)), + server_to_client: Some(Duration::from_millis(2)), + }), + received_stats: Some(ReceivedStatsSample { + count: Some(seq + 1), + window: Some(0xff), + }), + bytes: 64, + packet_meta: PacketMeta::default(), + } +} + +pub fn late_reply(seq: u32, raw_ms: u64, effective_ms: i128) -> ClientEvent { + let ClientEvent::EchoReply { + seq, + remote, + sent_at, + received_at, + rtt, + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + .. + } = reply(seq, raw_ms, effective_ms) + else { + unreachable!(); + }; + ClientEvent::LateReply { + seq, + highest_seen: seq + 1, + remote, + sent_at: Some(sent_at), + received_at, + rtt: Some(rtt), + server_timing, + one_way, + received_stats, + bytes, + packet_meta, + } +} + +fn system_time_ns(time: std::time::SystemTime) -> i128 { + time.duration_since(UNIX_EPOCH) + .map(|duration| duration.as_nanos() as i128) + .unwrap() +} diff --git a/crates/irtt-stats/tests/stats.rs b/crates/irtt-stats/tests/stats.rs new file mode 100644 index 0000000..060b119 --- /dev/null +++ b/crates/irtt-stats/tests/stats.rs @@ -0,0 +1,270 @@ +mod common; + +use std::time::{Duration, Instant}; + +use common::{late_reply, reply, sent, ts}; +use irtt_client::{ClientEvent, PacketMeta}; +use irtt_stats::{EventStatsUpdate, IpdvPairUpdate, MedianMode, StatsCollector, StatsConfig}; + +#[test] +fn disabled_median_avoids_finite_retention() { + let mut collector = StatsCollector::new(StatsConfig::continuous()); + collector.process(&reply(0, 10, 9)); + collector.process(&reply(1, 20, 19)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.rtt.primary.median_ns, None); + assert_eq!(snapshot.rtt.raw.median_ns, None); + assert_eq!(snapshot.rtt.adjusted.median_ns, None); + assert_eq!(snapshot.ipdv.round_trip.median_ns, None); + assert_eq!(snapshot.one_way_delay.send_delay.median_ns, None); + assert_eq!(snapshot.one_way_delay.receive_delay.median_ns, None); +} + +#[test] +fn continuous_mode_bounds_sequence_tracking() { + let mut collector = StatsCollector::new(StatsConfig::continuous()); + for seq in 0..4104 { + collector.process(&reply(seq, 10, 10)); + } + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.rtt.primary.stats.count, 4104); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 4103); + assert_eq!(snapshot.rtt.primary.median_ns, None); + assert_eq!(snapshot.ipdv.round_trip.median_ns, None); +} + +#[test] +fn cumulative_rtt_uses_signed_effective_and_tracks_raw() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 1, -2)); + collector.process(&reply(1, 10, 8)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.rtt.primary.stats.count, 2); + assert_eq!(snapshot.rtt.primary.stats.min_ns, Some(-2_000_000)); + assert_eq!(snapshot.rtt.primary.median_ns, Some(3_000_000.0)); + assert_eq!(snapshot.rtt.raw.stats.total_ns, 11_000_000); +} + +#[test] +fn late_unique_counts_and_duplicates_do_not_update_measurements() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, ts(0))); + collector.process(&sent(1, ts(10))); + collector.process(&reply(1, 10, 9)); + collector.process(&late_reply(0, 20, 19)); + collector.process(&ClientEvent::DuplicateReply { + seq: 0, + remote: "127.0.0.1:2112".parse().unwrap(), + received_at: ts(50), + bytes: 64, + }); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.packets.packets_sent, 2); + assert_eq!(snapshot.packets.packets_received, 3); + assert_eq!(snapshot.packets.unique_replies, 2); + assert_eq!(snapshot.packets.duplicates, 1); + assert_eq!(snapshot.packets.late_packets, 1); + assert_eq!(snapshot.rtt.primary.stats.count, 2); + assert_eq!(snapshot.loss.lost_packets, 0); + assert_eq!(snapshot.loss.duplicate_percent, 100.0 / 3.0); +} + +#[test] +fn final_loss_uses_sent_minus_unique_replies_not_echo_loss_events() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, ts(0))); + collector.process(&sent(1, ts(10))); + collector.process(&ClientEvent::EchoLoss { + seq: 0, + sent_at: ts(0), + timeout_at: Instant::now(), + }); + collector.process(&late_reply(0, 10, 9)); + + let snapshot = collector.summary(); + assert_eq!(snapshot.events.loss_events, 1); + assert_eq!(snapshot.packets.unique_replies, 1); + assert_eq!(snapshot.loss.lost_packets, 1); + assert_eq!(snapshot.loss.packet_loss_percent, 50.0); +} + +fn assert_no_ipdv_pairs(update: &EventStatsUpdate) { + assert!(update.ipdv_pairs.is_empty(), "{update:?}"); +} + +fn assert_one_ipdv_pair( + update: &EventStatsUpdate, + previous_seq: u32, + current_seq: u32, + rtt_ipdv: Duration, +) -> &IpdvPairUpdate { + assert_eq!(update.ipdv_pairs.len(), 1, "{update:?}"); + let pair = &update.ipdv_pairs[0]; + assert_eq!(pair.previous_seq, previous_seq); + assert_eq!(pair.current_seq, current_seq); + assert_eq!(pair.rtt_ipdv, rtt_ipdv); + pair +} + +#[test] +fn ipdv_is_sequence_adjacent_and_gap_preserving() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + let first = collector.process(&reply(0, 10, 10)); + let gap = collector.process(&reply(2, 15, 15)); + let adjacent = collector.process(&reply(3, 12, 12)); + + let snapshot = collector.cumulative(); + assert!(first.contributed_sample); + assert_no_ipdv_pairs(&first); + + assert!(gap.contributed_sample); + assert_no_ipdv_pairs(&gap); + + assert!(adjacent.contributed_sample); + assert_one_ipdv_pair(&adjacent, 2, 3, Duration::from_millis(3)); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 3_000_000); +} + +#[test] +fn late_reply_can_complete_ipdv_pair() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(1, 20, 20)); + let update = collector.process(&late_reply(0, 10, 10)); + + let snapshot = collector.cumulative(); + + assert!(update.contributed_sample); + assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(10)); + + assert_eq!(snapshot.ipdv.round_trip.stats.count, 1); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); +} + +#[test] +fn update_exposes_directional_ipdv_when_available() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 10, 10)); + let update = collector.process(&reply(1, 13, 13)); + + assert!(update.contributed_sample); + + let pair = assert_one_ipdv_pair(&update, 0, 1, Duration::from_millis(3)); + assert!(pair.send_ipdv.is_some()); + assert!(pair.receive_ipdv.is_some()); +} + +#[test] +fn gap_fill_update_exposes_both_completed_ipdv_pairs() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + + let first = collector.process(&reply(0, 10, 10)); + let gap = collector.process(&reply(2, 20, 20)); + let fill = collector.process(&reply(1, 13, 13)); + + assert!(first.contributed_sample); + assert!(first.ipdv_pairs.is_empty()); + + assert!(gap.contributed_sample); + assert!(gap.ipdv_pairs.is_empty()); + + assert!(fill.contributed_sample); + assert_eq!(fill.ipdv_pairs.len(), 2); + + assert_eq!(fill.ipdv_pairs[0].previous_seq, 0); + assert_eq!(fill.ipdv_pairs[0].current_seq, 1); + assert_eq!(fill.ipdv_pairs[0].rtt_ipdv, Duration::from_millis(3)); + + assert_eq!(fill.ipdv_pairs[1].previous_seq, 1); + assert_eq!(fill.ipdv_pairs[1].current_seq, 2); + assert_eq!(fill.ipdv_pairs[1].rtt_ipdv, Duration::from_millis(7)); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.ipdv.round_trip.stats.count, 2); + assert_eq!(snapshot.ipdv.round_trip.stats.total_ns, 10_000_000); +} + +#[test] +fn server_processing_and_one_way_require_available_samples() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&reply(0, 10, 9)); + collector.process(&ClientEvent::LateReply { + seq: 9, + highest_seen: 10, + remote: "127.0.0.1:2112".parse().unwrap(), + sent_at: None, + received_at: ts(100), + rtt: None, + server_timing: None, + one_way: None, + received_stats: None, + bytes: 64, + packet_meta: PacketMeta::default(), + }); + + let snapshot = collector.cumulative(); + assert_eq!(snapshot.server_processing.processing.count, 1); + assert_eq!(snapshot.one_way_delay.send_delay.stats.count, 1); + assert_eq!(snapshot.events.untracked_late_replies, 1); +} + +#[test] +fn rolling_count_eviction_recomputes_from_bounded_events() { + let mut collector = StatsCollector::new(StatsConfig { + median: MedianMode::Disabled, + rolling_count: Some(2), + rolling_time: None, + }); + collector.process(&sent(0, ts(0))); + collector.process(&reply(0, 10, 10)); + collector.process(&reply(1, 20, 20)); + + let rolling = collector.rolling_count().unwrap(); + assert_eq!(rolling.packets.packets_sent, 0); + assert_eq!(rolling.packets.unique_replies, 2); + assert_eq!(rolling.rtt.primary.stats.count, 2); +} + +#[test] +fn rolling_time_eviction_uses_event_timestamps() { + let mut collector = StatsCollector::new(StatsConfig { + median: MedianMode::Disabled, + rolling_count: None, + rolling_time: Some(Duration::from_millis(15)), + }); + collector.process(&sent(0, ts(0))); + collector.process(&sent(1, ts(10))); + collector.process(&sent(2, ts(30))); + + let rolling = collector.rolling_time().unwrap(); + assert_eq!(rolling.packets.packets_sent, 1); +} + +#[test] +fn empty_and_all_lost_edges_are_defined() { + let empty = StatsCollector::new(StatsConfig::finite()).summary(); + assert_eq!(empty.loss.packet_loss_percent, 0.0); + + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, ts(0))); + let all_lost = collector.summary(); + assert_eq!(all_lost.loss.lost_packets, 1); + assert_eq!(all_lost.loss.packet_loss_percent, 100.0); +} + +#[test] +fn directional_loss_uses_server_received_count_when_available() { + let mut collector = StatsCollector::new(StatsConfig::finite()); + collector.process(&sent(0, ts(0))); + collector.process(&sent(1, ts(10))); + collector.process(&reply(0, 10, 10)); + + let loss = collector.summary().loss; + assert_eq!(loss.upstream_loss_packets, Some(1)); + assert_eq!(loss.downstream_loss_packets, Some(0)); + assert_eq!(loss.upstream_loss_percent, 50.0); +}