Pushmatch assertions

Src/FluentAssertions.Reactive/ReactiveAssertions.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Linq.Expressions;
 using System.Reactive;
 using System.Reactive.Linq;
 using System.Reactive.Threading.Tasks;
@@ -9,6 +10,7 @@
 using FluentAssertions.Execution;
 using FluentAssertions.Primitives;
 using FluentAssertions.Specialized;
+using JetBrains.Annotations;
 using Microsoft.Reactive.Testing;
 
 namespace FluentAssertions.Reactive
@@ -106,18 +108,18 @@ public async Task<AndWhichConstraint<ReactiveAssertions<TPayload>, IEnumerable<T
         }
 
         /// <summary>
-        /// Asserts that at least <paramref name="numberOfNotifications"/> notifications are pushed to the <see cref="FluentTestObserver{TPayload}"/> within the next 1 second.<br />
+        /// Asserts that at least <paramref name="numberOfNotifications"/> notifications are pushed to the <see cref="FluentTestObserver{TPayload}"/> within the next 1 seconds.<br />
         /// This includes any previously recorded notifications since it has been created or cleared. 
         /// </summary>
         /// <param name="numberOfNotifications">the number of notifications the observer should have recorded by now</param>
         /// <param name="because"></param>
         /// <param name="becauseArgs"></param>
         public AndWhichConstraint<ReactiveAssertions<TPayload>, IEnumerable<TPayload>> Push(int numberOfNotifications, string because = "", params object[] becauseArgs)
-            => Push(numberOfNotifications, TimeSpan.FromSeconds(10), because, becauseArgs);
+            => Push(numberOfNotifications, TimeSpan.FromSeconds(1), because, becauseArgs);
 
         /// <inheritdoc cref="Push(int,string,object[])"/>
         public Task<AndWhichConstraint<ReactiveAssertions<TPayload>, IEnumerable<TPayload>>> PushAsync(int numberOfNotifications, string because = "", params object[] becauseArgs)
-            => PushAsync(numberOfNotifications, TimeSpan.FromSeconds(10), because, becauseArgs);
+            => PushAsync(numberOfNotifications, TimeSpan.FromSeconds(1), because, becauseArgs);
 
         /// <summary>
         /// Asserts that at least 1 notification is pushed to the <see cref="FluentTestObserver{TPayload}"/> within the next 1 second.<br />
@@ -248,6 +250,133 @@ public AndConstraint<ReactiveAssertions<TPayload>> NotComplete(TimeSpan timeout,
         public AndConstraint<ReactiveAssertions<TPayload>> NotComplete(string because = "", params object[] becauseArgs)
             => NotComplete(TimeSpan.FromMilliseconds(100), because, becauseArgs);
 
+
+        /// <summary>
+        /// Asserts that at least one notification matching <paramref name="predicate"/> was pushed to the <see cref="FluentTestObserver{TPayload}"/>
+        /// within the specified <paramref name="timeout"/>.<br />
+        /// This includes any previously recorded notifications since it has been created or cleared.
+        /// </summary>
+        /// <param name="predicate">A predicate to match the items in the collection against.</param>
+        /// <param name="timeout">the maximum time to wait for the notification to arrive</param>
+        /// <param name="because">
+        /// A formatted phrase as is supported by <see cref="string.Format(string,object[])" /> explaining why the assertion
+        /// is needed. If the phrase does not start with the word <i>because</i>, it is prepended automatically.
+        /// </param>
+        /// <param name="becauseArgs">
+        /// Zero or more objects to format using the placeholders in <paramref name="because"/>.
+        /// </param>
+        /// <exception cref="ArgumentNullException"><paramref name="predicate"/> is <c>null</c>.</exception>
+        public AndConstraint<ReactiveAssertions<TPayload>> PushMatch(
+            [NotNull] Expression<Func<TPayload, bool>> predicate,
+            TimeSpan timeout,
+            string because = "",
+            params object[] becauseArgs)
+        {
+            if (predicate == null) throw new ArgumentNullException(nameof(predicate));
+
+            IList<TPayload> notifications = new List<TPayload>();
+
+            try
+            {
+                Func<TPayload, bool> func = predicate.Compile();
+                notifications = Observer.RecordedNotificationStream
+                    .Select(r => r.Value)
+                    .Dematerialize()
+                    .Where(func)
+                    .Take(1)
+                    .Timeout(timeout)
+                    .Catch<TPayload, TimeoutException>(exception => Observable.Empty<TPayload>())
+                    .ToList()
+                    .ToTask()
+                    .ExecuteInDefaultSynchronizationContext();
+            }
+            catch (Exception e)
+            {
+                if (e is AggregateException aggregateException)
+                    e = aggregateException.InnerException;
+                Execute.Assertion
+                    .BecauseOf(because, becauseArgs)
+                    .FailWith("Expected {context:observable} to push an item matching {0}{reason}, but it failed with a {1}.", predicate.Body, e);
+            }
+
+            Execute.Assertion
+                .BecauseOf(because, becauseArgs)
+                .ForCondition(notifications.Any())
+                .FailWith("Expected {context:observable} to push an item matching {0}{reason} within {1}.", predicate.Body, timeout);
+
+            return new AndConstraint<ReactiveAssertions<TPayload>>(this);
+        }
+
+        /// <summary>
+        /// Asserts that at least one notification matching <paramref name="predicate"/> was pushed to the <see cref="FluentTestObserver{TPayload}"/>
+        /// within the next 1 second.<br />
+        /// This includes any previously recorded notifications since it has been created or cleared.
+        /// </summary>
+        /// <param name="predicate">A predicate to match the items in the collection against.</param>
+        /// <param name="timeout">the maximum time to wait for the notification to arrive</param>
+        /// <param name="because">
+        /// A formatted phrase as is supported by <see cref="string.Format(string,object[])" /> explaining why the assertion
+        /// is needed. If the phrase does not start with the word <i>because</i>, it is prepended automatically.
+        /// </param>
+        /// <param name="becauseArgs">
+        /// Zero or more objects to format using the placeholders in <paramref name="because"/>.
+        /// </param>
+        /// <exception cref="ArgumentNullException"><paramref name="predicate"/> is <c>null</c>.</exception>
+        public AndConstraint<ReactiveAssertions<TPayload>> PushMatch(
+            [NotNull] Expression<Func<TPayload, bool>> predicate,
+            string because = "",
+            params object[] becauseArgs)
+            => PushMatch(predicate, TimeSpan.FromSeconds(1), because, becauseArgs);
+
+        /// <inheritdoc cref="PushMatch(Expression{Func{TPayload, bool}},TimeSpan,string,object[])"/>
+        public async Task<AndConstraint<ReactiveAssertions<TPayload>>> PushMatchAsync(
+            [NotNull] Expression<Func<TPayload, bool>> predicate,
+            TimeSpan timeout,
+            string because = "",
+            params object[] becauseArgs)
+        {
+            if (predicate == null)
+                throw new ArgumentNullException(nameof(predicate));
+
+            IList<TPayload> notifications = new List<TPayload>();
+
+            try
+            {
+                Func<TPayload, bool> func = predicate.Compile();
+                notifications = await Observer.RecordedNotificationStream
+                    .Select(r => r.Value)
+                    .Dematerialize()
+                    .Where(func)
+                    .Take(1)
+                    .Timeout(timeout)
+                    .Catch<TPayload, TimeoutException>(exception => Observable.Empty<TPayload>())
+                    .ToList()
+                    .ToTask().ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                if (e is AggregateException aggregateException)
+                    e = aggregateException.InnerException;
+                Execute.Assertion
+                    .BecauseOf(because, becauseArgs)
+                    .FailWith("Expected {context:observable} to push an item matching {0}{reason}, but it failed with a {1}.", predicate.Body, e);
+            }
+
+            Execute.Assertion
+                .BecauseOf(because, becauseArgs)
+                .ForCondition(notifications.Any())
+                .FailWith("Expected {context:observable} to push an item matching {0}{reason} within {1}.", predicate.Body, timeout);
+
+            return new AndWhichConstraint<ReactiveAssertions<TPayload>, IEnumerable<TPayload>>(this, notifications);
+        }
+
+        /// <inheritdoc cref="PushMatch(Expression{Func{TPayload, bool}},string,object[])"/>
+        public Task<AndConstraint<ReactiveAssertions<TPayload>>> PushMatchAsync(
+            [NotNull] Expression<Func<TPayload, bool>> predicate,
+            string because = "",
+            params object[] becauseArgs)
+            => PushMatchAsync(predicate, TimeSpan.FromSeconds(1), because, becauseArgs);
+
         protected Task<IList<Recorded<Notification<TPayload>>>> GetRecordedNotifications(TimeSpan timeout) =>
             Observer.RecordedNotificationStream
                 .TakeUntil(recorded => recorded.Value.Kind == NotificationKind.OnError)

Tests/FluentAssertions.Reactive.Specs/ReactiveAssertionSpecs.cs

@@ -126,8 +126,7 @@ public void When_the_observable_is_expected_to_fail_but_does_not_it_should_throw
 
             observer.Error.Should().BeNull();
         }
-
-
+
         [Fact]
         public void When_the_observable_completes_as_expected_it_should_not_throw()
         {
@@ -159,5 +158,68 @@ public void When_the_observable_is_expected_to_complete_but_does_not_it_should_t
             observer.Error.Should().BeNull();
         }
 
+        [Fact]
+        public void When_the_observable_pushes_an_expected_match_it_should_not_throw()
+        {
+            var scheduler = new TestScheduler();
+            var observable = scheduler.CreateColdObservable(
+                OnNext(100, 1),
+                OnNext(200, 2));
+
+            // observe the sequence
+            using var observer = observable.Observe(scheduler);
+            // push subscriptions
+            scheduler.AdvanceTo(250);
+
+            // Act
+            Action act = () => observer.Should().PushMatch(i => i > 1);
+
+            // Assert
+            act.Should().NotThrow();
+
+            observer.RecordedNotifications.Should().BeEquivalentTo(observable.Messages);
+        }
+
+        [Fact]
+        public void When_the_observable_does_not_push_a_match_it_should_throw()
+        {
+            var scheduler = new TestScheduler();
+            var observable = scheduler.CreateColdObservable(
+                OnNext(100, 1),
+                OnNext(200, 2));
+
+            // observe the sequence
+            using var observer = observable.Observe(scheduler);
+            // push subscriptions
+            scheduler.AdvanceTo(250);
+
+            // Act
+            Action act = () => observer.Should().PushMatch(i => i > 3, TimeSpan.FromMilliseconds(1));
+
+            // Assert
+            act.Should().Throw<XunitException>().WithMessage(
+                $"Expected observable to push an item matching (i > 3) within {Formatter.ToString(TimeSpan.FromMilliseconds(1))}.");
+
+            observer.RecordedNotifications.Should().BeEquivalentTo(observable.Messages);
+        }
+
+        [Fact]
+        public void When_the_observable_fails_instead_of_pushing_a_match_it_should_throw()
+        {
+            var exception = new ArgumentException("That was wrong.");
+            var scheduler = new TestScheduler();
+            var observable = scheduler.CreateColdObservable(
+                OnError<int>(1, exception));
+
+            // observe the sequence
+            using var observer = observable.Observe(scheduler);
+            scheduler.AdvanceTo(10);
+            // Act
+            Action act = () => observer.Should().PushMatch(i => i > 1);
+            // Assert
+            act.Should().Throw<XunitException>().WithMessage(
+                $"Expected observable to push an item matching (i > 1), but it failed with a {Formatter.ToString(exception)}.");
+            observer.Error.Should().BeEquivalentTo(exception);
+        }
     }
 }