如何用StepVerifier和WebTestClient测试Kotlin的Flux<String> Controller

我在Kotlin中使用StepVerifier测试Spring WebFlux的@RestController，Flux接口的测试完全正常，但测试Flux接口时，返回的三个字符串被拼接成了一个整体，而浏览器访问该接口能正常看到字符串逐个显示。我需要让按顺序匹配单个字符串的测试用例通过。

正常工作的Flux接口及测试 #

控制器代码 #

@GetMapping("/helloInt")
fun helloInt(): Flux<Int> {
  return Flux.range(1, 3).delayElements(Duration.ofMillis(300))
}

测试代码 #

@Test
fun callingIntEndpoint(){
  var responseBody = webTestClient
    .get()
    .uri("/helloInt")
    .exchange().returnResult<Int>()
  StepVerifier
    .create(responseBody.getResponseBody())
    .expectSubscription()
    .expectNext(1)
    .expectNext(2)
    .expectNext(3)
    .verifyComplete()
}

出现问题的Flux接口及异常测试 #

控制器代码 #

@GetMapping("/hello")
fun hello(): Flux<String> {
  return Flux.just("testing", "hello", "Kotlin")
    .delayElements(Duration.ofMillis(300)) // 确保是三条独立消息
}

未失败但不符合预期的测试代码 #

@Test
fun callingTheEndpoint() {
  webTestClient
    .get()
    .uri("/hello")
    .exchange()
    .returnResult<String>()
    .responseBody
    .test()
    .expectNextMatches { it.equals("testinghelloKotlin") } // 这并非预期结果
    .verifyComplete()
}

期望通过的测试代码 #

@Test
fun callingTheEndpoint() {
  webTestClient
    .get()
    .uri("/hello")
    .exchange()
    .returnResult<String>()
    .responseBody
    .test()
    .expectNextMatches { it.equals("testing") }
    .expectNextMatches { it.equals("hello") }
    .expectNextMatches { it.equals("Kotlin") }
    .verifyComplete()  
}

原因分析 #

Spring WebFlux处理Flux<String>时，默认会将所有字符串元素作为text/plain类型的响应体拼接返回——HTTP普通文本流是连续的，WebTestClient会等待整个流结束后读取完整内容，因此三个字符串被合并成了一个。

而Flux<Int>能正常拆分，是因为Spring默认将整数序列化为application/json格式的数组，每个整数是数组的独立元素，WebTestClient可以正确解析出单个Int对象。

浏览器能看到逐个显示，是因为它支持流式响应的实时渲染，而WebTestClient默认对普通文本流的处理逻辑不同。

解决方案 #

要让每个字符串作为独立响应元素返回，需要将接口指定为**文本事件流（text/event-stream）**格式，也就是Server-Sent Events（SSE），这样每个字符串会作为独立事件发送，WebTestClient就能识别单个元素。

修改控制器代码 #

在@GetMapping中指定生产的媒体类型：

import org.springframework.http.MediaType

@GetMapping("/hello", produces = [MediaType.TEXT_EVENT_STREAM_VALUE])
fun hello(): Flux<String> {
  return Flux.just("testing", "hello", "Kotlin")
    .delayElements(Duration.ofMillis(300))
}

优化测试代码（可选） #

显式指定接受SSE格式，让测试逻辑更严谨：

@Test
fun callingTheEndpoint() {
  webTestClient
    .get()
    .uri("/hello")
    .accept(MediaType.TEXT_EVENT_STREAM) // 指定接受SSE格式
    .exchange()
    .returnResult<String>()
    .responseBody
    .test()
    .expectNext("testing")
    .expectNext("hello")
    .expectNext("Kotlin")
    .verifyComplete()  
}

内容的提问来源于stack exchange，提问作者moldovean