[Reactor 3 Reference Guide] 6. Testing

6.1. Testing a Scenario with StepVerifier

The most common case for testing a Reactor sequence is to have a Flux or a Mono defined in your code (for example, it might be returned by a method) and to want to test how it behaves when subscribed to.

This situation translates well to defining a “test scenario,” where you define your expectations in terms of events, step-by-step. You can ask and answer questions such as the following:

• What is the next expected event?
• Do you expect the Flux to emit a particular value?
• Or maybe to do nothing for the next 300ms?
  
  

You can express all of that through the StepVerifier API.

StepVerifier API 를 통해 아래 같은 모든 상황을 표현 할 수 있다.

• 다음 예상 이벤트는 무엇입니까?
• Flux가 특정 값을 방출할 것으로 예상하십니까?
• 아니면 다음 300ms 동안 아무 것도 하지 않을 수 있습니까?
  
  

  

@Test
    fun testAppendBoomError() {
        val source = Flux.just("thing1", "thing2")
        StepVerifier.create(appendBoomError(source))
            .expectNext("thing1")
            .expectNext("thing2")
            .expectErrorMessage("boom")
            .verify()
    }

    fun <T> appendBoomError(source: Flux<T>): Flux<T> {
        return source.concatWith(IllegalArgumentException("boom").toMono())
    }

 

boom 부분을 바꾸면 실패함!


The API is a builder. You start by creating a StepVerifier and passing the sequence to be tested. This offers a choice of methods that let you:

Express expectations about the next signals to occur. If any other signal is received (or the content of the signal does not match the expectation), the whole test fails with a meaningful AssertionError. For example, you might use expectNext(T…​) and expectNextCount(long).

Consume the next signal. This is used when you want to skip part of the sequence or when you want to apply a custom assertion on the content of the signal (for example, to check that there is an onNext event and assert that the emitted item is a list of size 5). For example, you might use consumeNextWith(Consumer<T>).

Take miscellaneous actions such as pausing or running arbitrary code. For example, if you want to manipulate a test-specific state or context. To that effect, you might use thenAwait(Duration) and then(Runnable).

For terminal events, the corresponding expectation methods (expectComplete() and expectError() and all their variants) switch to an API where you cannot express expectations anymore. In that last step, all you can do is perform some additional configuration on the StepVerifier and then trigger the verification, often with verify() or one of its variants.

What happens at this point is that the StepVerifier subscribes to the tested Flux or Mono and plays the sequence, comparing each new signal with the next step in the scenario. As long as these match, the test is considered a success. As soon as there is a discrepancy, an AssertionError is thrown.

Remember the verify() step, which triggers the verification. To help, the API includes a few shortcut methods that combine the terminal expectations with a call to verify(): verifyComplete(), verifyError(), verifyErrorMessage(String), and others.
Note that, if one of the lambda-based expectations throws an AssertionError, it is reported as is, failing the test. This is useful for custom assertions.

By default, the verify() method and derived shortcut methods (verifyThenAssertThat, verifyComplete(), and so on) have no timeout. They can block indefinitely. You can use StepVerifier.setDefaultTimeout(Duration) to globally set a timeout for these methods, or specify one on a per-call basis with verify(Duration).

6.2. Manipulating Time

You can use StepVerifier with time-based operators to avoid long run times for corresponding tests. You can do so through the StepVerifier.withVirtualTime builder.

 

시간 기반 operator 와 StepVerifier를 사용하면, 시간을 조작하여 테스트 할수 있다.

It looks like the following example:

StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofDays(1))) //... continue expectations here

This virtual time feature plugs in a custom Scheduler in Reactor’s Schedulers factory. Since these timed operators usually use the default Schedulers.parallel() scheduler, replacing it with a VirtualTimeScheduler does the trick. However, an important prerequisite is that the operator be instantiated after the virtual time scheduler has been activated.

To increase the chances that this happens correctly, the StepVerifier does not take a simple Flux as input. withVirtualTime takes a Supplier, which guides you into lazily creating the instance of the tested flux after having done the scheduler set up.

Take extra care to ensure the Supplier<Publisher<T>> can be used in a lazy fashion. Otherwise, virtual time is not guaranteed. Especially avoid instantiating the Flux earlier in the test code and having the Supplier return that variable. Instead, always instantiate the Flux inside the lambda.

테스트 코드에서 Flux를 먼저 인스턴스화하고 Supplier가 해당 변수를 반환하도록 하지 말것.
항상 람다 내부에서 Flux를 인스턴스화 해야한다. Supplier가 Lazy로 사용되지 않으면 시간조작이 보장되지 않음.

There are two expectation methods that deal with time, and they are both valid with or without virtual time:

• thenAwait(Duration): Pauses the evaluation of steps (allowing a few signals to occur or delays to run out).
• expectNoEvent(Duration): Also lets the sequence play out for a given duration but fails the test if any signal occurs during that time.

Both methods pause the thread for the given duration in classic mode and advance the virtual clock instead in virtual mode.

expectNoEvent also considers the subscription as an event. If you use it as a first step, it usually fails because the subscription signal is detected. Use expectSubscription().expectNoEvent(duration) instead.

 

6.3. Performing Post-execution Assertions with StepVerifier

After having described the final expectation of your scenario, you can switch to a complementary assertion API instead of triggering verify(). To do so, use verifyThenAssertThat() instead.

verifyThenAssertThat() returns a StepVerifier.Assertions object, which you can use to assert a few elements of state once the whole scenario has played out successfully (because it also calls verify()). Typical (albeit advanced) usage is to capture elements that have been dropped by some operator and assert them (see the section on Hooks).

6.4. Testing the Context

For more information about the Context, see Adding a Context to a Reactive Sequence.

StepVerifier comes with a couple of expectations around the propagation of a Context:

• expectAccessibleContext: Returns a ContextExpectations object that you can use to set up expectations on the propagated Context. Be sure to call then() to return to the set of sequence expectations.
• expectNoAccessibleContext: Sets up an expectation that NO Context can be propagated up the chain of operators under test. This most likely occurs when the Publisher under test is not a Reactor one or does not have any operator that can propagate the Context (for example, a generator source).

Additionally, you can associate a test-specific initial Context to a StepVerifier by using StepVerifierOptions to create the verifier.

 

    @Test
    fun testContext(){
        StepVerifier.create(
            Mono.just(1).map { i: Int -> i + 10 },
            StepVerifierOptions.create().withInitialContext(reactor.util.context.Context.of("thing1", "thing2"))
        )
            .expectAccessibleContext()
            .contains("thing1", "thing2")
            .then()
            .expectNext(11)
            .verifyComplete()
    }

 

i + 9 로 바꾸면 실패함!

6.5. Manually Emitting with TestPublisher

For more advanced test cases, it might be useful to have complete mastery over the source of data, to trigger finely chosen signals that closely match the particular situation you want to test.

Another situation is when you have implemented your own operator and you want to verify how it behaves with regards to the Reactive Streams specification, especially if its source is not well behaved.

For both cases, reactor-test offers the TestPublisher class. This is a Publisher<T> that lets you programmatically trigger various signals:

• next(T) and next(T, T…​) triggers 1-n onNext signals.
• emit(T…​) triggers 1-n onNext signals and does complete().
• complete() terminates with an onComplete signal.
• error(Throwable) terminates with an onError signal.

You can get a well behaved TestPublisher through the create factory method. Also, you can create a misbehaving TestPublisher by using the createNonCompliant factory method. The latter takes a value or multiple values from the TestPublisher.Violation enum. The values define which parts of the specification the publisher can overlook. These enum values include:

• REQUEST_OVERFLOW: Allows next calls to be made despite an insufficient request, without triggering an IllegalStateException.
• ALLOW_NULL: Allows next calls to be made with a null value without triggering a NullPointerException.
• CLEANUP_ON_TERMINATE: Allows termination signals to be sent several times in a row. This includes complete(), error(), and emit().
• DEFER_CANCELLATION: Allows the TestPublisher to ignore cancellation signals and continue emitting signals as if the cancellation lost the race against said signals.

Finally, the TestPublisher keeps track of internal state after subscription, which can be asserted through its various assert* methods.

You can use it as a Flux or Mono by using the conversion methods, flux() and mono().

TestPublisher는 구독 후 내부 상태를 추적하며 다양한 assert* 메서드를 통해 테스트 가능함.

6.6. Checking the Execution Path with PublisherProbe

When building complex chains of operators, you could come across cases where there are several possible execution paths, materialized by distinct sub-sequences.

Most of the time, these sub-sequences produce a specific-enough onNext signal that you can assert that it was executed by looking at the end result.

For instance, consider the following method, which builds a chain of operators from a source and uses a switchIfEmpty to fall back to a particular alternative if the source is empty:

public Flux<String> processOrFallback(Mono<String> source, Publisher<String> fallback) { return source .flatMapMany(phrase -> Flux.fromArray(phrase.split("\\s+"))) .switchIfEmpty(fallback); }

You can test which logical branch of the switchIfEmpty was used, as follows:

@Test public void testSplitPathIsUsed() { StepVerifier.create(processOrFallback(Mono.just("just a phrase with tabs!"), Mono.just("EMPTY_PHRASE"))) .expectNext("just", "a", "phrase", "with", "tabs!") .verifyComplete(); } @Test public void testEmptyPathIsUsed() { StepVerifier.create(processOrFallback(Mono.empty(), Mono.just("EMPTY_PHRASE"))) .expectNext("EMPTY_PHRASE") .verifyComplete(); }

However, think about an example where the method produces a Mono<Void> instead. It waits for the source to complete, performs an additional task, and completes. If the source is empty, a fallback Runnable-like task must be performed instead. The following example shows such a case:

private Mono<String> executeCommand(String command) { return Mono.just(command + " DONE"); } public Mono<Void> processOrFallback(Mono<String> commandSource, Mono<Void> doWhenEmpty) { return commandSource .flatMap(command -> executeCommand(command).then()) .switchIfEmpty(doWhenEmpty); }

	then() forgets about the command result. It cares only that it was completed.
	How to distinguish between two cases that are both empty sequences?

To verify that your processOrFallback method does indeed go through the doWhenEmpty path, you need to write a bit of boilerplate. Namely you need a Mono<Void> that:

• Captures the fact that it has been subscribed to.
• Lets you assert that fact after the whole process has terminated.

Before version 3.1, you would need to manually maintain one AtomicBoolean per state you wanted to assert and attach a corresponding doOn* callback to the publisher you wanted to evaluate. This could be a lot of boilerplate when having to apply this pattern regularly. Fortunately, 3.1.0 introduced an alternative with PublisherProbe. The following example shows how to use it:

@Test public void testCommandEmptyPathIsUsed() { PublisherProbe<Void> probe = PublisherProbe.empty(); StepVerifier.create(processOrFallback(Mono.empty(), probe.mono())) .verifyComplete(); probe.assertWasSubscribed(); probe.assertWasRequested(); probe.assertWasNotCancelled(); }

	Create a probe that translates to an empty sequence.
	Use the probe in place of Mono<Void> by calling probe.mono().
	After completion of the sequence, the probe lets you assert that it was used. You can check that is was subscribed to…​
	…​as well as actually requested data…​
	…​and whether or not it was cancelled.

You can also use the probe in place of a Flux<T> by calling .flux() instead of .mono(). For cases where you need to probe an execution path but also need the probe to emit data, you can wrap any Publisher<T> by using PublisherProbe.of(Publisher).

Suggest Edit to "Testing"