Testing reactive streams presents unique challenges that traditional testing approaches can't adequately address. How do you verify asynchronous, non-blocking data flows? How do you test back pressure? How do you ensure your reactive pipelines behave correctly under various scenarios? The answer lies in StepVerifier, the powerful testing utility from Project Reactor that makes testing reactive code predictable, reliable, and comprehensive.

This article explores StepVerifier—the essential tool for verifying the behavior of Flux and Mono publishers in your reactive applications.

The Challenge of Testing Reactive Streams

Traditional testing approaches fall short with reactive code because:

1. Asynchronous Execution: Results aren't immediately available
2. Complex Lifecycles: Multiple events (onNext, onComplete, onError) need verification
3. Timing Issues: Race conditions can make tests flaky
4. Back Pressure: Requires testing flow control mechanisms
5. Virtual Time: Testing delays and time-based operations

StepVerifier addresses these challenges by providing a DSL (Domain Specific Language) for defining expected sequences of events and verifying them.

Setting Up StepVerifier

First, ensure you have the necessary dependency:

Maven:

<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<version>3.6.4</version>
<scope>test</scope>
</dependency>

Gradle:

testImplementation 'io.projectreactor:reactor-test:3.6.4'

Basic StepVerifier Usage

Testing a Simple Flux:

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.junit.jupiter.api.Test;
class BasicStepVerifierTest {
@Test
void testSimpleFlux() {
Flux<String> flux = Flux.just("Apple", "Banana", "Cherry");
StepVerifier.create(flux)
.expectNext("Apple")
.expectNext("Banana")
.expectNext("Cherry")
.expectComplete()
.verify();
}
}

Testing a Mono:

@Test
void testSimpleMono() {
Mono<String> mono = Mono.just("Hello");
StepVerifier.create(mono)
.expectNext("Hello")
.expectComplete()
.verify();
}

Advanced Verification Techniques

1. Using expectNextCount for Large Sequences

@Test
void testNextCount() {
Flux<Integer> flux = Flux.range(1, 100);
StepVerifier.create(flux)
.expectNextCount(100)
.expectComplete()
.verify();
}

2. Combining expectNext and expectNextCount

@Test
void testMixedVerification() {
Flux<String> flux = Flux.just("A", "B", "C", "D", "E");
StepVerifier.create(flux)
.expectNext("A")
.expectNextCount(2) // Verifies "B" and "C" without checking values
.expectNext("D", "E") // Can verify multiple values at once
.expectComplete()
.verify();
}

3. Testing Errors

@Test
void testError() {
Flux<String> flux = Flux.error(new RuntimeException("Something went wrong!"));
StepVerifier.create(flux)
.expectError()
.verify();
}
@Test
void testSpecificError() {
Flux<String> flux = Flux.error(new IllegalArgumentException("Invalid input"));
StepVerifier.create(flux)
.expectError(IllegalArgumentException.class)
.verify();
}
@Test
void testErrorMessage() {
Flux<String> flux = Flux.error(new RuntimeException("Database connection failed"));
StepVerifier.create(flux)
.expectErrorMessage("Database connection failed")
.verify();
}

4. Testing with Consumers

@Test
void testWithConsumers() {
Flux<Integer> flux = Flux.range(1, 3);
StepVerifier.create(flux)
.expectNextMatches(value -> value > 0) // Predicate verification
.expectNext(2)
.assertNext(value -> {
// Custom assertions
assert value == 3;
assert value instanceof Integer;
})
.expectComplete()
.verify();
}

Testing Time-Based Operations

One of StepVerifier's most powerful features is its ability to control virtual time.

Without Virtual Time (Flaky):

@Test
void testDelayWithoutVirtualTime() {
Flux<String> flux = Flux.just("A", "B")
.delayElements(Duration.ofSeconds(1)); // Would make test slow and flaky
// This would take 2+ seconds to run!
StepVerifier.create(flux)
.expectNext("A")
.expectNext("B")
.expectComplete()
.verify();
}

With Virtual Time (Recommended):

@Test
void testDelayWithVirtualTime() {
StepVerifier.withVirtualTime(() -> 
Flux.just("A", "B")
.delayElements(Duration.ofHours(1)) // Virtual time ignores this
)
.expectSubscription()
.thenAwait(Duration.ofHours(2)) // Advance virtual time
.expectNext("A", "B")
.expectComplete()
.verify();
}

Advanced Virtual Time Usage:

@Test
void testComplexVirtualTime() {
StepVerifier.withVirtualTime(() -> 
Flux.interval(Duration.ofDays(1))
.take(5)
.map(i -> "Day " + (i + 1))
)
.expectSubscription()
.thenAwait(Duration.ofDays(5)) // Fast-forward 5 days
.expectNext("Day 1", "Day 2", "Day 3", "Day 4", "Day 5")
.expectComplete()
.verify();
}

Testing Context in Reactive Streams

Reactive streams often use Context for passing metadata. StepVerifier can verify this:

@Test
void testContext() {
Mono<String> mono = Mono.deferContextual(ctx -> 
Mono.just("Hello, " + ctx.get("user"))
).contextWrite(ctx -> ctx.put("user", "Alice"));
StepVerifier.create(mono)
.expectAccessibleContext()
.contains("user", "Alice")
.then()
.expectNext("Hello, Alice")
.expectComplete()
.verify();
}

Testing Back Pressure

StepVerifier can simulate and verify back pressure behavior:

@Test
void testBackPressure() {
Flux<Integer> flux = Flux.range(1, 5);
StepVerifier.create(flux, 1) // Request 1 element initially
.expectNext(1)
.thenRequest(2) // Request 2 more elements
.expectNext(2, 3)
.thenRequest(2) // Request remaining elements
.expectNext(4, 5)
.expectComplete()
.verify();
}

Real-World Testing Scenarios

Scenario 1: Testing a Reactive Service

@Service
class UserService {
private final UserRepository userRepository;
public UserService(UserRepository userRepository) {
this.userRepository = userRepository;
}
public Flux<User> getActiveUsers() {
return userRepository.findAll()
.filter(User::isActive)
.switchIfEmpty(Flux.error(new NoActiveUsersException()));
}
public Mono<User> findUserById(String id) {
return userRepository.findById(id)
.timeout(Duration.ofSeconds(5))
.switchIfEmpty(Mono.error(new UserNotFoundException()));
}
}
class UserServiceTest {
@Test
void testGetActiveUsers() {
UserService service = new UserService(new TestUserRepository());
StepVerifier.create(service.getActiveUsers())
.expectNextMatches(User::isActive)
.expectNextCount(2)
.expectComplete()
.verify();
}
@Test
void testGetActiveUsers_NoneFound() {
UserService service = new UserService(new EmptyUserRepository());
StepVerifier.create(service.getActiveUsers())
.expectError(NoActiveUsersException.class)
.verify();
}
@Test 
void testFindUserById_Timeout() {
UserService service = new UserService(new SlowUserRepository());
StepVerifier.create(service.findUserById("slow-user"))
.expectError(TimeoutException.class)
.verify();
}
}

Scenario 2: Testing Error Recovery

@Test
void testErrorRecovery() {
Flux<String> flux = Flux.just("data1", "data2")
.map(data -> {
if ("data2".equals(data)) {
throw new RuntimeException("Processing failed");
}
return data;
})
.onErrorResume(e -> Flux.just("fallback1", "fallback2"));
StepVerifier.create(flux)
.expectNext("data1")
.expectNext("fallback1", "fallback2")
.expectComplete()
.verify();
}

Best Practices for StepVerifier

1. Always Call verify(): StepVerifier doesn't execute until verify() is called
2. Use Virtual Time for Delays: Avoid slow tests with real-time delays
3. Test Error Scenarios: Verify both happy path and error conditions
4. Verify Completion: Always check for expectComplete() or expectError()
5. Use Descriptive Tests: Name tests to describe the expected behavior
6. Test Back Pressure: Ensure your streams handle flow control properly

// Good practice example
@Test
void givenEmptyInput_whenProcessingFlux_thenReturnEmptyFlux() {
Flux<String> emptyFlux = processData(Flux.empty());
StepVerifier.create(emptyFlux)
.expectNextCount(0)
.expectComplete()
.verify(Duration.ofSeconds(1)); // Add timeout to prevent hanging
}

Common Pitfalls and Solutions

Pitfall 1: Forgetting verify()

// WRONG - Test passes but doesn't actually verify anything
StepVerifier.create(flux)
.expectNext("data")
.expectComplete();
// Missing .verify()
// CORRECT
StepVerifier.create(flux)
.expectNext("data")
.expectComplete()
.verify();

Pitfall 2: Incorrect Ordering

// WRONG - expectComplete before expectNext
StepVerifier.create(flux)
.expectComplete()
.expectNext("data")  // This will never happen after complete!
.verify();
// CORRECT
StepVerifier.create(flux)
.expectNext("data")
.expectComplete()
.verify();

Conclusion

StepVerifier is an indispensable tool for anyone working with reactive streams in Java. It provides:

• Predictable Testing: Makes asynchronous code testable synchronously
• Comprehensive Verification: Covers values, completion, errors, timing, and context
• Time Control: Virtual time enables fast testing of time-based operations
• Back Pressure Testing: Verifies flow control mechanisms
• Integration Support: Works seamlessly with Spring WebFlux and other reactive frameworks

By mastering StepVerifier, you can build reliable, well-tested reactive applications with confidence, ensuring your data streams behave exactly as expected under all conditions.