There are two main methods for verifying a Publisher. The first one is StepVerifier.<T>create(Publisher<T> source). The test that can be built with this technique looks like the following:
StepVerifier
.create(Flux.just("foo", "bar")) .expectSubscription() .expectNext("foo") .expectNext("bar") .expectComplete() .verify();
In this example, our Publisher should produce two particular elements, and subsequent operations verify whether particular elements have been delivered to the final subscriber. From the preceding example, we might understand the workings of a part of the StepVerifier API. The builder technique offered by that class allows us to define the order in which events will occur during the verification process. According to the preceding code, the first emitted event must be the event concerning the subscription, and the next events must be the "foo" and "bar" strings. Finally, StepVerifier#expectCompletion defines the presence of a terminal signal, which in our case must be the invocation of Subscriber#onComplete or simply the successful completion of the given Flux. To execute the verification or in other words, subscribe to creating a flow—we must call the .verify() method. This is a blocking call, so it will block the execution until the flow has emitted all expected events.
By using this simple technique, we may verify Publisher with a countable amount of elements and events. However, it will be hard to verify the flow with a huge amount of elements. In cases in which it is more important to check whether our publisher has emitted the particular amount of elements rather than the particular values, the .expectNextCount() method might be useful. This is depicted in the following code:
StepVerifier
.create(Flux.range(0, 100)) .expectSubscription() .expectNext(0) .expectNextCount(98) .expectNext(99) .expectComplete() .verify();
As we might remember from previous chapters, Flux.range(0, 100) produces the range of elements from 0 to 99 inclusively. In such cases, it is more important to check whether this emitted the particular amount of elements and, for example, that the elements are emitted in the right order. The case mentioned here might be achievable by applying .expectNext() and .expectNextCount() together. According to the code, the first element will be checked by the .expectNext(0) statement. The test flow will then check that the given publisher produced another 98 elements, so the given producer emits 99 elements at a given point in total. Since our publisher should produce 100 elements, the last element should then be 99, which is verified using the .expectNext(99) statement.
Despite the fact that the .expectNextCount() method addresses a part of the problem, there are some cases where it is not enough to simply check the count of emitted elements. For example, when it comes to verifying the part of the code that is responsible for filtering or selecting elements by specific rules, it is important to check that all emitted items match the defined filtering rule. For that purpose, StepVerifier makes it possible to record emitted data and its verification at once, using tools such as Java Hamcrest (see http://hamcrest.org/JavaHamcrest). The following code depicts a unit test that uses this library:
Publisher<Wallet> usersWallets = findAllUsersWallets(); StepVerifier
.create(usersWallets) .expectSubscription() .recordWith(ArrayList::new) .expectNextCount(1) .consumeRecordedWith(wallets -> assertThat( wallets, everyItem(hasProperty("owner", equalTo("admin"))) )) .expectComplete() .verify();
Using the preceding example, we can see how to record all elements and then match them with the given matcher. In contrast to the previous examples, where each expectation covered the verification of only one element or a specified count of elements, .consumeRecordedWith() makes it possible to verify all elements that are published by a given Publisher. It should be noted that .consumeRecordedWith() only works if .recordWith() is specified. In turn, we should carefully define the collection class in which records will be stored. In the case of a multithreaded publisher, the type of collection used for recording events should support the concurrent access, so in these cases, it is better to use .recordWith(ConcurrentLinkedQueue::new) instead of .recordWith(ArrayList::new), because ConcurrentLinkedQueue is thread safe in contrast with ArrayList.
From the previous few paragraphs, we can familiarize ourselves with the essentials of the Reactor Test API. Along with this, there are also other functionally similar methods. For example, an expectation of the next element might be defined as depicted in the following code:
StepVerifier
.create(Flux.just("alpha-foo", "betta-bar")) .expectSubscription() .expectNextMatches(e -> e.startsWith("alpha")) .expectNextMatches(e -> e.startsWith("betta")) .expectComplete() .verify();
The only difference between .expectNextMatches() and .expectNext() is that the former makes it possible to define the custom matcher Predicate, making it more flexible than the latter. This is because .expectNext() is based on the element's comparison using the .equals() method.
Similarly, .assertNext() and .consumeNextWith() make it possible to write custom assertions. It should be noted that .assertNext() is the alias for .consumeNextWith(). The difference between .expectNextMatches() and .assertNext() is that the former accepts a Predicate, which has to return true or false, while the latter accepts a Consumer that could throw an exception, and any AssertionError thrown by the consumer will be captured and thrown by the .verify() method, as shown in the following code:
StepVerifier
.create(findUsersUSDWallet()) .expectSubscription() .assertNext(wallet -> assertThat( wallet, hasProperty("currency", equalTo("USD")) )) .expectComplete() .verify();
Finally, we are left with uncovered error cases, which are also part of the normal system's life cycle. There are few API methods that make it possible to check error signals. The simplest one is .expectError(), with no arguments, as shown in the following code:
StepVerifier
.create(Flux.error(new RuntimeException("Error"))) .expectError() .verify();
Despite the fact that we may verify that the error was emitted, there are cases in which it is vital to test the particular error type. For example, if an incorrect credential is entered during a user's login, then the security service should emit BadCredentialsException.class. To verify the emitted error, we may use .expectError(Class<? extends Throwable>), as demonstrated in the following code:
StepVerifier
.create(securityService.login("admin", "wrong")) .expectSubscription() .expectError(BadCredentialsException.class) .verify();
In cases in which the checking of the error type is still not enough, there are additional extensions called .expectErrorMatches() and .consumeErrorWith() that allow a direct interaction with a signaled Throwable.
At this point, we may discover the essentials of the testing code written using Reactor 3 or any Reactive Streams specification compatible library. The StepVerifier API covers most of the reactive workflows. However, when it comes to the real development, there are some extra cases.