Testing the Reactive Stack: runTest, StepVerifier, and WebTestClient

The single hardest thing about testing reactive and concurrent code is that the work does not finish when the line of code returns. A suspend function suspends, a Flux emits later, a coroutine launches onto another dispatcher. If your test just calls the code and asserts immediately, you are racing the runtime and your assertions will pass or fail at random. The whole point of the testing tools in this lesson is to make asynchronous execution deterministic: you tell the framework to drive the work to completion, then you assert on the result. Pick the tool that matches the type you are testing — runTest for suspend and Flow, StepVerifier for Mono and Flux, and WebTestClient for HTTP endpoints — and your tests become as boring and reliable as plain synchronous tests.

For coroutines, kotlinx-coroutines-test gives you runTest. It runs the body on a TestScope backed by a special scheduler that uses virtual time, so any delay() inside the coroutine is skipped instantly rather than actually sleeping. A test that exercises code with a five-second delay finishes in microseconds. runTest also installs a StandardTestDispatcher, waits for all child coroutines launched in its scope to complete before returning, and rethrows uncaught exceptions so failures surface as test failures. You never need Thread.sleep, and you never need to add fudge-factor timeouts hoping the work finished in time.

Virtual time is the feature that turns flaky timing tests into precise ones. Inside runTest you can grab the TestScope's scheduler and step time forward yourself with advanceTimeBy(duration), advanceUntilIdle() to run everything pending, or runCurrent() to execute tasks scheduled at the current instant. This lets you assert on the exact intermediate states of time-dependent logic — debounce, retry-with-backoff, polling, timeout windows — without ever waiting in real wall-clock time. A common gotcha: if you launch work on a real dispatcher like Dispatchers.IO instead of the test dispatcher, virtual time does not apply, so inject dispatchers into your classes rather than hard-coding them.

Testing a Flow follows the same philosophy. Because a Flow is cold, nothing happens until you collect it, so the canonical pattern is to call toList() inside runTest and assert on the collected items. For flows that involve timing operators you advance virtual time between assertions. For richer, more readable Flow assertions many teams reach for the Turbine library, whose test { } block lets you await items one at a time with awaitItem(), assert completion with awaitComplete(), and assert errors with awaitError() — it also fails loudly if the flow emits something you did not consume, catching subtle bugs.

When you are working directly with Reactor types — Mono and Flux, which is what Spring WebFlux handlers and reactive repositories return — the idiomatic tool is StepVerifier from reactor-test. You wrap the publisher in StepVerifier.create(...), then describe the expected sequence step by step: expectNext for emitted values, expectError or expectErrorMatches for failures, expectComplete for normal termination, and finally verify() to actually subscribe and block until the assertions are checked. Forgetting the terminal verify() call is the classic mistake — without it nothing subscribes and the test silently passes. StepVerifier has its own virtual time support via StepVerifier.withVirtualTime, which swaps in a virtual clock so you can call thenAwait(Duration) to fast-forward delays and time-based operators deterministically.

For the HTTP layer, Spring provides WebTestClient, the reactive counterpart to MockMvc. It can bind to your full application context over a real port, bind to a single controller, or bind directly to a RouterFunction for functional endpoints, and it speaks a fluent assertion API: you build a request, exchange() it, then chain expectStatus, expectHeader, and expectBody assertions. Crucially WebTestClient handles the async nature of WebFlux for you — exchange() subscribes and awaits the response under the hood — so your test code reads synchronously even though the server is fully non-blocking. You can assert on the deserialized body, on raw JSON with jsonPath, or stream Server-Sent Events back as a Flux and verify them with StepVerifier.

A few practices keep the whole suite trustworthy. Always inject your dispatchers (and the Reactor Scheduler equivalents) so tests can substitute deterministic ones; hard-coded Dispatchers.IO or Schedulers.boundedElastic() defeat virtual time. Prefer testing behavior through the public suspend or reactive API rather than reaching into internal coroutine state. Add an UnconfinedTestDispatcher only when you specifically want child coroutines to run eagerly and depth-first; the default StandardTestDispatcher is usually what you want because it models realistic scheduling. And remember the bridge between worlds: a suspend function that internally consumes a Mono via .awaitSingle() can be tested with runTest, while a controller returning Flux is best tested end-to-end with WebTestClient.

Finally, treat error and edge cases as first-class. StepVerifier.expectError lets you assert the exact exception type a reactive pipeline emits; runTest with assertFailsWith verifies a suspend function throws; and WebTestClient.expectStatus().is5xxServerError() or .isBadRequest() pins down how your endpoint maps failures to HTTP. Combined with virtual time for retries and timeouts, these tools let you prove not just the happy path but also cancellation, backpressure-driven errors, and time-bounded fallbacks — the exact scenarios that are nearly impossible to reproduce reliably with naive sleep-based tests.

import kotlinx.coroutines.launch
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.runCurrent
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals

fun tickerFlow(): Flow<Int> = flow {
    repeat(3) { i ->
        delay(1000)
        emit(i)
    }
}

class TickerTest {
    @Test
    fun emitsEverySecond() = runTest {
        // toList drives the cold flow to completion; the 3s of delays
        // are virtual, so the test finishes almost instantly.
        val result = tickerFlow().toList()
        assertEquals(listOf(0, 1, 2), result)
    }

    @Test
    fun stepsThroughVirtualTime() = runTest {
        val collected = mutableListOf<Int>()
        val job = launch { tickerFlow().collect { collected.add(it) } }

        advanceTimeBy(1000); runCurrent()
        assertEquals(listOf(0), collected)

        advanceTimeBy(2000); runCurrent()
        assertEquals(listOf(0, 1, 2), collected)
        job.cancel()
    }
}

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

class ReactorTest {
    @Test
    fun verifiesFluxSequence() {
        val flux = Flux.just("a", "b", "c")
        StepVerifier.create(flux)
            .expectNext("a", "b")
            .expectNext("c")
            .verifyComplete() // subscribes and blocks until assertions checked
    }

    @Test
    fun verifiesError() {
        val mono = Mono.error<String>(IllegalStateException("boom"))
        StepVerifier.create(mono)
            .expectErrorMatches { it is IllegalStateException && it.message == "boom" }
            .verify()
    }

    @Test
    fun fastForwardsDelaysWithVirtualTime() {
        // The Flux would take 2 hours in real time; virtual time makes it instant.
        StepVerifier.withVirtualTime { Flux.interval(Duration.ofHours(1)).take(2) }
            .expectSubscription()
            .thenAwait(Duration.ofHours(2))
            .expectNext(0L, 1L)
            .verifyComplete()
    }
}

import org.junit.jupiter.api.Test
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.boot.test.context.SpringBootTest.WebEnvironment
import org.springframework.test.web.reactive.server.WebTestClient

@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT)
class UserEndpointTest(@Autowired val client: WebTestClient) {

    @Test
    fun returnsUserJson() {
        client.get().uri("/users/42")
            .exchange()
            .expectStatus().isOk
            .expectHeader().contentType("application/json")
            .expectBody()
            .jsonPath("$.id").isEqualTo(42)
            .jsonPath("$.name").isEqualTo("Ada")
    }

    @Test
    fun returns404ForMissingUser() {
        client.get().uri("/users/999")
            .exchange()
            .expectStatus().isNotFound
    }
}

import kotlinx.coroutines.reactive.awaitSingle
import kotlinx.coroutines.test.runTest
import reactor.core.publisher.Mono
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFailsWith

class PriceService(private val repo: PriceRepository) {
    suspend fun priceFor(sku: String): Int =
        repo.findPrice(sku).awaitSingle() // Mono -> suspend bridge
}

interface PriceRepository { fun findPrice(sku: String): Mono<Int> }

class PriceServiceTest {
    @Test
    fun returnsPrice() = runTest {
        val service = PriceService(object : PriceRepository {
            override fun findPrice(sku: String) = Mono.just(1500)
        })
        assertEquals(1500, service.priceFor("ABC"))
    }

    @Test
    fun propagatesError() = runTest {
        val service = PriceService(object : PriceRepository {
            override fun findPrice(sku: String) = Mono.error<Int>(NoSuchElementException())
        })
        assertFailsWith<NoSuchElementException> { service.priceFor("missing") }
    }
}