Reactor StepVerifier: Testing Reactive Streams

Reactive publishers are lazy and asynchronous, which makes them awkward to test with plain assertions: a `Flux` does nothing until something subscribes, and its values may arrive on a different thread at an unpredictable time. Project Reactor ships `StepVerifier` (in the `reactor-test` artifact) precisely for this. Instead of collecting results into a list and asserting afterwards, you write a *script of expectations* that subscribes for you, pulls each signal as it arrives, and fails loudly the moment reality diverges from your script. Nothing actually runs until you call the terminal `verify()` method, which blocks the test thread until the sequence completes, errors, or times out.

Add the dependency first. With Spring Boot the WebFlux starter brings Reactor in, and `org.springframework.boot:spring-boot-starter-test` already includes `io.projectreactor:reactor-test`, so `StepVerifier` is on your test classpath out of the box. You build a verifier with `StepVerifier.create(publisher)`, chain expectations in emission order, and end with a terminal verification. The classic happy-path shape is `expectNext(...)` for each onNext value, then `expectComplete()` to assert the stream finished cleanly, then `verify()` to actually run it. If you forget `verify()` (or `verifyComplete()`), the whole script is dead code and the test passes vacuously — this is the single most common StepVerifier mistake.

The onNext family gives you several ways to assert values. `expectNext(a, b, c)` matches several elements by equality in order; `expectNextCount(n)` matches `n` elements without caring about their content; and `expectNextMatches { it > 0 }` lets you assert with a predicate when equality is too rigid. `assertNext { assertThat(it.name).isEqualTo("Ada") }` is the most expressive option, handing each element to a lambda where you can use AssertJ or JUnit assertions on its fields. Choose the loosest expectation that still proves what the test is about — over-specifying makes tests brittle against irrelevant changes.

Terminal expectations decide how the sequence is supposed to end. `expectComplete()` asserts an onComplete with no error, and the convenience method `verifyComplete()` is exactly `expectComplete().verify()` rolled into one. For failure paths, `expectError()` asserts that the stream terminated with *any* error, `expectError(IllegalStateException::class.java)` asserts the error type, and `expectErrorMatches { it is IllegalArgumentException && it.message == "bad id" }` checks both type and message. There is also `verifyError(...)` as the fused shorthand. Crucially, an error and a completion are mutually exclusive terminal signals — a Flux that errors never emits onComplete, so script the one you actually expect.

The terminal `verify()` returns a `Duration` (the real wall-clock time the verification took) and accepts a timeout overload, `verify(Duration.ofSeconds(5))`, so a hung publisher fails the test instead of blocking your CI forever. This matters because `verify()` is blocking by design: it is the bridge from the asynchronous reactive world back to the synchronous test method. Always end with a terminal call so the subscription is created and the assertions are evaluated on the calling thread.

Time-based operators are where StepVerifier truly shines. Suppose your Flux emits with `delayElements(Duration.ofHours(1))` — you do not want a test that sleeps for hours. `StepVerifier.withVirtualTime { ... }` swaps in a `VirtualTimeScheduler` so that scheduled work runs on a fake clock you control. You then advance that clock explicitly with `thenAwait(Duration.ofHours(1))`, and the elements scheduled for that instant are released immediately. Pass the publisher as a `Supplier` lambda (not a pre-built instance) so the virtual scheduler is installed *before* the operators capture their scheduler — building the Flux outside the lambda is a subtle bug that silently falls back to the real clock.

A typical virtual-time script interleaves waiting and asserting: `expectSubscription()`, then `expectNoEvent(Duration.ofMillis(900))` to prove nothing leaks out early, then `thenAwait(Duration.ofMillis(100))` to cross the delay boundary, then `expectNext(value)`. The combination of `expectNoEvent` and `thenAwait` lets you verify timing precisely — that an element appears *after* a delay and not before. Use `expectSubscription()` as the first step under virtual time because the subscription signal itself is an event you would otherwise have to account for.

Two practical notes round this out. First, when testing Spring `WebClient` or R2DBC repositories you usually verify the returned `Mono`/`Flux` directly with StepVerifier rather than blocking — keep the reactive chain intact end to end. Second, StepVerifier is for assertions about *signals and timing*; for setting up controllable upstream sequences in tests, pair it with `Flux.just`, `Flux.error`, `Flux.range`, or a `TestPublisher` when you need to emit signals manually. Together they let you describe almost any reactive scenario as a precise, fast, deterministic test.

import org.junit.jupiter.api.Test
import reactor.core.publisher.Flux
import reactor.test.StepVerifier

class GreetingFluxTest {
    @Test
    fun `emits names in order then completes`() {
        val names: Flux<String> = Flux.just("Ada", "Linus", "Grace")

        StepVerifier.create(names)
            .expectNext("Ada")
            .expectNext("Linus", "Grace")   // several at once
            .expectComplete()
            .verify()                        // without this the test is a no-op
    }
}

import org.junit.jupiter.api.Test
import reactor.core.publisher.Flux
import reactor.test.StepVerifier

class UserFluxTest {
    data class User(val id: Long, val name: String)

    @Test
    fun `emits one user then fails`() {
        val flux: Flux<User> = Flux.concat(
            Flux.just(User(1, "Ada")),
            Flux.error(IllegalArgumentException("bad id"))
        )

        StepVerifier.create(flux)
            .assertNext { user -> check(user.name == "Ada") }
            .expectErrorMatches { it is IllegalArgumentException && it.message == "bad id" }
            .verify()
    }
}

import org.junit.jupiter.api.Test
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
import java.time.Duration

class DelayedFluxTest {
    @Test
    fun `releases element only after the delay`() {
        StepVerifier.withVirtualTime {
            Flux.just("late").delayElements(Duration.ofHours(1))
        }
            .expectSubscription()
            .expectNoEvent(Duration.ofMinutes(59))   // still silent
            .thenAwait(Duration.ofMinutes(1))        // cross the boundary
            .expectNext("late")
            .verifyComplete()                        // = expectComplete().verify()
    }
}

import kotlinx.coroutines.delay
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

class VirtualClockTest {
    private suspend fun fetchLate(): String {
        delay(60_000)   // skipped instantly by runTest's virtual time
        return "late"
    }

    @Test
    fun `delay is fast-forwarded`() = runTest {
        val result = fetchLate()
        assertEquals("late", result)
    }
}