1use crate::{Entity, Subscription, TestAppContext, TestDispatcher};
 2use futures::StreamExt as _;
 3use rand::prelude::*;
 4use smol::channel;
 5use std::{
 6    env,
 7    panic::{self, RefUnwindSafe},
 8};
 9
10pub fn run_test(
11    mut num_iterations: u64,
12    max_retries: usize,
13    test_fn: &mut (dyn RefUnwindSafe + Fn(TestDispatcher, u64)),
14    on_fail_fn: Option<fn()>,
15    _fn_name: String, // todo!("re-enable fn_name")
16) {
17    let starting_seed = env::var("SEED")
18        .map(|seed| seed.parse().expect("invalid SEED variable"))
19        .unwrap_or(0);
20    let is_randomized = num_iterations > 1;
21    if let Ok(iterations) = env::var("ITERATIONS") {
22        num_iterations = iterations.parse().expect("invalid ITERATIONS variable");
23    }
24
25    for seed in starting_seed..starting_seed + num_iterations {
26        let mut retry = 0;
27        loop {
28            if is_randomized {
29                eprintln!("seed = {seed}");
30            }
31            let result = panic::catch_unwind(|| {
32                let dispatcher = TestDispatcher::new(StdRng::seed_from_u64(seed));
33                test_fn(dispatcher, seed);
34            });
35
36            match result {
37                Ok(_) => break,
38                Err(error) => {
39                    if retry < max_retries {
40                        println!("retrying: attempt {}", retry);
41                        retry += 1;
42                    } else {
43                        if is_randomized {
44                            eprintln!("failing seed: {}", seed);
45                        }
46                        on_fail_fn.map(|f| f());
47                        panic::resume_unwind(error);
48                    }
49                }
50            }
51        }
52    }
53}
54
55pub struct Observation<T> {
56    rx: channel::Receiver<T>,
57    _subscription: Subscription,
58}
59
60impl<T: 'static> futures::Stream for Observation<T> {
61    type Item = T;
62
63    fn poll_next(
64        mut self: std::pin::Pin<&mut Self>,
65        cx: &mut std::task::Context<'_>,
66    ) -> std::task::Poll<Option<Self::Item>> {
67        self.rx.poll_next_unpin(cx)
68    }
69}
70
71pub fn observe<T: 'static>(entity: &impl Entity<T>, cx: &mut TestAppContext) -> Observation<()> {
72    let (tx, rx) = smol::channel::unbounded();
73    let _subscription = cx.update(|cx| {
74        cx.observe(entity, move |_, _| {
75            let _ = smol::block_on(tx.send(()));
76        })
77    });
78
79    Observation { rx, _subscription }
80}