From bdf5455a3f4faddef8a5e2fbcd0032dcb3077afc Mon Sep 17 00:00:00 2001 From: Keenan Tims Date: Fri, 23 Oct 2026 09:52:57 -0700 Subject: [PATCH] Expand unit tests and fix bugs they found Add unit tests for packet parsing/serialization, interval arithmetic, the session state machine, validation discards, poll/final, the detection watchdog, hook scripts, and socket binding. Fixes: - BfdAuth parsing subtracted the header from Auth Len twice; Auth Len includes the 2-byte header. - A diagnostic code above 9 panicked the session task; map it to Reserved. - script_runner panicked on non-UTF-8 script output; decode lossily. - BfdError::InvalidFieldValue printed field and value swapped. - BfdInterval now panics on values exceeding u32 microseconds instead of silently truncating, and gains checked_mul. Detection time computation saturates rather than panicking on remote-supplied multipliers. --- src/bfd.rs | 316 +++++++++++++++++++++- src/bin/rust-bfdd.rs | 619 ++++++++++++++++++++++++++++++++++++++++++- src/events.rs | 223 +++++++++++++++- src/lib.rs | 116 ++++++++ 4 files changed, 1251 insertions(+), 23 deletions(-) diff --git a/src/bfd.rs b/src/bfd.rs index 8ef5764..6cb6b36 100644 --- a/src/bfd.rs +++ b/src/bfd.rs @@ -102,14 +102,35 @@ impl BfdInterval { pub fn from_micros(micros: u32) -> Self { Self(micros) } + /// # Panics + /// If the interval does not fit in 32 bits of microseconds. pub fn from_millis(millis: u32) -> Self { - Self(millis * 1000) + Self( + millis + .checked_mul(1_000) + .expect("BfdInterval overflow: milliseconds exceed u32 microseconds"), + ) } + /// # Panics + /// If the interval does not fit in 32 bits of microseconds. pub fn from_secs(secs: u32) -> Self { - Self(secs * 1000000) + Self( + secs.checked_mul(1_000_000) + .expect("BfdInterval overflow: seconds exceed u32 microseconds"), + ) } + /// # Panics + /// If the interval is negative, NaN, or does not fit in 32 bits of microseconds. pub fn from_secs_f32(secs: f32) -> Self { - Self((secs * 1000000.0) as u32) + let micros = secs * 1_000_000.0; + assert!( + (0.0..4_294_967_296.0).contains(µs), + "BfdInterval overflow: {secs}s out of range" + ); + Self(micros as u32) + } + pub fn checked_mul(self, rhs: u32) -> Option { + self.0.checked_mul(rhs).map(Self) } } @@ -119,8 +140,13 @@ impl From for time::Duration { } } impl From for BfdInterval { + /// # Panics + /// If the duration does not fit in 32 bits of microseconds. fn from(value: time::Duration) -> Self { - BfdInterval(value.as_micros() as u32) + BfdInterval( + u32::try_from(value.as_micros()) + .expect("BfdInterval overflow: duration exceeds u32 microseconds"), + ) } } impl Display for BfdInterval { @@ -131,13 +157,14 @@ impl Display for BfdInterval { impl Mul for BfdInterval { type Output = BfdInterval; fn mul(self, rhs: u32) -> Self::Output { - BfdInterval(self.0 * rhs) + self.checked_mul(rhs) + .expect("BfdInterval overflow in multiplication") } } impl Mul for u32 { type Output = BfdInterval; fn mul(self, rhs: BfdInterval) -> Self::Output { - BfdInterval(self * rhs.0) + rhs * self } } impl Div for BfdInterval { @@ -216,7 +243,7 @@ impl BfdAuthData { auth_type: BfdAuthType, auth_len: u8, ) -> IResult<&[u8], Self> { - let (new_i, data) = take(auth_len)(i)?; + let (new_i, data) = take(auth_len.saturating_sub(2))(i)?; let (_leftovers, retval) = BfdAuthData::parse_be(data, auth_type)?; Ok((new_i, retval)) } @@ -257,3 +284,278 @@ impl BfdPacket { Ok(Box::new(wtr.into_inner())) } } + +#[cfg(test)] +mod tests { + use super::*; + use nom_derive::Parse; + use std::time::Duration; + + // vers=1 diag=0 | state=Up | mult=3 | len=24, then RFC 5880 6.8.7 fields + const UP_PACKET: [u8; 24] = [ + 0x20, 0xc0, 0x03, 0x18, // flags + 0x00, 0x00, 0x00, 0x01, // my disc + 0x00, 0x00, 0x00, 0x02, // your disc + 0x00, 0x04, 0x93, 0xe0, // desired min tx: 300000us + 0x00, 0x04, 0x93, 0xe0, // required min rx: 300000us + 0x00, 0x00, 0x00, 0x00, // required min echo rx + ]; + + fn parse(bytes: &[u8]) -> BfdPacket { + BfdPacket::parse(bytes).expect("parse").1 + } + + #[test] + fn flags_bit_layout() { + let f = BfdFlags(0) + .with_vers(1) + .with_diag(BfdDiagnostic::NeighborDown as u8) + .with_state(BfdState::Up as u8) + .with_poll(true) + .with_detect_mult(3) + .with_length(24); + assert_eq!(f.0, 0x23e0_0318); + let f = BfdFlags(0) + .with_final_(true) + .with_cpi(true) + .with_auth_present(true) + .with_demand(true) + .with_multipoint(true); + assert_eq!(f.0, 0x001f_0000); + } + + #[test] + fn flags_fields_do_not_bleed() { + let f = BfdFlags(0).with_diag(0x1f).with_state(BfdState::Init as u8); + assert_eq!(f.vers(), 0); + assert_eq!(f.state(), Ok(BfdState::Init)); + assert!(!f.poll() && !f.final_() && !f.demand()); + assert_eq!(f.detect_mult(), 0); + assert_eq!(f.length(), 0); + } + + #[test] + fn parse_known_packet() { + let p = parse(&UP_PACKET); + assert_eq!(p.flags.vers(), 1); + assert_eq!(p.flags.diag(), Ok(BfdDiagnostic::None)); + assert_eq!(p.flags.state(), Ok(BfdState::Up)); + assert_eq!(p.flags.detect_mult(), 3); + assert_eq!(p.flags.length(), 24); + assert!(!p.flags.auth_present()); + assert_eq!(p.my_disc, BfdDiscriminator(1)); + assert_eq!(p.your_disc, BfdDiscriminator(2)); + assert_eq!(p.desired_min_tx, BfdInterval::from_millis(300)); + assert_eq!(p.required_min_rx, BfdInterval::from_millis(300)); + assert_eq!(p.required_min_echo_rx, BfdInterval::from_micros(0)); + assert!(p.auth.is_none()); + } + + #[test] + fn serialize_known_packet() { + assert_eq!(&*parse(&UP_PACKET).serialize().unwrap(), &UP_PACKET); + } + + #[test] + fn serialize_parse_roundtrip() { + let p = BfdPacket { + flags: BfdFlags(0) + .with_vers(1) + .with_diag(BfdDiagnostic::TimeExpired as u8) + .with_state(BfdState::Down as u8) + .with_final_(true) + .with_detect_mult(5) + .with_length(24), + my_disc: BfdDiscriminator(0xdead_beef), + your_disc: BfdDiscriminator(0), + desired_min_tx: BfdInterval::from_secs(1), + required_min_rx: BfdInterval::from_micros(u32::MAX), + required_min_echo_rx: BfdInterval::from_micros(7), + auth: None, + }; + let bytes = p.serialize().unwrap(); + assert_eq!(bytes.len(), 24); + let q = parse(&bytes); + assert_eq!(q.flags.0, p.flags.0); + assert_eq!(q.my_disc, p.my_disc); + assert_eq!(q.your_disc, p.your_disc); + assert_eq!(q.desired_min_tx, p.desired_min_tx); + assert_eq!(q.required_min_rx, p.required_min_rx); + assert_eq!(q.required_min_echo_rx, p.required_min_echo_rx); + } + + #[test] + fn parse_rejects_truncated() { + for len in 0..UP_PACKET.len() { + assert!(BfdPacket::parse(&UP_PACKET[..len]).is_err(), "len {len}"); + } + } + + #[test] + fn parse_ignores_trailing_bytes() { + let mut bytes = UP_PACKET.to_vec(); + bytes.extend_from_slice(&[0xff; 8]); + let (rest, p) = BfdPacket::parse(&bytes).unwrap(); + assert_eq!(rest.len(), 8); + assert_eq!(p.my_disc, BfdDiscriminator(1)); + } + + #[test] + fn parse_auth_present_requires_auth_section() { + let mut bytes = UP_PACKET; + bytes[1] |= 0x04; + assert!(BfdPacket::parse(&bytes).is_err()); + } + + #[test] + fn parse_keyed_md5_auth() { + // RFC 5880 4.3: type, len (incl. type and len bytes) = 24, key id, reserved, seq, 16 byte digest + let mut bytes = UP_PACKET.to_vec(); + bytes[1] |= 0x04; + bytes[3] = 48; + bytes.extend_from_slice(&[2, 24, 7, 0, 0, 0, 0, 9]); + bytes.extend_from_slice(&[0xaa; 16]); + let (rest, p) = BfdPacket::parse(&bytes).expect("parse"); + assert!(rest.is_empty()); + assert!(p.auth.is_some()); + } + + #[test] + fn invalid_state_and_diag_values() { + assert!(BfdState::try_from(4u8).is_err()); + assert!(BfdDiagnostic::try_from(10u8).is_err()); + assert!(BfdState::try_from(3u8).is_ok()); + } + + #[test] + fn state_display_and_osstr() { + for (s, name) in [ + (BfdState::AdminDown, "AdminDown"), + (BfdState::Down, "Down"), + (BfdState::Init, "Init"), + (BfdState::Up, "Up"), + ] { + assert_eq!(s.to_string(), name); + assert_eq!(<&OsStr>::from(s), OsStr::new(name)); + } + assert_eq!(BfdState::default(), BfdState::Down); + } + + #[test] + fn diagnostic_display() { + assert_eq!(BfdDiagnostic::None.to_string(), "None"); + assert_eq!(BfdDiagnostic::TimeExpired.to_string(), "TimeExpired"); + assert_eq!(BfdDiagnostic::AdminDown.to_string(), "AdminDown"); + } + + #[test] + fn interval_constructors() { + assert_eq!( + BfdInterval::from_millis(300), + BfdInterval::from_micros(300_000) + ); + assert_eq!(BfdInterval::from_secs(2), BfdInterval::from_millis(2000)); + assert_eq!( + BfdInterval::from_secs_f32(0.25), + BfdInterval::from_millis(250) + ); + } + + #[test] + fn interval_duration_conversion() { + let i = BfdInterval::from_millis(300); + assert_eq!(Duration::from(i), Duration::from_millis(300)); + assert_eq!(BfdInterval::from(Duration::from_millis(300)), i); + assert_eq!( + BfdInterval::from(Duration::from_micros(1999)), + BfdInterval::from_micros(1999) + ); + } + + #[test] + fn interval_arithmetic_and_display() { + let i = BfdInterval::from_millis(300); + assert_eq!(i * 3, BfdInterval::from_millis(900)); + assert_eq!(3 * i, BfdInterval::from_millis(900)); + assert_eq!(i / 3, BfdInterval::from_millis(100)); + assert_eq!(900_000 / i, BfdInterval::from_micros(3)); + assert_eq!(i.to_string(), "300000us"); + assert!(BfdInterval::from_millis(1) < BfdInterval::from_secs(1)); + } + + #[test] + fn discriminator_display() { + assert_eq!(BfdDiscriminator(42).to_string(), "42"); + } + + #[test] + fn interval_constructors_accept_the_maximum() { + assert_eq!( + BfdInterval::from_millis(4_294_967), + BfdInterval::from_micros(4_294_967_000) + ); + assert_eq!( + BfdInterval::from_secs(4294), + BfdInterval::from_micros(4_294_000_000) + ); + assert_eq!( + BfdInterval::from(Duration::from_micros(u32::MAX as u64)), + BfdInterval::from_micros(u32::MAX) + ); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_millis_overflow_panics() { + BfdInterval::from_millis(4_294_968); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_secs_overflow_panics() { + BfdInterval::from_secs(4295); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_secs_f32_overflow_panics() { + BfdInterval::from_secs_f32(4295.0); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_secs_f32_negative_panics() { + BfdInterval::from_secs_f32(-1.0); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_secs_f32_nan_panics() { + BfdInterval::from_secs_f32(f32::NAN); + } + + #[test] + #[should_panic(expected = "overflow")] + fn from_duration_overflow_panics() { + let _ = BfdInterval::from(Duration::from_micros(u32::MAX as u64 + 1)); + } + + #[test] + #[should_panic(expected = "overflow")] + fn mul_overflow_panics() { + let _ = BfdInterval::from_secs(2000) * 3; + } + + #[test] + #[should_panic(expected = "overflow")] + fn reverse_mul_overflow_panics() { + let _ = 3 * BfdInterval::from_secs(2000); + } + + #[test] + fn checked_mul() { + let i = BfdInterval::from_secs(1000); + assert_eq!(i.checked_mul(4), Some(BfdInterval::from_secs(4000))); + assert_eq!(i.checked_mul(5), None); + } +} diff --git a/src/bin/rust-bfdd.rs b/src/bin/rust-bfdd.rs index 7b478df..c00d2d1 100644 --- a/src/bin/rust-bfdd.rs +++ b/src/bin/rust-bfdd.rs @@ -91,7 +91,7 @@ impl Display for BfdError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::InvalidFieldValue(field, value) => { - write!(f, "invalid value `{}` for field `{}`", field, value) + write!(f, "invalid value `{}` for field `{}`", value, field) } } } @@ -777,7 +777,8 @@ impl BfdSession { // Set bfd.RemoteMinRxInterval to the value of Required Min RX Interval. self.state.set_remote_min_rx_interval(p.required_min_rx); // Not in spec: store the received diagnostic - self.state.set_last_remote_diag(p.flags.diag().unwrap()); + self.state + .set_last_remote_diag(p.flags.diag().unwrap_or(BfdDiagnostic::Reserved)); // If the Required Min Echo RX Interval field is zero, the transmission of Echo packets, if any, MUST cease. if p.required_min_echo_rx == BfdInterval::from_secs(0) { @@ -908,15 +909,25 @@ impl BfdSession { // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.4 async fn update_detection_time(&mut self, p: &BfdPacket) { let old = self.state.detection_time(); + // Remote-supplied values can overflow the 32-bit microsecond range; saturate rather than panic. + let saturating = |mult: u8, interval: BfdInterval| { + interval + .checked_mul(mult as u32) + .unwrap_or(BfdInterval::from_micros(u32::MAX)) + }; self.state.set_detection_time(if !self.state.demand_mode() { - p.flags.detect_mult() as u32 - * std::cmp::max(self.state.required_min_rx_interval(), p.desired_min_tx) + saturating( + p.flags.detect_mult(), + std::cmp::max(self.state.required_min_rx_interval(), p.desired_min_tx), + ) } else { - self.state.detect_mult() as u32 - * std::cmp::max( + saturating( + self.state.detect_mult(), + std::cmp::max( self.state.desired_min_tx_interval(), self.state.remote_min_rx_interval(), - ) + ), + ) }); if old != self.state.detection_time() { info!( @@ -1327,4 +1338,598 @@ mod tests { assert_eq!(args.bind.len(), 2); assert_eq!(args.peers.len(), 2); } + + use std::net::Ipv4Addr; + + const LOCAL: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)); + const PEER: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)); + const LOCAL_DISCR: u32 = 1000; + const REMOTE_DISCR: u32 = 2000; + + #[derive(Debug)] + struct CaptureSink { + tx: mpsc::Sender, + rx: mpsc::Receiver, + out: mpsc::UnboundedSender, + } + impl EventMessageSink for CaptureSink { + fn channel(&self) -> mpsc::Sender { + self.tx.clone() + } + fn run(mut self: Box) -> JoinHandle<()> { + tokio::spawn(async move { + while let Some(EventMessage::StateChange(ev)) = self.rx.recv().await { + let _ = self.out.send(ev); + } + }) + } + } + + async fn capture_dispatcher() -> ( + EventDispatcherHandle, + mpsc::UnboundedReceiver, + ) { + let dispatcher = EventDispatcherHandle::new().await; + let (tx, rx) = mpsc::channel(16); + let (out, events) = mpsc::unbounded_channel(); + dispatcher + .add_sink(Box::new(CaptureSink { tx, rx, out })) + .await; + (dispatcher, events) + } + + fn peer_cfg(desired_ms: u32, required_ms: u32, detect_mult: u8) -> ResolvedPeer { + ResolvedPeer { + address: PEER, + desired_min_tx: BfdInterval::from_millis(desired_ms), + required_min_rx: BfdInterval::from_millis(required_ms), + detect_mult, + } + } + + fn pkt(state: BfdState, my: u32, your: u32) -> BfdPacket { + BfdPacket { + flags: BfdFlags(0) + .with_vers(1) + .with_state(state as u8) + .with_detect_mult(3) + .with_length(24), + my_disc: BfdDiscriminator(my), + your_disc: BfdDiscriminator(your), + desired_min_tx: BfdInterval::from_millis(300), + required_min_rx: BfdInterval::from_millis(300), + required_min_echo_rx: BfdInterval::from_micros(0), + auth: None, + } + } + + fn with_flags(mut p: BfdPacket, f: impl FnOnce(BfdFlags) -> BfdFlags) -> BfdPacket { + p.flags = f(p.flags); + p + } + async fn session( + peer: ResolvedPeer, + ) -> (BfdSession, mpsc::UnboundedReceiver) { + let (dispatcher, events) = capture_dispatcher().await; + let (tx, rx) = mpsc::channel(32); + let sock = UdpSocket::bind((LOCAL, 0)).await.unwrap(); + let s = BfdSession::new( + rx, + BfdSessionHandle { tx }, + sock, + LOCAL, + peer, + BfdDiscriminator(LOCAL_DISCR), + dispatcher, + ); + (s, events) + } + + async fn next_event( + events: &mut mpsc::UnboundedReceiver, + ) -> StateChangeEvent { + time::timeout(Duration::from_secs(2), events.recv()) + .await + .expect("timed out waiting for event") + .expect("event channel closed") + } + + async fn assert_no_event(events: &mut mpsc::UnboundedReceiver) { + time::sleep(Duration::from_millis(50)).await; + assert!(events.try_recv().is_err(), "unexpected state change event"); + } + + async fn session_in( + state: BfdState, + ) -> (BfdSession, mpsc::UnboundedReceiver) { + let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await; + let path: &[BfdState] = match state { + BfdState::Down => &[], + BfdState::Init => &[BfdState::Down], + _ => &[BfdState::Down, BfdState::Init], + }; + for (i, remote) in path.iter().enumerate() { + let your = if i == 0 { 0 } else { LOCAL_DISCR }; + s.receive_control_packet(&pkt(*remote, REMOTE_DISCR, your)) + .await; + next_event(&mut events).await; + } + assert_eq!(s.state.session_state(), state); + (s, events) + } + + #[tokio::test] + async fn new_session_starts_down_and_slow() { + let (s, _events) = session(peer_cfg(300, 200, 3)).await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_eq!(s.state.remote_discr(), BfdDiscriminator(0)); + assert_eq!(s.state.local_diag(), BfdDiagnostic::None); + assert_eq!(s.state.desired_min_tx_interval(), BfdInterval::from_secs(1)); + assert_eq!( + s.state.required_min_rx_interval(), + BfdInterval::from_millis(200) + ); + assert_eq!(s.cur_interval, BfdInterval::from_secs(1)); + assert_eq!(s.state.detect_mult(), 3); + assert!(!s.state.poll_mode()); + } + + #[tokio::test] + async fn new_session_keeps_configured_interval_above_one_second() { + let (s, _events) = session(peer_cfg(2500, 300, 3)).await; + assert_eq!( + s.state.desired_min_tx_interval(), + BfdInterval::from_millis(2500) + ); + assert_eq!(s.effective_desired_min_tx, BfdInterval::from_millis(2500)); + } + + #[tokio::test] + async fn three_way_handshake_and_poll_sequence() { + let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await; + + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0)) + .await; + assert_eq!(s.state.session_state(), BfdState::Init); + assert_eq!(s.state.remote_discr(), BfdDiscriminator(REMOTE_DISCR)); + let ev = next_event(&mut events).await; + assert_eq!( + (ev.from_state, ev.to_state), + (BfdState::Down, BfdState::Init) + ); + assert_eq!(ev.local_discr, BfdDiscriminator(LOCAL_DISCR)); + assert_eq!(ev.remote_discr, BfdDiscriminator(REMOTE_DISCR)); + assert_eq!((ev.local_ip, ev.remote_ip), (LOCAL, PEER)); + + s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Up); + let ev = next_event(&mut events).await; + assert_eq!((ev.from_state, ev.to_state), (BfdState::Init, BfdState::Up)); + assert!(s.state.last_up.load().is_some()); + assert_eq!(s.state.local_diag(), BfdDiagnostic::None); + // Advertise the fast interval now, but keep transmitting slowly until the poll sequence completes + assert_eq!( + s.state.desired_min_tx_interval(), + BfdInterval::from_millis(300) + ); + assert!(s.state.poll_mode()); + assert_eq!(s.cur_interval, BfdInterval::from_secs(1)); + + let fin = with_flags(pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR), |f| { + f.with_final_(true) + }); + s.receive_control_packet(&fin).await; + assert!(!s.state.poll_mode()); + assert_eq!(s.effective_desired_min_tx, BfdInterval::from_millis(300)); + assert_eq!(s.cur_interval, BfdInterval::from_millis(300)); + assert_eq!(s.state.session_state(), BfdState::Up); + } + + #[tokio::test] + async fn up_without_poll_when_configured_interval_is_not_faster() { + let (mut s, _events) = session(peer_cfg(2000, 300, 3)).await; + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0)) + .await; + s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Up); + assert!(!s.state.poll_mode()); + assert_eq!(s.cur_interval, BfdInterval::from_secs(2)); + } + + #[tokio::test] + async fn down_goes_straight_to_up_when_remote_is_init() { + let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await; + s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Up); + let ev = next_event(&mut events).await; + assert_eq!((ev.from_state, ev.to_state), (BfdState::Down, BfdState::Up)); + } + + #[tokio::test] + async fn down_ignores_remote_up_and_admin_down() { + let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await; + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_no_event(&mut events).await; + } + + #[tokio::test] + async fn init_transitions() { + let (mut s, mut events) = session_in(BfdState::Init).await; + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Init); + assert_no_event(&mut events).await; + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Up); + + let (mut s, mut events) = session_in(BfdState::Init).await; + s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown); + let ev = next_event(&mut events).await; + assert_eq!( + (ev.from_state, ev.to_state), + (BfdState::Init, BfdState::Down) + ); + } + #[tokio::test] + async fn up_goes_down_when_remote_goes_down() { + let (mut s, mut events) = session_in(BfdState::Up).await; + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_no_event(&mut events).await; + + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown); + let ev = next_event(&mut events).await; + assert_eq!((ev.from_state, ev.to_state), (BfdState::Up, BfdState::Down)); + assert!(s.state.last_down.load().is_some()); + // Back to the slow rate and no poll sequence in flight + assert_eq!(s.state.desired_min_tx_interval(), BfdInterval::from_secs(1)); + assert_eq!(s.cur_interval, BfdInterval::from_secs(1)); + assert!(!s.state.poll_mode()); + } + + #[tokio::test] + async fn up_goes_down_on_remote_admin_down() { + let (mut s, mut events) = session_in(BfdState::Up).await; + s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown); + assert_eq!(next_event(&mut events).await.to_state, BfdState::Down); + } + + #[tokio::test] + async fn restarted_peer_relearns_discriminator() { + let (mut s, mut events) = session_in(BfdState::Up).await; + s.receive_control_packet(&pkt(BfdState::Down, 3000, 0)) + .await; + assert_eq!(s.state.session_state(), BfdState::Down); + assert_eq!(s.state.remote_discr(), BfdDiscriminator(3000)); + assert_eq!(next_event(&mut events).await.to_state, BfdState::Down); + s.receive_control_packet(&pkt(BfdState::Down, 3000, 0)) + .await; + assert_eq!(s.state.session_state(), BfdState::Init); + } + + #[tokio::test] + async fn up_resets_stats_and_down_records_time() { + let (mut s, _events) = session_in(BfdState::Init).await; + s.state.record_tx(); + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.session_state(), BfdState::Up); + // The Up transition resets the counters, then sends one packet announcing the new state + assert_eq!(s.state.rx_stats.snapshot().count(), 0); + assert_eq!(s.state.tx_stats.snapshot().count(), 1); + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.rx_stats.snapshot().count(), 1); + } + + #[tokio::test] + async fn invalid_packets_are_discarded() { + type Mutation = fn(BfdPacket) -> BfdPacket; + let cases: [(&str, Mutation); 8] = [ + ("version", |p| with_flags(p, |f| f.with_vers(2))), + ("short length", |p| with_flags(p, |f| f.with_length(23))), + ("zero detect mult", |p| { + with_flags(p, |f| f.with_detect_mult(0)) + }), + ("multipoint", |p| with_flags(p, |f| f.with_multipoint(true))), + ("zero my discr", |mut p| { + p.my_disc = BfdDiscriminator(0); + p + }), + ("wrong your discr", |mut p| { + p.your_disc = BfdDiscriminator(LOCAL_DISCR + 1); + p + }), + ("zero your discr, not down", |mut p| { + p.your_disc = BfdDiscriminator(0); + p.flags = p.flags.with_state(BfdState::Up as u8); + p + }), + ("auth without auth in use", |p| { + with_flags(p, |f| f.with_auth_present(true).with_length(26)) + }), + ]; + for (name, mutate) in cases { + let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await; + s.receive_control_packet(&mutate(pkt(BfdState::Down, REMOTE_DISCR, 0))) + .await; + assert_eq!(s.state.session_state(), BfdState::Down, "{name}"); + assert_eq!(s.state.remote_discr(), BfdDiscriminator(0), "{name}"); + assert_eq!( + s.state.remote_min_rx_interval(), + BfdInterval::from_millis(300), + "{name}" + ); + assert_no_event(&mut events).await; + } + } + + #[tokio::test] + async fn auth_length_minimum_is_26() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| { + f.with_auth_present(true).with_length(25) + }); + s.receive_control_packet(&p).await; + assert_eq!(s.state.remote_discr(), BfdDiscriminator(0)); + } + + #[tokio::test] + async fn unknown_diagnostic_does_not_panic() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| f.with_diag(20)); + s.receive_control_packet(&p).await; + assert_eq!(s.state.session_state(), BfdState::Init); + } + + #[tokio::test] + async fn remote_diag_is_recorded() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| { + f.with_diag(BfdDiagnostic::PathDown as u8) + }); + s.receive_control_packet(&p).await; + assert_eq!(s.state.last_remote_diag(), BfdDiagnostic::PathDown); + } + + #[tokio::test] + async fn poll_bit_elicits_final_response() { + let (mut s, _events) = session_in(BfdState::Up).await; + let before = s.state.tx_stats.snapshot().count(); + s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR)) + .await; + assert_eq!(s.state.tx_stats.snapshot().count(), before); + let poll = with_flags(pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR), |f| { + f.with_poll(true) + }); + s.receive_control_packet(&poll).await; + assert_eq!(s.state.tx_stats.snapshot().count(), before + 1); + } + + #[tokio::test] + async fn state_change_sends_immediate_packet() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0)) + .await; + assert_eq!(s.state.tx_stats.snapshot().count(), 1); + s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0)) + .await; + assert_eq!(s.state.tx_stats.snapshot().count(), 1); + } + + #[tokio::test] + async fn transmit_interval_follows_remote_min_rx() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0); + p.required_min_rx = BfdInterval::from_secs(2); + s.receive_control_packet(&p).await; + assert_eq!(s.state.remote_min_rx_interval(), BfdInterval::from_secs(2)); + assert_eq!(s.cur_interval, BfdInterval::from_secs(2)); + p.required_min_rx = BfdInterval::from_millis(100); + s.receive_control_packet(&p).await; + // never faster than our own effective desired-min-tx, which is 1s while not Up + assert_eq!(s.cur_interval, BfdInterval::from_secs(1)); + } + + #[tokio::test] + async fn detection_time_saturates_on_remote_overflow() { + let (mut s, _events) = session(peer_cfg(300, 300, 3)).await; + let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0); + p.flags = p.flags.with_detect_mult(255); + p.desired_min_tx = BfdInterval::from_micros(u32::MAX); + s.receive_control_packet(&p).await; + assert_eq!(s.state.detection_time(), BfdInterval::from_micros(u32::MAX)); + } + + #[tokio::test] + async fn detection_time_uses_remote_mult_and_larger_interval() { + let (mut s, _events) = session(peer_cfg(300, 500, 3)).await; + let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0); + p.flags = p.flags.with_detect_mult(5); + p.desired_min_tx = BfdInterval::from_millis(400); + s.receive_control_packet(&p).await; + assert_eq!(s.state.detection_time(), BfdInterval::from_millis(2500)); + p.desired_min_tx = BfdInterval::from_millis(800); + s.receive_control_packet(&p).await; + assert_eq!(s.state.detection_time(), BfdInterval::from_millis(4000)); + } + #[tokio::test] + async fn watchdog_expires_only_without_packets() { + let (dispatcher, mut events) = capture_dispatcher().await; + let handle = BfdSessionHandle::new( + LOCAL, + peer_cfg(20, 20, 3), + BfdDiscriminator(LOCAL_DISCR), + dispatcher, + ) + .await + .unwrap(); + let fast = |state, your| { + let mut p = pkt(state, REMOTE_DISCR, your); + p.flags = p.flags.with_detect_mult(10); + p.desired_min_tx = BfdInterval::from_millis(20); + p.required_min_rx = BfdInterval::from_millis(20); + p + }; + let send = |p| handle.tx.send(SessionControlCommand::RxPacket(p)); + send(fast(BfdState::Down, 0)).await.unwrap(); + send(fast(BfdState::Init, LOCAL_DISCR)).await.unwrap(); + assert_eq!(next_event(&mut events).await.to_state, BfdState::Init); + assert_eq!(next_event(&mut events).await.to_state, BfdState::Up); + + // detection time is 10 x 20ms; keep-alives at 20ms must hold the session up + for _ in 0..15 { + send(fast(BfdState::Up, LOCAL_DISCR)).await.unwrap(); + time::sleep(Duration::from_millis(20)).await; + } + assert!( + events.try_recv().is_err(), + "session dropped despite keep-alives" + ); + + let ev = next_event(&mut events).await; + assert_eq!((ev.from_state, ev.to_state), (BfdState::Up, BfdState::Down)); + let stats = handle.get_stats().await.unwrap(); + assert_eq!(stats.state, BfdState::Down); + assert!(stats.last_down.is_some()); + handle.tx.send(SessionControlCommand::Quit).await.unwrap(); + } + + #[test] + fn bfd_error_display_orders_value_and_field() { + let e = BfdError::InvalidFieldValue("state", "bogus"); + assert_eq!(e.to_string(), "invalid value `bogus` for field `state`"); + } + + #[tokio::test] + async fn control_socket_uses_source_port_range_and_ttl_255() { + let a = bind_control_socket(LOCAL).await.unwrap(); + let b = bind_control_socket(LOCAL).await.unwrap(); + for s in [&a, &b] { + let addr = s.local_addr().unwrap(); + assert!(SOURCE_PORT_RANGE.contains(&addr.port()), "{addr}"); + assert_eq!(addr.ip(), LOCAL); + assert_eq!(s.ttl().unwrap(), 255); + } + assert_ne!( + a.local_addr().unwrap().port(), + b.local_addr().unwrap().port() + ); + } + + #[tokio::test] + async fn control_socket_bind_error_is_propagated() { + let err = bind_control_socket(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1))) + .await + .unwrap_err(); + assert_ne!(err.kind(), ErrorKind::AddrInUse); + } + + const OTHER: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 3)); + + async fn bfdd() -> (Bfdd, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(8); + let discr = BfdDiscriminator(LOCAL_DISCR); + let d = Bfdd { + local: SocketAddr::new(LOCAL, CONTROL_PORT), + peers: vec![peer_cfg(300, 300, 3)], + sessions_by_ip: HashMap::from([((LOCAL, PEER), discr)]), + sessions_by_discr: HashMap::from([(discr, BfdSessionHandle { tx })]), + unmatched_sources: std::sync::Mutex::new(HashSet::new()), + control_socket: Arc::new(UdpSocket::bind((LOCAL, 0)).await.unwrap()), + event_dispatcher: EventDispatcherHandle::new().await, + }; + (d, rx) + } + + fn from(ip: IpAddr) -> SocketAddr { + SocketAddr::new(ip, 49152) + } + + #[tokio::test] + async fn demux_by_source_when_your_discr_is_zero() { + let (d, _rx) = bfdd().await; + let p = pkt(BfdState::Down, REMOTE_DISCR, 0); + assert!(d.session_for_packet(&from(PEER), &p).is_some()); + assert!(d.session_for_packet(&from(OTHER), &p).is_none()); + } + + #[tokio::test] + async fn demux_by_discriminator_ignores_source_address() { + let (d, _rx) = bfdd().await; + let p = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR); + assert!(d.session_for_packet(&from(PEER), &p).is_some()); + assert!(d.session_for_packet(&from(OTHER), &p).is_some()); + let p = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR + 1); + assert!(d.session_for_packet(&from(PEER), &p).is_none()); + } + + #[tokio::test] + async fn unmatched_sources_are_tracked_until_they_match() { + let (d, _rx) = bfdd().await; + let miss = pkt(BfdState::Down, REMOTE_DISCR, 0); + let hit = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR); + assert!(d.session_for_packet(&from(OTHER), &miss).is_none()); + assert!(d.unmatched_sources.lock().unwrap().contains(&OTHER)); + assert!(d.session_for_packet(&from(OTHER), &hit).is_some()); + assert!(!d.unmatched_sources.lock().unwrap().contains(&OTHER)); + } + + #[tokio::test] + async fn datagram_is_parsed_and_delivered_to_session() { + let (d, mut rx) = bfdd().await; + let bytes = pkt(BfdState::Down, REMOTE_DISCR, 0).serialize().unwrap(); + d.handle_datagram(&bytes, from(PEER)).await; + match rx.try_recv() { + Ok(SessionControlCommand::RxPacket(p)) => { + assert_eq!(p.my_disc, BfdDiscriminator(REMOTE_DISCR)) + } + _ => panic!("expected RxPacket"), + } + } + + #[tokio::test] + async fn bad_datagrams_are_dropped() { + let (d, mut rx) = bfdd().await; + let good = pkt(BfdState::Down, REMOTE_DISCR, 0).serialize().unwrap(); + d.handle_datagram(&good[..23], from(PEER)).await; + d.handle_datagram(&[], from(PEER)).await; + d.handle_datagram(&[0xff; 7], from(PEER)).await; + let long = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| f.with_length(48)); + d.handle_datagram(&long.serialize().unwrap(), from(PEER)) + .await; + let unknown = pkt(BfdState::Down, REMOTE_DISCR, 0); + d.handle_datagram(&unknown.serialize().unwrap(), from(OTHER)) + .await; + assert!(rx.try_recv().is_err()); + } + + #[tokio::test] + async fn bfdd_rejects_multiple_bind_addresses() { + let cfg = ResolvedConfig { + bind: vec![LOCAL, PEER], + hooks_dir: PathBuf::from("/nonexistent"), + peers: vec![], + }; + let err = Bfdd::new(cfg).await.err().expect("should fail"); + assert!(err.to_string().contains("only one"), "{err}"); + } } diff --git a/src/events.rs b/src/events.rs index ca13bac..78c340c 100644 --- a/src/events.rs +++ b/src/events.rs @@ -1,4 +1,4 @@ -use core::{fmt, str}; +use core::fmt; use std::{ ffi::{OsStr, OsString}, net::IpAddr, @@ -92,16 +92,10 @@ impl ScriptHookSink { ); if !output.status.success() { if !output.stdout.is_empty() { - warn!( - " stdout: \n{}", - str::from_utf8(output.stdout.as_slice()).unwrap() - ); + warn!(" stdout: \n{}", String::from_utf8_lossy(&output.stdout)); } if !output.stderr.is_empty() { - warn!( - " stderr: \n{}", - str::from_utf8(output.stderr.as_slice()).unwrap() - ); + warn!(" stderr: \n{}", String::from_utf8_lossy(&output.stderr)); } } } @@ -177,3 +171,214 @@ impl ScriptHookSink { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::{ + fs::{self as stdfs, Permissions}, + net::{Ipv4Addr, Ipv6Addr}, + os::unix::fs::PermissionsExt, + path::Path, + sync::atomic::{AtomicU32, Ordering}, + time::Duration, + }; + + static COUNTER: AtomicU32 = AtomicU32::new(0); + + struct TempDir(PathBuf); + impl TempDir { + fn new() -> Self { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let path = + std::env::temp_dir().join(format!("rust-bfd-hooks-{}-{}", std::process::id(), n)); + stdfs::create_dir_all(path.join("hooks")).unwrap(); + Self(path) + } + fn hooks(&self) -> PathBuf { + self.0.join("hooks") + } + fn out(&self) -> PathBuf { + self.0.join("out") + } + fn script(&self, name: &str, mode: u32, body: &str) { + let path = self.hooks().join(name); + stdfs::write(&path, format!("#!/bin/sh\n{body}\n")).unwrap(); + stdfs::set_permissions(&path, Permissions::from_mode(mode)).unwrap(); + } + fn append_script(&self, name: &str, mode: u32, line: &str) { + self.script( + name, + mode, + &format!("echo \"{line}\" >> {}", self.out().display()), + ); + } + fn output(&self) -> Vec { + stdfs::read_to_string(self.out()) + .map(|s| s.lines().map(str::to_owned).collect()) + .unwrap_or_default() + } + } + impl Drop for TempDir { + fn drop(&mut self) { + let _ = stdfs::remove_dir_all(&self.0); + } + } + + fn event(local: IpAddr, remote: IpAddr, from: BfdState, to: BfdState) -> EventMessage { + EventMessage::StateChange(StateChangeEvent { + local_discr: BfdDiscriminator(11), + remote_discr: BfdDiscriminator(22), + local_ip: local, + remote_ip: remote, + from_state: from, + to_state: to, + }) + } + + fn v4_event() -> EventMessage { + event( + Ipv4Addr::new(192, 0, 2, 1).into(), + Ipv4Addr::new(192, 0, 2, 2).into(), + BfdState::Init, + BfdState::Up, + ) + } + + const ENV_LINE: &str = "$1|$2|$BFD_STATE|$BFD_LAST_STATE|$BFD_LOCAL_ADDR|$BFD_PEER_ADDR|$BFD_LOCAL_DISCRIMINATOR|$BFD_PEER_DISCRIMINATOR"; + + #[tokio::test] + async fn passes_args_and_environment() { + let t = TempDir::new(); + t.append_script("10-env", 0o755, ENV_LINE); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert_eq!( + t.output(), + ["Up|192.0.2.2|Up|Init|192.0.2.1|192.0.2.2|11|22"] + ); + } + + #[tokio::test] + async fn passes_ipv6_addresses() { + let t = TempDir::new(); + t.append_script("10-env", 0o755, ENV_LINE); + let ev = event( + Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1).into(), + Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 2).into(), + BfdState::Up, + BfdState::Down, + ); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&ev) + .await; + assert_eq!( + t.output(), + ["Down|2001:db8::2|Down|Up|2001:db8::1|2001:db8::2|11|22"] + ); + } + + #[tokio::test] + async fn environment_is_cleared() { + let t = TempDir::new(); + std::env::set_var("RUST_BFD_TEST_LEAK", "leaked"); + t.append_script( + "10-env", + 0o755, + "${RUST_BFD_TEST_LEAK:-unset}|${HOME:-unset}", + ); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert_eq!(t.output(), ["unset|unset"]); + } + + #[tokio::test] + async fn runs_scripts_in_lexicographic_order() { + let t = TempDir::new(); + t.append_script("20-b", 0o755, "b"); + t.append_script("30-c", 0o755, "c"); + t.append_script("10-a", 0o755, "a"); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert_eq!(t.output(), ["a", "b", "c"]); + } + + #[tokio::test] + async fn skips_non_executable_files_and_directories() { + let t = TempDir::new(); + t.append_script("10-run", 0o755, "run"); + t.append_script("20-noexec", 0o644, "noexec"); + let dir = t.hooks().join("30-dir"); + stdfs::create_dir(&dir).unwrap(); + stdfs::set_permissions(&dir, Permissions::from_mode(0o755)).unwrap(); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert_eq!(t.output(), ["run"]); + } + + #[tokio::test] + async fn failing_script_does_not_stop_later_scripts() { + let t = TempDir::new(); + t.script( + "10-fail", + 0o755, + "printf '\\377\\376'; printf '\\377' >&2; exit 3", + ); + t.append_script("20-ok", 0o755, "ok"); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert_eq!(t.output(), ["ok"]); + } + + #[tokio::test] + async fn missing_script_dir_is_not_fatal() { + let t = TempDir::new(); + let missing: &Path = &t.0.join("nope"); + ScriptHookSink::with_script_dir(missing) + .handle_event(&v4_event()) + .await; + assert!(t.output().is_empty()); + } + + #[tokio::test] + async fn empty_script_dir_is_a_noop() { + let t = TempDir::new(); + ScriptHookSink::with_script_dir(t.hooks()) + .handle_event(&v4_event()) + .await; + assert!(t.output().is_empty()); + } + + #[tokio::test] + async fn events_sent_over_channel_run_hooks() { + let t = TempDir::new(); + t.append_script("10-env", 0o755, ENV_LINE); + let sink = ScriptHookSink::with_script_dir(t.hooks()); + let tx = sink.channel(); + let handle = Box::new(sink).run(); + tx.send(v4_event()).await.unwrap(); + tx.send(event( + Ipv4Addr::new(192, 0, 2, 1).into(), + Ipv4Addr::new(192, 0, 2, 2).into(), + BfdState::Up, + BfdState::Down, + )) + .await + .unwrap(); + for _ in 0..100 { + if t.output().len() >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + handle.abort(); + let out = t.output(); + assert_eq!(out.len(), 2); + assert!(out[0].starts_with("Up|") && out[1].starts_with("Down|")); + } +} diff --git a/src/lib.rs b/src/lib.rs index aecbe6c..6f4469a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -300,4 +300,120 @@ mod tests { a.clear(); assert_eq!(a.load(), None); } + + #[test] + fn atomic_instant_swap_returns_previous() { + let a = AtomicInstant::new(); + let t1 = Instant::now() + Duration::from_secs(1); + let t2 = t1 + Duration::from_secs(1); + a.store(t1); + assert_eq!(a.swap(t2), Some(t1)); + assert_eq!(a.load(), Some(t2)); + } + + #[test] + fn atomic_instant_clamps_before_epoch() { + let a = AtomicInstant::new(); + let before = Instant::now().checked_sub(Duration::from_secs(1)); + let Some(before) = before else { return }; + a.store(before); + let loaded = a.load().unwrap(); + assert!(loaded >= before); + assert!(loaded <= Instant::now()); + } + + #[test] + fn interval_stats_display() { + assert_eq!( + IntervalStats::default().to_string(), + "0 pkts, interval min/avg/max -/-/- ms" + ); + let stats = AtomicIntervalStats::new(); + let t0 = Instant::now(); + for ms in [0, 100, 400, 500] { + stats.record(t0 + Duration::from_millis(ms)); + } + assert_eq!( + stats.snapshot().to_string(), + "4 pkts, interval min/avg/max 100/167/300 ms" + ); + } + + fn session_stats(last_up: Option, last_down: Option) -> BfdSessionStats { + let rx = AtomicIntervalStats::new(); + let tx = AtomicIntervalStats::new(); + let t0 = Instant::now(); + rx.record(t0); + rx.record(t0 + Duration::from_millis(300)); + tx.record(t0); + BfdSessionStats { + local_ip: "192.0.2.1".parse().unwrap(), + remote_ip: "192.0.2.2".parse().unwrap(), + local_discr: BfdDiscriminator(10), + remote_discr: BfdDiscriminator(20), + state: BfdState::Up, + last_diag: BfdDiagnostic::TimeExpired, + rx: rx.snapshot(), + tx: tx.snapshot(), + last_up, + last_down, + last_change: Instant::now(), + detect_time: BfdInterval::from_millis(900), + base_interval: BfdInterval::from_secs(1), + } + } + + #[test] + fn session_stats_display_never() { + let out = session_stats(None, None).to_string(); + let lines: Vec<&str> = out.lines().collect(); + assert_eq!(lines.len(), 5); + assert_eq!( + lines[0].split_whitespace().collect::>(), + ["192.0.2.2", "Up", "2", "1"] + ); + assert_eq!( + lines[1].trim(), + "LocalDiscr:10 RemoteDiscr:20 DetectTime: 900000us TxTime: 1000000us Diag:TimeExpired Last: 0s" + ); + assert_eq!( + lines[2], + " Rx: 2 pkts, interval min/avg/max 300/300/300 ms" + ); + assert_eq!(lines[3], " Tx: 1 pkts, interval min/avg/max -/-/- ms"); + assert_eq!(lines[4], " Last Up: never Last Down: never"); + } + + #[test] + fn session_stats_display_last_up_down() { + let now = Instant::now(); + let out = session_stats(Some(now), None).to_string(); + assert!(out.ends_with("Last Up: 0s ago Last Down: never"), "{out}"); + let out = session_stats(None, Some(now)).to_string(); + assert!(out.ends_with("Last Up: never Last Down: 0s ago"), "{out}"); + } + + #[test] + fn session_stats_with_base_interval() { + let s = session_stats(None, None).with_base_interval(BfdInterval::from_millis(300)); + assert_eq!(s.base_interval, BfdInterval::from_millis(300)); + } + + #[tokio::test] + async fn set_ttl_v4() { + let sock = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + set_ttl_or_hop_limit(&sock, 255).unwrap(); + assert_eq!(sock.ttl().unwrap(), 255); + set_ttl_or_hop_limit(&sock, 64).unwrap(); + assert_eq!(sock.ttl().unwrap(), 64); + } + + #[tokio::test] + async fn set_hop_limit_v6() { + let Ok(sock) = UdpSocket::bind("[::1]:0").await else { + return; + }; + set_ttl_or_hop_limit(&sock, 255).unwrap(); + assert_eq!(getsockopt(&sock, sockopt::Ipv6Ttl).unwrap(), 255); + } }