Testing Reactive Code with StepVerifier

Introduction

This tutorial ends the Reactive Spring module by testing reactive code. A Mono or Flux is a description of signals that arrive later, so ordinary assertions on a return value do not work. Reactor's StepVerifier subscribes to a publisher and checks its signals one step at a time: the elements, their order, completion or the error. We will test operators, errors and empty results with it, see the report of a failing expectation, and use virtual time to test a pipeline that runs for three hours - in milliseconds. TestPublisher will feed hand-made signals into the code under test, and finally @WebFluxTest with WebTestClient will test a WebFlux controller, including a streaming endpoint. Every test run below comes from running the tests with JUnit 6 and Reactor 3.8.

Why not just block()?

The quickest way to test a Mono is assertThat(mono.block()).isEqualTo(...), and for a single value that is acceptable. But block() cannot tell a Flux that completed after three elements from one that would have emitted a fourth, does not check that a stream completes rather than hangs, loses the difference between an empty Mono and a null, and makes a test of a one-hour delay take an hour. StepVerifier, in the reactor-test library (part of Spring Boot's test starter), checks the signals themselves -

StepVerifier.create(publisher) starts a script;
expectNext(...), expectNextCount(n), expectNextMatches(predicate) and assertNext(consumer) describe the elements, in order;
expectComplete(), expectError(...) and their variants describe the end;
verify() - or a shortcut such as verifyComplete() - subscribes, runs the script, and fails the test at the first signal that does not match. Without it, nothing is tested at all.

The code under test

A small price service with typical reactive logic: a running minimum over a stream of prices, a lookup that fails for an unknown book, a remote call with a timeout and a fallback, and an hourly report -

src/main/java/.../testing/PriceFeed.java

//PriceFeed.java
package com.javabrahman.reactive.testing;

import reactor.core.publisher.Mono;

public interface PriceFeed {
  Mono<Integer> currentPrice(String isbn);   // a remote call
}

src/main/java/.../testing/PriceService.java

//PriceService.java
package com.javabrahman.reactive.testing;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class PriceService {
  private static final Map<String, String> TITLES = Map.of("111", "Effective Java", "222", "Clean Code");
  private final PriceFeed feed;

  public PriceService(PriceFeed feed) {
    this.feed = feed;
  }

  // emits a price whenever it is lower than every price before it
  public Flux<Integer> newLows(Flux<Integer> prices) {
    return prices.scan(Math::min).distinctUntilChanged();
  }

  public Mono<String> title(String isbn) {
    return Mono.justOrEmpty(TITLES.get(isbn))
        .switchIfEmpty(Mono.error(new IllegalArgumentException("unknown isbn " + isbn)));
  }

  // the remote price, or the list price 450 if the feed does not answer within 2 seconds
  public Mono<Integer> price(String isbn) {
    return feed.currentPrice(isbn)
        .timeout(Duration.ofSeconds(2))
        .onErrorReturn(TimeoutException.class, 450);
  }

  public Flux<String> hourlyReport() {
    return Flux.interval(Duration.ofHours(1)).map(i -> "report " + (i + 1)).take(3);
  }
}

Elements, completion, errors and empty results

src/test/java/.../testing/PriceServiceTest.java

//PriceServiceTest.java
package com.javabrahman.reactive.testing;

import static org.assertj.core.api.Assertions.assertThat;

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

class PriceServiceTest {
  private final PriceService service = new PriceService(isbn -> Mono.just(399));

  @Test
  void emitsOnlyNewLows() {
    StepVerifier.create(service.newLows(Flux.just(450, 470, 420, 430, 399, 399)))
        .expectNext(450, 420, 399)
        .verifyComplete();
  }

  @Test
  void checksElementsWithConditions() {
    StepVerifier.create(service.newLows(Flux.range(1, 100).map(i -> 1000 - i)))
        .expectNextMatches(price -> price == 999)
        .expectNextCount(98)
        .assertNext(last -> assertThat(last).isEqualTo(900))
        .verifyComplete();
  }

  @Test
  void failsForAnUnknownIsbn() {
    StepVerifier.create(service.title("999"))
        .expectErrorSatisfies(e -> assertThat(e)
            .isInstanceOf(IllegalArgumentException.class)
            .hasMessage("unknown isbn 999"))
        .verify();
  }

  @Test
  void anEmptyStreamOnlyCompletes() {
    StepVerifier.create(service.newLows(Flux.empty()))
        .verifyComplete();   // no onNext at all
  }

  @Test
  void cancelsAnInfiniteStream() {
    StepVerifier.create(service.newLows(Flux.generate(sink -> sink.next(500))))
        .expectNext(500)
        .thenCancel()        // the source never completes; stop after the first element
        .verify();
  }
}

OUTPUT of running the tests

╷
└─ JUnit Jupiter ✔
   └─ PriceServiceTest ✔
      ├─ emitsOnlyNewLows() ✔
      ├─ checksElementsWithConditions() ✔
      ├─ failsForAnUnknownIsbn() ✔
      ├─ cancelsAnInfiniteStream() ✔
      └─ anEmptyStreamOnlyCompletes() ✔

[         2 containers found      ]
[         0 containers skipped    ]
[         2 containers started    ]
[         0 containers aborted    ]
[         2 containers successful ]
[         0 containers failed     ]
[         5 tests found           ]
[         0 tests skipped         ]
[         5 tests started         ]
[         0 tests aborted         ]
[         5 tests successful      ]
[         0 tests failed          ]

expectNext(450, 420, 399) followed by verifyComplete() checks the exact sequence and that nothing more follows: a fourth element would fail the test.
For long streams, expectNextCount() skips over elements, and assertNext() runs any AssertJ assertion on one.
An error is the expected end of failsForAnUnknownIsbn(); expectErrorSatisfies() inspects it, and expectError(Type.class) or verifyErrorMessage() are shorter forms.
An infinite stream is ended by the script with thenCancel().

When an expectation fails

A test that expects 400 where the stream emits 420 -

src/test/java/.../testing/WrongExpectationTest.java

//WrongExpectationTest.java
package com.javabrahman.reactive.testing;

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

class WrongExpectationTest {
  private final PriceService service = new PriceService(isbn -> Mono.just(399));

  @Test
  void expectsTheWrongLow() {
    StepVerifier.create(service.newLows(Flux.just(450, 470, 420)))
        .expectNext(450)
        .expectNext(400)
        .verifyComplete();
  }
}

OUTPUT of running the test (the relevant lines)

      └─ expectsTheWrongLow() ✘ expectation "expectNext(400)" failed (expected value: 400; actual value: 420)
[         0 tests successful      ]
[         1 tests failed          ]

The message names the step that failed - expectNext(400) - with the expected and the actual value. Naming steps with .as("description") makes such reports easier to read in long scripts.

Virtual time

hourlyReport() emits three reports, one hour apart. A real-time test would take three hours, and the timeout of price() would cost two seconds per test. StepVerifier.withVirtualTime() replaces Reactor's time-based schedulers with a VirtualTimeScheduler whose clock only moves when the script says so: thenAwait(duration) advances it, and expectNoEvent(duration) advances it while checking that nothing happens -

src/test/java/.../testing/VirtualTimeTest.java

//VirtualTimeTest.java
package com.javabrahman.reactive.testing;

import static org.assertj.core.api.Assertions.assertThat;

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

class VirtualTimeTest {

  @Test
  void threeHourlyReportsInMilliseconds() {
    PriceService service = new PriceService(isbn -> Mono.just(399));
    Duration took = StepVerifier.withVirtualTime(service::hourlyReport)   // a Supplier: built inside virtual time
        .expectSubscription()
        .expectNoEvent(Duration.ofHours(1).minusSeconds(1))
        .thenAwait(Duration.ofSeconds(1))
        .expectNext("report 1")
        .thenAwait(Duration.ofHours(2))
        .expectNext("report 2", "report 3")
        .verifyComplete();
    assertThat(took).isLessThan(Duration.ofSeconds(1));   // three virtual hours, well under a real second
  }

  @Test
  void fallsBackWhenTheFeedNeverAnswers() {
    PriceService service = new PriceService(isbn -> Mono.never());   // a remote call that hangs
    StepVerifier.withVirtualTime(() -> service.price("111"))
        .expectSubscription()
        .expectNoEvent(Duration.ofSeconds(2))
        .expectNext(450)
        .verifyComplete();
  }

  @Test
  void usesTheFeedWhenItAnswersInTime() {
    PriceService service = new PriceService(isbn -> Mono.just(399).delayElement(Duration.ofSeconds(1)));
    StepVerifier.withVirtualTime(() -> service.price("111"))
        .thenAwait(Duration.ofSeconds(1))
        .expectNext(399)
        .verifyComplete();
  }
}

OUTPUT of running the tests

╷
└─ JUnit Jupiter ✔
   └─ VirtualTimeTest ✔
      ├─ fallsBackWhenTheFeedNeverAnswers() ✔
      ├─ usesTheFeedWhenItAnswersInTime() ✔
      └─ threeHourlyReportsInMilliseconds() ✔

[         2 containers found      ]
[         0 containers skipped    ]
[         2 containers started    ]
[         0 containers aborted    ]
[         2 containers successful ]
[         0 containers failed     ]
[         3 tests found           ]
[         0 tests skipped         ]
[         3 tests started         ]
[         0 tests aborted         ]
[         3 tests successful      ]
[         0 tests failed          ]

Three hours of reports, a two-second timeout and a one-second delay, all verified in a fraction of a second; verify() returns the real duration, which the first test asserts. Two rules: the publisher must be created inside the supplier passed to withVirtualTime(), so that its operators pick up the virtual scheduler; and the script should start with expectSubscription() before expectNoEvent(), because the subscription itself counts as an event. withVirtualTime replaces the clock of Reactor's schedulers - thenAwait and expectNoEvent move virtual time forward instantly, so a pipeline with hour-long delays is verified in milliseconds

withVirtualTime replaces the clock of Reactor's schedulers - thenAwait and expectNoEvent move virtual time forward instantly, so a pipeline with hour-long delays is verified in milliseconds

TestPublisher: signals made by hand

To test how code reacts to a source, it helps to control that source exactly. TestPublisher emits whatever the test tells it to, when it tells it to - and can also misbehave, to test robustness -

src/test/java/.../testing/TestPublisherTest.java

//TestPublisherTest.java
package com.javabrahman.reactive.testing;

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;

class TestPublisherTest {
  private final PriceService service = new PriceService(isbn -> Mono.just(399));

  @Test
  void reactsToEachPriceAsItArrives() {
    TestPublisher<Integer> prices = TestPublisher.create();
    StepVerifier.create(service.newLows(prices.flux()))
        .then(() -> prices.next(450))
        .expectNext(450)
        .then(() -> prices.next(470, 440))
        .expectNext(440)
        .then(() -> prices.error(new IllegalStateException("feed lost")))
        .expectErrorMessage("feed lost")
        .verify();
    prices.assertWasSubscribed();
    prices.assertNoRequestOverflow();   // the code never received more than it requested
  }
}

OUTPUT of running the tests

╷
└─ JUnit Jupiter ✔
   └─ TestPublisherTest ✔
      └─ reactsToEachPriceAsItArrives() ✔

[         2 containers found      ]
[         0 containers skipped    ]
[         2 containers started    ]
[         0 containers aborted    ]
[         2 containers successful ]
[         0 containers failed     ]
[         1 tests found           ]
[         0 tests skipped         ]
[         1 tests started         ]
[         0 tests aborted         ]
[         1 tests successful      ]
[         0 tests failed          ]

then() runs code between expectations - here, emitting the next prices - so the test controls the exact interleaving of input and output. The service passed the error on, and never received more elements than it requested.

Testing a WebFlux controller with WebTestClient

For the web layer, @WebFluxTest is the reactive counterpart of @WebMvcTest: a slice with only the WebFlux infrastructure and the named controller, and a WebTestClient that calls it without a real server. The controller under test, with a JSON list, a 404 and an NDJSON stream -

src/main/java/.../testing/web/BookController.java

//BookController.java
package com.javabrahman.reactive.testing.web;

import java.time.Duration;
import java.util.Map;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
public class BookController {
  private static final Map<Long, Book> BOOKS = Map.of(1L, new Book(1, "Effective Java", 450), 2L, new Book(2, "Clean Code", 380));

  public record Book(long id, String title, int price) {
  }

  @GetMapping("/books/{id}")
  public Mono<Book> one(@PathVariable long id) {
    return Mono.justOrEmpty(BOOKS.get(id))
        .switchIfEmpty(Mono.error(new ResponseStatusException(HttpStatus.NOT_FOUND)));
  }

  @GetMapping(value = "/books/stream", produces = MediaType.APPLICATION_NDJSON_VALUE)
  public Flux<Book> stream() {
    return Flux.just(BOOKS.get(1L), BOOKS.get(2L)).delayElements(Duration.ofMillis(100));
  }
}

src/main/java/.../testing/web/TestApp.java

//TestApp.java
package com.javabrahman.reactive.testing.web;

import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class TestApp {
}

src/test/java/.../testing/web/BookControllerTest.java

//BookControllerTest.java
package com.javabrahman.reactive.testing.web;

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.webflux.test.autoconfigure.WebFluxTest;
import org.springframework.http.MediaType;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

@WebFluxTest(BookController.class)
class BookControllerTest {

  @Autowired
  WebTestClient client;

  @Test
  void returnsABook() {
    client.get().uri("/books/1").exchange()
        .expectStatus().isOk()
        .expectBody()
        .jsonPath("$.title").isEqualTo("Effective Java")
        .jsonPath("$.price").isEqualTo(450);
  }

  @Test
  void answers404ForAnUnknownBook() {
    client.get().uri("/books/42").exchange()
        .expectStatus().isNotFound();
  }

  @Test
  void streamsBooksOneByOne() {
    Flux<BookController.Book> body = client.get().uri("/books/stream")
        .accept(MediaType.APPLICATION_NDJSON)
        .exchange()
        .expectStatus().isOk()
        .returnResult(BookController.Book.class)
        .getResponseBody();
    StepVerifier.create(body)
        .expectNextMatches(book -> book.title().equals("Effective Java"))
        .expectNextMatches(book -> book.title().equals("Clean Code"))
        .verifyComplete();
  }
}

OUTPUT of running the tests

╷
└─ JUnit Jupiter ✔
   └─ BookControllerTest ✔
      ├─ answers404ForAnUnknownBook() ✔
      ├─ streamsBooksOneByOne() ✔
      └─ returnsABook() ✔

[         2 containers found      ]
[         0 containers skipped    ]
[         2 containers started    ]
[         0 containers aborted    ]
[         2 containers successful ]
[         0 containers failed     ]
[         3 tests found           ]
[         0 tests skipped         ]
[         3 tests started         ]
[         0 tests aborted         ]
[         3 tests successful      ]
[         0 tests failed          ]

expectBody().jsonPath() checks a JSON response, and returnResult().getResponseBody() hands a streaming body back as a Flux, which StepVerifier verifies element by element. Spring Boot 4 places @WebFluxTest in spring-boot-webflux-test (package org.springframework.boot.webflux.test.autoconfigure), included with spring-boot-starter-webflux-test. WebTestClient also works against a full @SpringBootTest on a random port, and even against a Spring MVC application.

Habits for reactive tests

Always end with a verify call: a StepVerifier script without one never subscribes, and the test passes without checking anything.
Check the end of the stream: verifyComplete() or an error expectation, so that extra elements and missing completions fail.
Use virtual time for anything with delays, intervals, timeouts or retries with backoff, and build the publisher inside the supplier.
Set a safety net: verify(Duration.ofSeconds(5)), or StepVerifier.setDefaultTimeout(), turns a stream that hangs into a failure instead of a frozen build.
Test blocking behaviour too: the BlockHound agent makes tests fail when code blocks on a non-blocking thread.

Summary

In this tutorial we tested reactive code with Reactor's test tools. StepVerifier checked a running minimum element by element and its completion, inspected an error, verified an empty stream and cancelled an infinite one, and reported exactly which step failed when an expectation was wrong. Virtual time verified three hourly reports, a two-second timeout and a delay in well under a second, and TestPublisher fed prices and an error into the service under the test's control. @WebFluxTest with WebTestClient tested a controller's JSON, its 404 and an NDJSON stream. This completes the Reactive Spring module: Reactor, WebFlux, WebClient, R2DBC, SSE and WebSockets, and when to choose virtual threads instead.

Reactive Spring Tutorials
Virtual Threads vs WebFlux: Which to Choose in 2026 Reactive Programming with Project Reactor: Mono and Flux Building REST APIs with Spring WebFlux Calling HTTP APIs with Spring WebClient Reactive Database Access with Spring Data R2DBC Server-Sent Events and WebSockets in Spring WebFlux