ferroui_reactive_testing/
lib.rs

1//! ferroui-reactive-testing
2//!
3//! A port of Microsoft.Reactive.Testing (<https://github.com/dotnet/reactive>) for
4//! `ferroui-reactive`: the virtual-time [`TestScheduler`], recorded notifications ([`Recorded`]) and
5//! subscriptions ([`Subscription`]), hot and cold test observables, the observer that records what
6//! it receives, the factories of [`ReactiveTest`] and the assertions of `ReactiveAssert`.
7//!
8//! A test in the style of upstream:
9//!
10//! ```
11//! use ferroui_reactive::ObservableExt;
12//! use ferroui_reactive_testing::{assert_messages, assert_subscriptions, ReactiveTest as RT, TestScheduler};
13//!
14//! let scheduler = TestScheduler::new();
15//! let xs = scheduler.create_hot_observable(vec![RT::on_next(150, 1), RT::on_next(210, 2), RT::on_completed(250)]);
16//! let source = xs.as_observable();
17//! let res = scheduler.start(move || source.clone());
18//! assert_messages!(res.messages(), RT::on_next(210, 2), RT::on_completed::<i32>(250));
19//! // A hot observable subscribed directly keeps the subscription until it is disposed.
20//! assert_subscriptions!(xs.subscriptions(), RT::subscribe(200, 1000));
21//! ```
22
23use ferroui_reactive::concurrency::{IScheduler, VirtualTimeClock, VirtualTimeScheduler};
24use ferroui_reactive::disposables::{CompositeDisposable, Disposable};
25use ferroui_reactive::{IDisposable, IObservable, IObserver, Notification, ObservableError};
26use std::cell::RefCell;
27use std::fmt;
28use std::ops::Deref;
29use std::rc::Rc;
30use std::time::{Duration, SystemTime};
31
32/// The virtual time of the test scheduler: ticks of 100 nanoseconds, as `TimeSpan` counts them.
33pub struct TestTime;
34
35impl VirtualTimeClock for TestTime {
36    type Absolute = i64;
37    type Relative = i64;
38
39    fn add(absolute: i64, relative: i64) -> i64 {
40        absolute + relative
41    }
42
43    /// `new DateTimeOffset(absolute, TimeSpan.Zero)`, measured from the epoch of the system time.
44    fn to_system_time(absolute: i64) -> SystemTime {
45        SystemTime::UNIX_EPOCH + ticks_to_duration(absolute.max(0))
46    }
47
48    fn to_relative(duration: Duration) -> i64 {
49        duration_to_ticks(duration)
50    }
51
52    fn subtract(absolute: i64, duration: Duration) -> i64 {
53        absolute - duration_to_ticks(duration)
54    }
55
56    /// Work scheduled for now or earlier runs one tick later (`TestScheduler.ScheduleAbsolute`).
57    fn adjust_due_time(due_time: i64, clock: i64) -> i64 {
58        if due_time <= clock {
59            clock + 1
60        } else {
61            due_time
62        }
63    }
64}
65
66/// A duration as ticks of 100 nanoseconds (`TimeSpan.Ticks`).
67pub fn duration_to_ticks(duration: Duration) -> i64 {
68    (duration.as_nanos() / 100) as i64
69}
70
71/// Ticks of 100 nanoseconds as a duration (`TimeSpan.FromTicks`).
72pub fn ticks_to_duration(ticks: i64) -> Duration {
73    Duration::from_nanos(ticks.max(0) as u64 * 100)
74}
75
76/// The time of the test scheduler at `ticks` (`new DateTimeOffset(ticks, TimeSpan.Zero)`).
77pub fn ticks_to_time(ticks: i64) -> SystemTime {
78    TestTime::to_system_time(ticks)
79}
80
81/// A virtual-time scheduler for tests, whose clock counts ticks (`TestScheduler`). It dereferences to
82/// the virtual-time scheduler (`clock`, `start`, `advance_to`, `advance_by`, `sleep`,
83/// `schedule_absolute`, ...).
84#[derive(Clone)]
85pub struct TestScheduler {
86    inner: Rc<VirtualTimeScheduler<TestTime>>,
87}
88
89impl Default for TestScheduler {
90    fn default() -> Self {
91        Self::new()
92    }
93}
94
95impl Deref for TestScheduler {
96    type Target = VirtualTimeScheduler<TestTime>;
97
98    fn deref(&self) -> &Self::Target {
99        &self.inner
100    }
101}
102
103impl TestScheduler {
104    /// A scheduler with its clock at zero.
105    pub fn new() -> Self {
106        Self { inner: VirtualTimeScheduler::new(0) }
107    }
108
109    /// The scheduler as the handle operators take.
110    pub fn scheduler(&self) -> Rc<dyn IScheduler> {
111        self.inner.clone()
112    }
113
114    /// Schedules an action at an absolute virtual time.
115    pub fn schedule_absolute_action(&self, due_time: i64, action: impl FnOnce() + 'static) -> Rc<dyn IDisposable> {
116        self.inner.schedule_absolute(
117            due_time,
118            Box::new(move |_| {
119                action();
120                Disposable::empty()
121            }),
122        )
123    }
124
125    /// Creates the sequence at `created`, subscribes an observer at `subscribed` and disposes the
126    /// subscription at `disposed`, runs the scheduler, and returns the observer with what it
127    /// recorded (`Start(Func<IObservable<T>>, long, long, long)`).
128    pub fn start_with<T: Clone + 'static>(
129        &self,
130        create: impl FnOnce() -> Rc<dyn IObservable<T>> + 'static,
131        created: i64,
132        subscribed: i64,
133        disposed: i64,
134    ) -> Rc<MockObserver<T>> {
135        assert!(subscribed >= created, "Specified argument was out of the range of valid values. (Parameter 'subscribed')");
136        assert!(disposed >= created && disposed >= subscribed, "Specified argument was out of the range of valid values. (Parameter 'disposed')");
137        let observer = self.create_observer::<T>();
138        let source: Rc<RefCell<Option<Rc<dyn IObservable<T>>>>> = Rc::new(RefCell::new(None));
139        let subscription: Rc<RefCell<Option<Rc<dyn IDisposable>>>> = Rc::new(RefCell::new(None));
140        {
141            let source = source.clone();
142            self.schedule_absolute_action(created, move || *source.borrow_mut() = Some(create()));
143        }
144        {
145            let (source, subscription, observer) = (source.clone(), subscription.clone(), observer.clone());
146            self.schedule_absolute_action(subscribed, move || {
147                let source = source.borrow().clone().expect("the sequence was created");
148                *subscription.borrow_mut() = Some(source.subscribe(observer));
149            });
150        }
151        {
152            let subscription = subscription.clone();
153            self.schedule_absolute_action(disposed, move || {
154                let subscription = subscription.borrow().clone();
155                if let Some(subscription) = subscription {
156                    subscription.dispose();
157                }
158            });
159        }
160        self.inner.start();
161        observer
162    }
163
164    /// [`start_with`](Self::start_with) at the default times, created at 100, subscribed at 200 and
165    /// disposed at `disposed` (`Start(Func<IObservable<T>>, long)`).
166    pub fn start_disposed<T: Clone + 'static>(&self, create: impl FnOnce() -> Rc<dyn IObservable<T>> + 'static, disposed: i64) -> Rc<MockObserver<T>> {
167        self.start_with(create, ReactiveTest::CREATED, ReactiveTest::SUBSCRIBED, disposed)
168    }
169
170    /// [`start_with`](Self::start_with) at the default times: created at 100, subscribed at 200,
171    /// disposed at 1000 (`Start(Func<IObservable<T>>)`).
172    pub fn start<T: Clone + 'static>(&self, create: impl FnOnce() -> Rc<dyn IObservable<T>> + 'static) -> Rc<MockObserver<T>> {
173        self.start_with(create, ReactiveTest::CREATED, ReactiveTest::SUBSCRIBED, ReactiveTest::DISPOSED)
174    }
175
176    /// Runs the queued work (`VirtualTimeScheduler.Start()`).
177    pub fn run(&self) {
178        self.inner.start();
179    }
180
181    /// A sequence that notifies its messages at their absolute times to the observers subscribed
182    /// then (`CreateHotObservable`).
183    pub fn create_hot_observable<T: Clone + 'static>(&self, messages: Vec<Recorded<Notification<T>>>) -> Rc<HotObservable<T>> {
184        HotObservable::new(self.clone(), messages)
185    }
186
187    /// A sequence that notifies its messages at their times relative to each subscription
188    /// (`CreateColdObservable`).
189    pub fn create_cold_observable<T: Clone + 'static>(&self, messages: Vec<Recorded<Notification<T>>>) -> Rc<ColdObservable<T>> {
190        Rc::new(ColdObservable { scheduler: self.clone(), messages, subscriptions: Rc::new(RefCell::new(Vec::new())) })
191    }
192
193    /// An observer that records what it receives with the time (`CreateObserver`).
194    pub fn create_observer<T: 'static>(&self) -> Rc<MockObserver<T>> {
195        Rc::new(MockObserver { scheduler: self.clone(), messages: RefCell::new(Vec::new()) })
196    }
197}
198
199/// A value with the virtual time it was recorded at (`Recorded<T>`).
200#[derive(Clone, PartialEq)]
201pub struct Recorded<T> {
202    /// The virtual time (`Time`).
203    pub time: i64,
204    /// The value (`Value`).
205    pub value: T,
206}
207
208impl<T> Recorded<T> {
209    pub fn new(time: i64, value: T) -> Self {
210        Self { time, value }
211    }
212}
213
214impl<T: fmt::Debug> fmt::Debug for Recorded<T> {
215    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
216        write!(f, "{:?}@{}", self.value, self.time)
217    }
218}
219
220/// A subscription to a test observable: the virtual times it was made and disposed
221/// (`Subscription`).
222#[derive(Clone, Copy, PartialEq, Eq, Hash)]
223pub struct Subscription {
224    /// The time of the subscription (`Subscribe`).
225    pub subscribe: i64,
226    /// The time of its disposal; [`Subscription::INFINITE`] while it lasts (`Unsubscribe`).
227    pub unsubscribe: i64,
228}
229
230impl Subscription {
231    /// The time of a subscription that has not been disposed (`Infinite`).
232    pub const INFINITE: i64 = i64::MAX;
233
234    pub fn new(subscribe: i64, unsubscribe: i64) -> Self {
235        Self { subscribe, unsubscribe }
236    }
237
238    pub fn starting(subscribe: i64) -> Self {
239        Self { subscribe, unsubscribe: Self::INFINITE }
240    }
241}
242
243impl fmt::Debug for Subscription {
244    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
245        if self.unsubscribe == Self::INFINITE {
246            write!(f, "({}, Infinite)", self.subscribe)
247        } else {
248            write!(f, "({}, {})", self.subscribe, self.unsubscribe)
249        }
250    }
251}
252
253/// An error made by a test, which compares by identity (`new Exception()` of upstream's tests).
254#[derive(Debug)]
255pub struct TestException(pub String);
256
257impl fmt::Display for TestException {
258    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
259        f.write_str(&self.0)
260    }
261}
262
263impl std::error::Error for TestException {}
264
265/// A new error for a test (`new Exception(message)`).
266pub fn new_error(message: &str) -> ObservableError {
267    Rc::new(TestException(message.to_owned()))
268}
269
270/// A notification a test expects: one notification, or a notification that satisfies a predicate
271/// (the predicate forms of `ReactiveTest.OnNext` and `OnError`).
272#[derive(Clone)]
273pub enum Expected<T> {
274    /// The notification itself.
275    Exact(Notification<T>),
276    /// A value for which the predicate is true (`OnNext<T>(long, Func<T, bool>)`).
277    OnNextWhere(Rc<dyn Fn(&T) -> bool>),
278    /// An error for which the predicate is true (`OnError<T>(long, Func<Exception, bool>)`).
279    OnErrorWhere(Rc<dyn Fn(&ObservableError) -> bool>),
280}
281
282impl<T: PartialEq> Expected<T> {
283    /// Whether the notification is the expected one.
284    pub fn matches(&self, notification: &Notification<T>) -> bool {
285        match (self, notification) {
286            (Self::Exact(expected), actual) => expected == actual,
287            (Self::OnNextWhere(predicate), Notification::OnNext(value)) => predicate(value),
288            (Self::OnErrorWhere(predicate), Notification::OnError(error)) => predicate(error),
289            _ => false,
290        }
291    }
292}
293
294impl<T: fmt::Debug> fmt::Debug for Expected<T> {
295    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
296        match self {
297            Self::Exact(notification) => notification.fmt(f),
298            Self::OnNextWhere(_) => f.write_str("OnNext(<predicate>)"),
299            Self::OnErrorWhere(_) => f.write_str("OnError(<predicate>)"),
300        }
301    }
302}
303
304impl<T> From<Recorded<Notification<T>>> for Recorded<Expected<T>> {
305    fn from(recorded: Recorded<Notification<T>>) -> Self {
306        Recorded::new(recorded.time, Expected::Exact(recorded.value))
307    }
308}
309
310/// The factories of recorded notifications and subscriptions, and the default times of
311/// [`TestScheduler::start`] (`ReactiveTest`).
312pub struct ReactiveTest;
313
314impl ReactiveTest {
315    /// The default time the sequence is created at (`Created`).
316    pub const CREATED: i64 = 100;
317    /// The default time the observer subscribes at (`Subscribed`).
318    pub const SUBSCRIBED: i64 = 200;
319    /// The default time the subscription is disposed at (`Disposed`).
320    pub const DISPOSED: i64 = 1000;
321
322    /// A value at `ticks` (`OnNext<T>(long, T)`).
323    pub fn on_next<T>(ticks: i64, value: T) -> Recorded<Notification<T>> {
324        Recorded::new(ticks, Notification::OnNext(value))
325    }
326
327    /// A value at `ticks` for which `predicate` is true (`OnNext<T>(long, Func<T, bool>)`).
328    pub fn on_next_where<T>(ticks: i64, predicate: impl Fn(&T) -> bool + 'static) -> Recorded<Expected<T>> {
329        Recorded::new(ticks, Expected::OnNextWhere(Rc::new(predicate)))
330    }
331
332    /// The end of the sequence at `ticks` (`OnCompleted<T>(long)`).
333    pub fn on_completed<T>(ticks: i64) -> Recorded<Notification<T>> {
334        Recorded::new(ticks, Notification::OnCompleted)
335    }
336
337    /// An error at `ticks` (`OnError<T>(long, Exception)`).
338    pub fn on_error<T>(ticks: i64, error: ObservableError) -> Recorded<Notification<T>> {
339        Recorded::new(ticks, Notification::OnError(error))
340    }
341
342    /// An error at `ticks` for which `predicate` is true (`OnError<T>(long, Func<Exception, bool>)`).
343    pub fn on_error_where<T>(ticks: i64, predicate: impl Fn(&ObservableError) -> bool + 'static) -> Recorded<Expected<T>> {
344        Recorded::new(ticks, Expected::OnErrorWhere(Rc::new(predicate)))
345    }
346
347    /// A subscription from `start` to `end` (`Subscribe(long, long)`).
348    pub fn subscribe(start: i64, end: i64) -> Subscription {
349        Subscription::new(start, end)
350    }
351
352    /// A subscription from `start` that has not been disposed (`Subscribe(long)`).
353    pub fn subscribe_from(start: i64) -> Subscription {
354        Subscription::starting(start)
355    }
356}
357
358/// Asserts that the recorded notifications are the expected ones, in order (`AssertEqual` on the
359/// messages of an observer or observable).
360pub fn assert_recorded_eq<T: PartialEq + fmt::Debug>(actual: &[Recorded<Notification<T>>], expected: &[Recorded<Expected<T>>]) {
361    let matches = actual.len() == expected.len()
362        && actual.iter().zip(expected).all(|(a, e)| a.time == e.time && e.value.matches(&a.value));
363    if !matches {
364        panic!("the recorded notifications differ\n  expected: {expected:?}\n    actual: {actual:?}");
365    }
366}
367
368/// Asserts that the recorded notifications are the expected ones; each expected one is a
369/// `Recorded<Notification<T>>` or a `Recorded<Expected<T>>` (`res.Messages.AssertEqual(...)`).
370#[macro_export]
371macro_rules! assert_messages {
372    ($actual:expr $(, $expected:expr)* $(,)?) => {{
373        let expected = ::std::vec![$($crate::Recorded::<$crate::Expected<_>>::from($expected)),*];
374        $crate::assert_recorded_eq(&$actual, &expected);
375    }};
376}
377
378/// Asserts that the subscriptions are the expected ones, in order (`xs.Subscriptions.AssertEqual(...)`).
379#[macro_export]
380macro_rules! assert_subscriptions {
381    ($actual:expr $(, $expected:expr)* $(,)?) => {{
382        let expected: ::std::vec::Vec<$crate::Subscription> = ::std::vec![$($expected),*];
383        assert_eq!($actual, expected, "the subscriptions differ");
384    }};
385}
386
387/// Asserts that `action` panics (`ReactiveAssert.Throws`): the form of an exception thrown to the
388/// caller in the port.
389pub fn assert_throws(action: impl FnOnce()) {
390    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(action));
391    assert!(result.is_err(), "expected the action to panic");
392}
393
394/// An observer that records what it receives with the virtual time (`MockObserver<T>`,
395/// `ITestableObserver<T>`).
396pub struct MockObserver<T> {
397    scheduler: TestScheduler,
398    messages: RefCell<Vec<Recorded<Notification<T>>>>,
399}
400
401impl<T: Clone> MockObserver<T> {
402    /// What the observer received (`Messages`).
403    pub fn messages(&self) -> Vec<Recorded<Notification<T>>> {
404        self.messages.borrow().clone()
405    }
406}
407
408impl<T> IObserver<T> for MockObserver<T> {
409    fn on_next(&self, value: T) {
410        let time = self.scheduler.clock();
411        self.messages.borrow_mut().push(Recorded::new(time, Notification::OnNext(value)));
412    }
413
414    fn on_error(&self, error: ObservableError) {
415        let time = self.scheduler.clock();
416        self.messages.borrow_mut().push(Recorded::new(time, Notification::OnError(error)));
417    }
418
419    fn on_completed(&self) {
420        let time = self.scheduler.clock();
421        self.messages.borrow_mut().push(Recorded::new(time, Notification::OnCompleted));
422    }
423}
424
425/// A sequence of recorded notifications at absolute times, shared by its observers
426/// (`HotObservable<T>`, `ITestableObservable<T>`).
427pub struct HotObservable<T> {
428    scheduler: TestScheduler,
429    messages: Vec<Recorded<Notification<T>>>,
430    observers: Rc<RefCell<Vec<Rc<dyn IObserver<T>>>>>,
431    subscriptions: Rc<RefCell<Vec<Subscription>>>,
432}
433
434impl<T: Clone + 'static> HotObservable<T> {
435    fn new(scheduler: TestScheduler, messages: Vec<Recorded<Notification<T>>>) -> Rc<Self> {
436        let observers: Rc<RefCell<Vec<Rc<dyn IObserver<T>>>>> = Rc::new(RefCell::new(Vec::new()));
437        for message in &messages {
438            let observers = observers.clone();
439            let notification = message.value.clone();
440            scheduler.schedule_absolute_action(message.time, move || {
441                let current = observers.borrow().clone();
442                for observer in current {
443                    notification.clone().accept(&*observer);
444                }
445            });
446        }
447        Rc::new(Self { scheduler, messages, observers, subscriptions: Rc::new(RefCell::new(Vec::new())) })
448    }
449
450    /// The subscriptions made to the sequence (`Subscriptions`).
451    pub fn subscriptions(&self) -> Vec<Subscription> {
452        self.subscriptions.borrow().clone()
453    }
454
455    /// The notifications of the sequence (`Messages`).
456    pub fn messages(&self) -> Vec<Recorded<Notification<T>>> {
457        self.messages.clone()
458    }
459
460    /// The sequence as the handle operators take.
461    pub fn as_observable(self: &Rc<Self>) -> Rc<dyn IObservable<T>> {
462        self.clone()
463    }
464}
465
466impl<T: Clone + 'static> IObservable<T> for HotObservable<T> {
467    fn subscribe(&self, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
468        self.observers.borrow_mut().push(observer.clone());
469        let index = {
470            let mut subscriptions = self.subscriptions.borrow_mut();
471            subscriptions.push(Subscription::starting(self.scheduler.clock()));
472            subscriptions.len() - 1
473        };
474        let (observers, subscriptions, scheduler) = (self.observers.clone(), self.subscriptions.clone(), self.scheduler.clone());
475        Disposable::create(move || {
476            {
477                let mut observers = observers.borrow_mut();
478                if let Some(position) = observers.iter().position(|o| Rc::ptr_eq(o, &observer)) {
479                    observers.remove(position);
480                }
481            }
482            let mut subscriptions = subscriptions.borrow_mut();
483            let subscribe = subscriptions[index].subscribe;
484            subscriptions[index] = Subscription::new(subscribe, scheduler.clock());
485        })
486    }
487}
488
489/// A sequence of recorded notifications at times relative to each subscription
490/// (`ColdObservable<T>`, `ITestableObservable<T>`).
491pub struct ColdObservable<T> {
492    scheduler: TestScheduler,
493    messages: Vec<Recorded<Notification<T>>>,
494    subscriptions: Rc<RefCell<Vec<Subscription>>>,
495}
496
497impl<T: Clone + 'static> ColdObservable<T> {
498    /// The subscriptions made to the sequence (`Subscriptions`).
499    pub fn subscriptions(&self) -> Vec<Subscription> {
500        self.subscriptions.borrow().clone()
501    }
502
503    /// The notifications of the sequence (`Messages`).
504    pub fn messages(&self) -> Vec<Recorded<Notification<T>>> {
505        self.messages.clone()
506    }
507
508    /// The sequence as the handle operators take.
509    pub fn as_observable(self: &Rc<Self>) -> Rc<dyn IObservable<T>> {
510        self.clone()
511    }
512}
513
514impl<T: Clone + 'static> IObservable<T> for ColdObservable<T> {
515    fn subscribe(&self, observer: Rc<dyn IObserver<T>>) -> Rc<dyn IDisposable> {
516        let index = {
517            let mut subscriptions = self.subscriptions.borrow_mut();
518            subscriptions.push(Subscription::starting(self.scheduler.clock()));
519            subscriptions.len() - 1
520        };
521        let group = CompositeDisposable::new();
522        for message in &self.messages {
523            let notification = message.value.clone();
524            let target = observer.clone();
525            group.add(self.scheduler.schedule_relative(
526                message.time,
527                Box::new(move |_| {
528                    notification.accept(&*target);
529                    Disposable::empty()
530                }),
531            ));
532        }
533        let (subscriptions, scheduler) = (self.subscriptions.clone(), self.scheduler.clone());
534        Disposable::create(move || {
535            {
536                let mut subscriptions = subscriptions.borrow_mut();
537                let subscribe = subscriptions[index].subscribe;
538                subscriptions[index] = Subscription::new(subscribe, scheduler.clock());
539            }
540            group.dispose();
541        })
542    }
543}
ferroui_reactive_testing/lib.rs

ferroui_reactive_testing/
lib.rs
