java

6.3 WebFlux 测试

Spring WebFlux 测试工具链：WebTestClient 绑定 controller、router 与完整应用上下文、请求构建与响应断言、JSON Path 断言、Mock 服务替换与安全上下文注入、以及 WebFilter 测试。

Leeting Yan 3 分钟阅读 1108 字

WebTestClient 三种绑定方式

WebTestClient 支持三种构建方式，适用于不同测试场景：

1. 绑定单个 Controller

import org.springframework.test.web.reactive.server.WebTestClient;

WebTestClient client = WebTestClient.bindToController(new UserController())
    .build();

client.get().uri("/api/users")
    .exchange()
    .expectStatus().isOk()
    .expectBodyList(User.class).hasSize(2);

适用于单元测试：只测 controller 逻辑，不启动完整 Spring 容器。

2. 绑定 RouterFunction

WebTestClient client = WebTestClient.bindToRouterFunction(orderRouter)
    .build();

client.post().uri("/api/orders")
    .bodyValue(new Order("BTC", 1, 40000))
    .exchange()
    .expectStatus().isOk()
    .expectBody()
    .jsonPath("$.symbol").isEqualTo("BTC");

适用于测试函数式路由。

3. 绑定完整应用（集成测试）

import org.springframework.boot.test.autoconfigure.web.reactive.AutoConfigureWebTestClient;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.beans.factory.annotation.Autowired;

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@AutoConfigureWebTestClient   // 没有它，WebTestClient Bean 不会自动注册
public class IntegrationTest {

    @Autowired
    private WebTestClient webTestClient;

    @Test
    void testGetUser() {
        webTestClient.get().uri("/api/users/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody()
            .jsonPath("$.id").isEqualTo(1);
    }
}

@SpringBootTest 启动完整 Spring 容器，WebFlux 在随机端口上运行。这是最接近生产环境的测试方式。

请求构建

webTestClient.post().uri("/api/orders")
    .header("Authorization", "Bearer token123")
    .header("X-Request-ID", "req-456")
    .contentType(MediaType.APPLICATION_JSON)
    .bodyValue(new Order("BTC", 1, 40000))
    .exchange();

支持链式添加多个 header、cookie、查询参数：

webTestClient.get()
    .uri(uriBuilder -> uriBuilder
        .path("/api/orders")
        .queryParam("status", "PENDING")
        .queryParam("limit", 10)
        .build())
    .cookie("session", "abc123")
    .exchange();

响应断言

状态码

.expectStatus().isOk()              // 200
.expectStatus().isCreated()         // 201
.expectStatus().isBadRequest()      // 400
.expectStatus().isNotFound()        // 404
.expectStatus().isEqualTo(503)      // 任意状态码

响应头

.expectHeader().contentType(MediaType.APPLICATION_JSON)
.expectHeader().valueEquals("X-Request-ID", "req-456")
.expectHeader().exists("X-RateLimit-Remaining")

响应体

// JSON Path 断言（最常用）
.expectBody()
.jsonPath("$.code").isEqualTo("6001")
.jsonPath("$.message").isNotEmpty()
.jsonPath("$.data.orders[0].symbol").isEqualTo("BTC")
.jsonPath("$.data.orders.length()").isEqualTo(3);

// 反序列化为对象
.expectBody(OrderResponse.class)
.value(resp -> assertEquals("BTC", resp.symbol()));

// 列表
.expectBodyList(Order.class)
.hasSize(10)
.contains(new Order("BTC", 1, 40000));

// 空体
.expectBody().isEmpty();

Mock 服务替换

测试中替换真实依赖为 Mock：

@TestConfiguration
static class TestConfig {
    @Bean
    @Primary
    public OrderService mockOrderService() {
        OrderService mock = Mockito.mock(OrderService.class);
        when(mock.findById(1L)).thenReturn(
            Mono.just(new Order(1L, "BTC", 1, 40000)));
        when(mock.findById(999L)).thenReturn(Mono.empty());
        return mock;
    }
}

@SpringBootTest
@Import(TestConfig.class)
public class MockedIntegrationTest {
    // ...
}

@Primary 让 mock bean 覆盖真实 bean。Spring Boot 3.4+ 更推荐 @MockitoBean（@MockBean 已弃用）：

import org.springframework.test.context.bean.override.mockito.MockitoBean;

@SpringBootTest
public class MockedTest {
    @MockitoBean
    private OrderService orderService;

    @Test
    void testWithMock() {
        when(orderService.findById(1L))
            .thenReturn(Mono.just(new Order(1L, "BTC", 1, 40000)));

        webTestClient.get().uri("/api/orders/1")
            .exchange()
            .expectStatus().isOk();
    }
}

安全上下文注入

测试受保护端点时，需要模拟登录用户。注意 @WithMockUser 依赖 SecurityContextHolder 的线程本地存储，在 WebFlux + WebTestClient 中通常不生效；WebFlux 要用 SecurityMockServerConfigurers.mockUser() 把认证信息写进请求的 Reactor Context：

import static org.springframework.security.test.web.reactive.server.SecurityMockServerConfigurers.mockUser;

@Test
void testUserAccess() {
    webTestClient
        .mutateWith(mockUser("alice").roles("USER"))
        .get().uri("/api/orders/my")
        .exchange()
        .expectStatus().isOk();
}

@Test
void testAdminAccess() {
    webTestClient
        .mutateWith(mockUser("admin").roles("ADMIN"))
        .delete().uri("/api/orders/1")
        .exchange()
        .expectStatus().isOk();
}

@Test
void testAnonymousAccessDenied() {
    webTestClient.get().uri("/api/orders/my")
        .exchange()
        .expectStatus().isUnauthorized();
}

mockUser() 会为这次请求注入认证信息，不需要修改全局安全上下文。需要依赖：

<dependency>
    <groupId>org.springframework.security</groupId>
    <artifactId>spring-security-test</artifactId>
    <scope>test</scope>
</dependency>

测试 WebFilter

WebFilter 测试需要完整容器：

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class FilterTest {

    @Autowired
    private WebTestClient webTestClient;

    @Test
    void testRequestIdHeader() {
        webTestClient.get().uri("/api/status")
            .exchange()
            .expectStatus().isOk()
            .expectHeader().exists("X-Request-ID");  // 由 RequestIdFilter 添加
    }
}

也可以直接测试 Filter：

@Test
void testLoggingFilter() {
    LoggingWebFilter filter = new LoggingWebFilter();
    MockServerWebExchange exchange = MockServerWebExchange.from(
        MockServerHttpRequest.get("/test").build());

    filter.filter(exchange, chain -> Mono.empty()).block();

    // 断言副作用（如日志输出）
}

SSE 端点测试

@Test
void testSseStream() {
    webTestClient.get().uri("/sse/ticks")
        .accept(MediaType.TEXT_EVENT_STREAM)
        .exchange()
        .expectStatus().isOk()
        .expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
        .returnResult(String.class)
        .getResponseBody()
        .take(3)
        .collectList()
        .block()
        .forEach(event -> {
            // returnResult(String.class) 解码的是 SSE 的 data 字段，不含 "data:" 前缀
            assertTrue(event.startsWith("{"));
        });
}

.returnResult() 获取 FluxExchangeResult，通过 .getResponseBody() 消费 SSE 流。.take(3) 只取前 3 个事件，防止测试无限等待。

错误响应测试

@Test
void testNotFound() {
    webTestClient.get().uri("/api/users/999")
        .exchange()
        .expectStatus().isNotFound()
        .expectBody()
        .jsonPath("$.code").isEqualTo("4201")
        .jsonPath("$.status").isEqualTo(404);
}

@Test
void testValidationError() {
    webTestClient.post().uri("/api/orders")
        .bodyValue(new Order("", -1, -100))  // 无效数据
        .exchange()
        .expectStatus().isBadRequest()
        .expectBody()
        .jsonPath("$.code").isEqualTo("4003");
}

eventd v2 测试策略

eventd v2 的测试分层：

单元测试（bindToController）：
    UserControllerTest —— 只测 controller，mock service

集成测试（@SpringBootTest）：
    OrderApiTest —— 启动完整容器，测端到端请求响应
    SecurityTest —— 测认证、授权、CORS

契约测试：
    ApiContractTest —— 断言响应 JSON 结构符合 OpenAPI 定义

小结

能力API场景
单元测试bindToController()只测 controller 逻辑
路由测试bindToRouterFunction()函数式路由
集成测试@SpringBootTest + Autowired WebTestClient端到端
Mock@MockitoBean / @Primary test config替换外部依赖（@MockBean 已弃用）
安全测试mockUser() + mutateWithWebFlux 不用 @WithMockUser
JSON 断言.jsonPath()快速验证响应结构
SSE 测试.returnResult().getResponseBody().take(n)消费流式响应

第 6 章结束，第二部分（WebFlux）完成。下一部分进入 R2DBC：7.1 R2DBC 驱动原理 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南