Testing Observables

observable_extensions.cs

    public static class ObservableExtensions
    {
        public static Func<bool> WasObserved<T>(this IObservable<T> observableUnderTest, T expected)
        {
            var wasExpectedObserved = false;
            var result = observableUnderTest.Subscribe(s =>
            {
                wasExpectedObserved = s.Equals(expected);
            });

            return () =>
            {
                result.Dispose();
                return wasExpectedObserved;
            };
        }

        public static Func<bool> Completed<T>(this IObservable<T> observableUnderTest)
        {
            var isCompleted = false;

            var sub = observableUnderTest.Subscribe(
                s =>{}, 
                () => isCompleted = true);

            return () =>
            {
                sub.Dispose();
                return isCompleted;
            };
        }

        public class ObservableCapture<T>
        {
            public ManualResetEventSlim HasObserved { get; } = new ManualResetEventSlim();

            private readonly IList<T> _captured;
            public T[] Captured => _captured.ToArray();
            public bool IsCompleted { get; set; }
            public Exception Error { get; set; }

            public ObservableCapture()
            {
                _captured = new List<T>();
            }

            public void Add(T capture)
            {
                _captured.Add(capture);
            }
        }

        public static Func<ObservableCapture<T>> Capture<T>(this IObservable<T> observableUnderTest)
        {
            var result = new ObservableCapture<T>();
            var sub = observableUnderTest
                .ObserveOn(Scheduler.Immediate)
                .Subscribe(
                    c =>
                    {
                        result.HasObserved.Set();
                        result.Add(c);
                    },
                    ex => result.Error = ex,
                    () => result.IsCompleted = true
                );

            return () =>
            {
                sub.Dispose();
                return result;
            };
        }

        public static ObservableCapture<T> CaptureUntilObserved<T>(this IObservable<T> observableUnderTest, int count = 1)
        {
            var captureCount = 0;
            return CaptureUntil(observableUnderTest, t => ++captureCount == count);
        }

        public static void ShouldNotCapture<T>(this IObservable<T> observableUnderTest, TimeSpan? wait = null)
        {
            var result = CaptureUntil(observableUnderTest, t => false, wait, false);

            result.Captured.Should().BeEmpty("Observable expected to not be observed.");
            result.HasObserved.IsSet.Should().BeFalse("expected no observations");
        }

        public static ObservableCapture<T> CaptureUntil<T>(this IObservable<T> observableUnderTest, 
            Predicate<T> untilPredicate, 
            TimeSpan? wait = null,
            bool requireCapture = true)
        {
            var result = new ObservableCapture<T>();
            var waitSpan = wait ?? TimeSpan.FromMilliseconds(200);

            var sub = observableUnderTest.Subscribe(
                c =>
                {
                    result.Add(c);
                    if (untilPredicate(c))
                    {
                        result.HasObserved.Set();
                    }
                },
                ex =>
                {
                    result.Error = ex;
                    result.HasObserved.Set();
                },
                () =>
                {
                    result.IsCompleted = true;
                    result.HasObserved.Set();
                }
                );

            var waitResult = result.HasObserved.Wait(waitSpan);
            if (requireCapture)
            {
                waitResult.Should().BeTrue();
            }

            sub.Dispose();
            return result;
        }
    }