banc_host/
expect.rs

1//! Event-stream assertion helpers: wait for a matching event within a
2//! deadline, or assert silence over a window. The negative form succeeds via
3//! the timeout path — "nothing observed" is a verdict, not an error.
4
5use std::future::Future;
6use std::time::Duration;
7
8/// Anything that yields events asynchronously (a postcard-rpc subscription,
9/// an mpsc receiver, an RTT line stream...). `None` means the source closed.
10pub trait EventSource<T> {
11    fn next(&mut self) -> impl Future<Output = Option<T>> + Send;
12}
13
14impl<T: Send> EventSource<T> for tokio::sync::mpsc::Receiver<T> {
15    async fn next(&mut self) -> Option<T> {
16        self.recv().await
17    }
18}
19
20impl<T: Send> EventSource<T> for tokio::sync::mpsc::UnboundedReceiver<T> {
21    async fn next(&mut self) -> Option<T> {
22        self.recv().await
23    }
24}
25
26#[derive(Debug, thiserror::Error)]
27pub enum ExpectError {
28    #[error("deadline ({0:?}) elapsed without a matching event")]
29    Deadline(Duration),
30    #[error("event source closed without a matching event")]
31    Closed,
32    #[error("expected silence but observed: {0}")]
33    Unexpected(String),
34}
35
36/// Wait until `pred` matches an event, discarding non-matching events.
37pub async fn expect_matching<T, S: EventSource<T>>(
38    source: &mut S,
39    deadline: Duration,
40    mut pred: impl FnMut(&T) -> bool,
41) -> Result<T, ExpectError> {
42    let result = tokio::time::timeout(deadline, async {
43        loop {
44            match source.next().await {
45                Some(ev) if pred(&ev) => return Ok(ev),
46                Some(_) => continue,
47                None => return Err(ExpectError::Closed),
48            }
49        }
50    })
51    .await;
52    match result {
53        Ok(inner) => inner,
54        Err(_) => Err(ExpectError::Deadline(deadline)),
55    }
56}
57
58/// Assert that no event matching `pred` arrives within `window`.
59/// The timeout elapsing is the success path.
60pub async fn expect_quiet<T: std::fmt::Debug, S: EventSource<T>>(
61    source: &mut S,
62    window: Duration,
63    mut pred: impl FnMut(&T) -> bool,
64) -> Result<(), ExpectError> {
65    let result = tokio::time::timeout(window, async {
66        loop {
67            match source.next().await {
68                Some(ev) if pred(&ev) => return Err(ExpectError::Unexpected(format!("{ev:?}"))),
69                Some(_) => continue,
70                // Source closing early counts as quiet: nothing more can match.
71                None => return Ok(()),
72            }
73        }
74    })
75    .await;
76    match result {
77        Ok(inner) => inner,
78        Err(_) => Ok(()),
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[tokio::test]
87    async fn matching_event_found_among_noise() {
88        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
89        tx.send(1).await.unwrap();
90        tx.send(7).await.unwrap();
91        let got = expect_matching(&mut rx, Duration::from_millis(100), |v| *v == 7)
92            .await
93            .unwrap();
94        assert_eq!(got, 7);
95    }
96
97    #[tokio::test]
98    async fn deadline_elapses_without_match() {
99        let (tx, mut rx) = tokio::sync::mpsc::channel::<u32>(8);
100        tx.send(1).await.unwrap();
101        let err = expect_matching(&mut rx, Duration::from_millis(50), |v| *v == 7).await;
102        assert!(matches!(err, Err(ExpectError::Deadline(_))));
103    }
104
105    #[tokio::test]
106    async fn quiet_window_passes_and_catches_offender() {
107        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
108        tx.send(1).await.unwrap();
109        expect_quiet(&mut rx, Duration::from_millis(50), |v| *v == 7)
110            .await
111            .unwrap();
112        tx.send(7).await.unwrap();
113        let err = expect_quiet(&mut rx, Duration::from_millis(50), |v| *v == 7).await;
114        assert!(matches!(err, Err(ExpectError::Unexpected(_))));
115    }
116}
banc_host/expect.rs

banc_host/
expect.rs
