fix #670 Add expect|verifyErrorSatisfies StepVerifier error expectations

This commit adds two convenience methods: expectErrorSatisfies and
verifyErrorSatisfies, taking a Consumer that is expected to apply
assertions to the error.

This is similar to consumeErrorWith, but more discoverable (due to
naming alignment) and with the slight difference that the consumer's
assertion error message is wrapped inside a StepVerifier-specific
AssertionError. `consumeErrorWith` behavior is unchanged despite
mutualized code.

reactor-core/src/test/java/reactor/core/publisher/FluxMergeTest.java

@@ -116,7 +116,6 @@ public void mergeDelayError() {
 				Flux.range(1, 4)
 		))
 		            .expectNext(1, 2, 3, 4)
-		            .consumeErrorWith(e -> assertThat(e).isEqualTo(boom))
-		            .verify();
+		            .verifyErrorSatisfies(e -> assertThat(e).isEqualTo(boom));
 	}
 }

reactor-core/src/test/java/reactor/core/publisher/FluxPeekFuseableTest.java

@@ -676,7 +676,7 @@ public void fusedDoOnNextOnErrorDoOnErrorAllFailing() {
 
 		StepVerifier.create(f)
 		            .then(f::connect)
-		            .consumeErrorWith(e -> {
+		            .verifyErrorSatisfies(e -> {
 					            assertThat(e)
 					                      .isInstanceOf(IllegalStateException.class)
 					                      .hasMessage("fromOnError2")
@@ -686,8 +686,7 @@ public void fusedDoOnNextOnErrorDoOnErrorAllFailing() {
 					                      .hasCauseInstanceOf(IllegalArgumentException.class);
 					            assertThat(e.getCause().getCause())
 					                      .hasMessage("fromOnNext");
-				            })
-		            .verify();
+				            });
 	}
 
 	@Test
@@ -746,7 +745,7 @@ public void conditionalFusedDoOnNextOnErrorDoOnErrorAllFailing() {
 
 		StepVerifier.create(f)
 		            .then(f::connect)
-		            .consumeErrorWith(e -> {
+		            .verifyErrorSatisfies(e -> {
 					            assertThat(e)
 					                      .isInstanceOf(IllegalStateException.class)
 					                      .hasMessage("fromOnError2")
@@ -756,8 +755,7 @@ public void conditionalFusedDoOnNextOnErrorDoOnErrorAllFailing() {
 					                      .hasCauseInstanceOf(IllegalArgumentException.class);
 					            assertThat(e.getCause().getCause())
 					                      .hasMessage("fromOnNext");
-				            })
-		            .verify();
+				            });
 	}
 
 	@Test

reactor-core/src/test/java/reactor/core/publisher/FluxPublishOnTest.java

@@ -74,8 +74,7 @@ protected Scenario<String, String> defaultScenarioOptions(Scenario<String, Strin
 
 	void assertRejected(StepVerifier.Step<String> step) {
 		try {
-			step.consumeErrorWith(e -> Assert.assertTrue(Exceptions.unwrap(e) instanceof RejectedExecutionException))
-			    .verify();
+			step.verifyErrorSatisfies(e -> Assert.assertTrue(Exceptions.unwrap(e) instanceof RejectedExecutionException));
 		}
 		catch (Exception e) {
 			assertTrue(Exceptions.unwrap(e) instanceof RejectedExecutionException);

reactor-core/src/test/java/reactor/core/publisher/FluxUsingTest.java

@@ -305,10 +305,9 @@ public void sourceFactoryAndResourceCleanupThrow() {
 				false);
 
 		StepVerifier.create(test)
-		            .consumeErrorWith(e -> assertThat(e)
+		            .verifyErrorSatisfies(e -> assertThat(e)
 				            .hasMessage("resourceCleanup")
-				            .is(suppressingFactory))
-		            .verify();
+				            .is(suppressingFactory));
 
 	}
 

reactor-core/src/test/java/reactor/core/publisher/MonoWhenTest.java

@@ -320,8 +320,7 @@ public void whenMonoError() {
 		            .then(() -> assertThat(mp.isError()).isTrue())
 		            .then(() -> assertThat(mp.isSuccess()).isFalse())
 		            .then(() -> assertThat(mp.isTerminated()).isTrue())
-		            .consumeErrorWith(e -> assertThat(e).hasMessage("test1"))
-		            .verify();
+		            .verifyErrorSatisfies(e -> assertThat(e).hasMessage("test1"));
 	}
 
 	@Test

reactor-core/src/test/java/reactor/test/publisher/BaseOperatorTest.java

@@ -169,13 +169,13 @@ public final void sequenceOfNextWithCallbackError() {
 			String m = scenario.producerError.getMessage();
 			Consumer<StepVerifier.Step<O>> errorVerifier = step -> {
 				try {
-					step.consumeErrorWith(e -> {
+					step.verifyErrorSatisfies(e -> {
 						if (e instanceof NullPointerException || e instanceof IllegalStateException || e.getMessage()
 						                                                                                .equals(m)) {
 							return;
 						}
 						throw Exceptions.propagate(e);
-					}).verify();
+					});
 //						step.expectErrorMessage(m)
 //						.verifyThenAssertThat()
 //						.hasOperatorErrorWithMessage(m);

reactor-test/src/main/java/reactor/test/DefaultStepVerifierBuilder.java

@@ -40,6 +40,7 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import javax.annotation.Nullable;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
@@ -54,7 +55,6 @@
 import reactor.util.Loggers;
 import reactor.util.function.Tuple2;
 import reactor.util.function.Tuples;
-import javax.annotation.Nullable;
 
 /**
  * Default implementation of {@link StepVerifier.Step} and
@@ -128,15 +128,26 @@ public DefaultStepVerifierBuilder<T> as(String description) {
 	@Override
 	public DefaultStepVerifier<T> consumeErrorWith(Consumer<Throwable> consumer) {
 		Objects.requireNonNull(consumer, "consumer");
+		return consumeErrorWith(consumer, "consumeErrorWith", false);
+	}
+
+	private DefaultStepVerifier<T> consumeErrorWith(Consumer<Throwable> assertionConsumer, String description, boolean wrap) {
+		Objects.requireNonNull(assertionConsumer, "assertionConsumer");
 		SignalEvent<T> event = new SignalEvent<>((signal, se) -> {
 			if (!signal.isOnError()) {
 				return fail(se, "expected: onError(); actual: %s", signal);
 			}
 			else {
-				consumer.accept(signal.getThrowable());
-				return Optional.empty();
+				try {
+					assertionConsumer.accept(signal.getThrowable());
+					return Optional.empty();
+				}
+				catch (AssertionError e) {
+					if (wrap) return fail(se, "assertion failed on exception <%s>: %s", signal.getThrowable(), e.getMessage());
+					throw e;
+				}
 			}
-		}, "consumeErrorWith");
+		}, description);
 		this.script.add(event);
 		return build();
 	}
@@ -288,6 +299,11 @@ else if (!predicate.test(signal.getThrowable())) {
 		return build();
 	}
 
+	@Override
+	public DefaultStepVerifier<T> expectErrorSatisfies(Consumer<Throwable> assertionConsumer) {
+		return consumeErrorWith(assertionConsumer, "expectErrorSatisfies", true);
+	}
+
 	@Override
 	public DefaultStepVerifierBuilder<T> expectNoFusionSupport() {
 		return expectFusion(Fuseable.ANY, Fuseable.NONE);
@@ -509,6 +525,11 @@ public Duration verifyErrorMatches(Predicate<Throwable> predicate) {
 		return expectErrorMatches(predicate).verify();
 	}
 
+	@Override
+	public Duration verifyErrorSatisfies(Consumer<Throwable> assertionConsumer) {
+		return consumeErrorWith(assertionConsumer, "verifyErrorSatisfies", true).verify();
+	}
+
 	@Override
 	public Duration verifyComplete() {
 		return expectComplete().verify();

reactor-test/src/main/java/reactor/test/StepVerifier.java

@@ -377,6 +377,19 @@ interface LastStep {
 		 */
 		StepVerifier expectErrorMatches(Predicate<Throwable> predicate);
 
+		/**
+		 * Expect an error and assert it via assertion(s) provided as a {@link Consumer}.
+		 * Any {@link AssertionError} thrown by the consumer has its message propagated
+		 * through a new StepVerifier-specific AssertionError.
+		 *
+		 * @param assertionConsumer the consumer that applies assertion(s) on the next received error
+		 *
+		 * @return the built verification scenario, ready to be verified
+		 *
+		 * @see Subscriber#onError(Throwable)
+		 */
+		StepVerifier expectErrorSatisfies(Consumer<Throwable> assertionConsumer);
+
 		/**
 		 * Expect the completion signal.
 		 *
@@ -466,6 +479,27 @@ interface LastStep {
 		 */
 		Duration verifyErrorMatches(Predicate<Throwable> predicate);
 
+		/**
+		 * Trigger the {@link #verify() verification}, expecting an error as terminal event
+		 * which gets asserted via assertion(s) provided as a {@link Consumer}.
+		 * Any {@link AssertionError} thrown by the consumer has its message propagated
+		 * through a new StepVerifier-specific AssertionError.
+		 * <p>
+		 * This is a convenience method that calls {@link #verify()} in addition to the
+		 * expectation. Explicitly use the expect method and verification method
+		 * separately if you need something more specific (like activating logging or
+		 * changing the default timeout).
+		 *
+		 * @param assertionConsumer the consumer that applies assertion(s) on the next received error
+		 *
+		 * @return the actual {@link Duration} the verification took.
+		 *
+		 * @see #expectErrorSatisfies(Consumer)
+		 * @see #verify()
+		 * @see Subscriber#onError(Throwable)
+		 */
+		Duration verifyErrorSatisfies(Consumer<Throwable> assertionConsumer);
+
 		/**
 		 * Trigger the {@link #verify() verification}, expecting a completion signal
 		 * as terminal event.

reactor-test/src/test/java/reactor/test/StepVerifierTests.java

@@ -361,6 +361,31 @@ public void errorMatchesInvalid() {
 	            .withMessage("expectation \"expectErrorMatches\" failed (predicate failed on exception: java.lang.IllegalArgumentException)");
 	}
 
+	@Test
+	public void errorSatisfies() {
+		Flux<String> flux = Flux.just("foo")
+		                        .concatWith(Mono.error(new IllegalArgumentException()));
+
+		StepVerifier.create(flux)
+		            .expectNext("foo")
+		            .expectErrorSatisfies(t -> assertThat(t).isInstanceOf(IllegalArgumentException.class))
+		            .verify();
+	}
+
+	@Test
+	public void errorSatisfiesInvalid() {
+		Flux<String> flux = Flux.just("foo")
+		                        .concatWith(Mono.error(new IllegalArgumentException()));
+
+		assertThatExceptionOfType(AssertionError.class)
+				.isThrownBy(() -> StepVerifier.create(flux)
+		            .expectNext("foo")
+		            .expectErrorSatisfies(t -> assertThat(t).hasMessage("foo"))
+		            .verify())
+	            .withMessage("expectation \"expectErrorSatisfies\" failed (assertion failed on exception <java.lang.IllegalArgumentException>: " +
+			            "\nExpecting message:\n <\"foo\">\nbut was:\n <null>)");
+	}
+
 	@Test
 	public void consumeErrorWith() {
 		Flux<String> flux = Flux.just("foo")
@@ -1321,19 +1346,50 @@ public void verifyErrorMessageTriggersVerificationSuccess() {
 	}
 
 	@Test
-	public void verifyErrorPredicateTriggersVerificationFail() {
+	public void verifyErrorPredicateTriggersVerificationFailBadSignal() {
 		assertThatExceptionOfType(AssertionError.class)
 				.isThrownBy(() -> StepVerifier.create(Flux.empty())
 				                              .verifyErrorMatches(e -> e instanceof IllegalArgumentException))
 		        .withMessage("expectation \"expectErrorMatches\" failed (expected: onError(); actual: onComplete())");
 	}
 
+	@Test
+	public void verifyErrorPredicateTriggersVerificationFailNoMatch() {
+		assertThatExceptionOfType(AssertionError.class)
+				.isThrownBy(() -> StepVerifier.create(Flux.error(new IllegalArgumentException("boom")))
+				                              .verifyErrorMatches(e -> e.getMessage() == null))
+		        .withMessage("expectation \"expectErrorMatches\" failed (predicate failed on exception: java.lang.IllegalArgumentException: boom)");
+	}
+
 	@Test
 	public void verifyErrorPredicateTriggersVerificationSuccess() {
 		StepVerifier.create(Flux.error(new IllegalArgumentException("boom")))
 		            .verifyErrorMatches(e -> e instanceof IllegalArgumentException);
 	}
 
+	@Test
+	public void verifyErrorAssertionTriggersVerificationFailBadSignal() {
+		assertThatExceptionOfType(AssertionError.class)
+				.isThrownBy(() -> StepVerifier.create(Flux.empty())
+				                              .verifyErrorSatisfies(e -> assertThat(e).isNotNull()))
+		        .withMessage("expectation \"verifyErrorSatisfies\" failed (expected: onError(); actual: onComplete())");
+	}
+
+	@Test
+	public void verifyErrorAssertionTriggersVerificationFailNoMatch() {
+		assertThatExceptionOfType(AssertionError.class)
+				.isThrownBy(() -> StepVerifier.create(Flux.error(new IllegalArgumentException("boom")))
+				                              .verifyErrorSatisfies(e -> assertThat(e).hasMessage("foo")))
+		        .withMessage("expectation \"verifyErrorSatisfies\" failed (assertion failed on exception <java.lang.IllegalArgumentException: boom>: "
+				        + "\nExpecting message:\n <\"foo\">\nbut was:\n <\"boom\">)");
+	}
+
+	@Test
+	public void verifyErrorAssertionTriggersVerificationSuccess() {
+		StepVerifier.create(Flux.error(new IllegalArgumentException("boom")))
+		            .verifyErrorSatisfies(e -> assertThat(e).hasMessage("boom"));
+	}
+
 	@Test
 	public void verifyCompleteTriggersVerificationFail() {
 		assertThatExceptionOfType(AssertionError.class)