如何使用REST-Assured测试Server-Sent Events端点？测试无法终止求解

这个问题我之前也碰到过！REST-Assured默认的处理方式不太适配SSE这种持久化HTTP流——服务器不会主动关闭连接，所以你的readText()会一直阻塞，导致测试无法终止。下面给你两个实用的解决方案：

方案一：基于RestAssured原生API的流式读取 #

不需要额外依赖，直接利用RestAssured的响应流逐行读取，直到拿到目标事件后主动关闭连接：

import org.hamcrest.MatcherAssert.assertThat
import org.hamcrest.core.Is.`is`
import io.restassured.RestAssured
import org.springframework.http.MediaType

fun testSseEndpoint() {
    val response = RestAssured.given()
        .contentType(MediaType.SERVER_SENT_EVENTS)
        .get("/status/stream")

    val reader = response.body.asReader()
    var line: String?
    var targetEvent: String? = null
    val timeoutMillis = 5000 // 设置5秒超时，避免无限等待
    val startTime = System.currentTimeMillis()

    // 逐行读取SSE流，直到找到目标事件或超时
    while (System.currentTimeMillis() - startTime < timeoutMillis && targetEvent == null) {
        line = reader.readLine()
        line?.let {
            // 匹配SSE的event字段，根据实际返回格式调整
            if (it.startsWith("event: my-event")) {
                targetEvent = it.substring(7).trim()
            }
            // 如果是验证data负载，就匹配data行：if (it.startsWith("data:")) { ... }
        }
    }

    // 必须主动关闭响应，释放HTTP连接
    response.close()

    assertThat("未收到预期的SSE事件", targetEvent, `is`("my-event"))
}

方案二：使用OkHttp的SSE客户端（更专业的方式） #

REST-Assured可以配置使用OkHttp作为底层客户端，配合OkHttp的EventSource原生处理SSE事件，代码更清晰，也更符合SSE规范：

第一步：配置RestAssured使用OkHttp #

import io.restassured.RestAssured
import io.restassured.config.HttpClientConfig
import io.restassured.http.client.OkHttpClientFactory

// 初始化时配置RestAssured用OkHttp作为底层客户端
RestAssured.config = RestAssured.config()
    .httpClient(HttpClientConfig.httpClientConfig()
        .httpClientFactory(OkHttpClientFactory()))

第二步：编写SSE测试代码 #

import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.sse.EventSource
import okhttp3.sse.EventSourceListener
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import org.junit.Assert.*

fun testSseWithOkHttp() {
    val client = OkHttpClient()
    val request = Request.Builder()
        .url("${RestAssured.baseURI}/status/stream")
        .header("Accept", MediaType.SERVER_SENT_EVENTS_VALUE)
        .build()

    val eventLatch = CountDownLatch(1)
    var capturedEvent: String? = null

    val listener = object : EventSourceListener() {
        override fun onEvent(eventSource: EventSource, id: String?, type: String?, data: String) {
            // 匹配目标事件类型
            if ("my-event" == type) {
                capturedEvent = type
                eventLatch.countDown() // 触发计数锁，结束等待
                eventSource.cancel() // 主动关闭SSE连接
            }
        }

        override fun onFailure(eventSource: EventSource, t: Throwable?, response: okhttp3.Response?) {
            // 出错时也要释放锁，避免测试卡住
            eventLatch.countDown()
            t?.printStackTrace()
        }
    }

    // 启动SSE事件监听
    EventSource.Factory(client).newEventSource(request, listener)

    // 等待事件触发或超时
    val isSuccess = eventLatch.await(5, TimeUnit.SECONDS)

    assertTrue("超时未收到目标SSE事件", isSuccess)
    assertEquals("my-event", capturedEvent)
}

关键注意点 #

• 主动关闭连接：不管用哪种方式，都要主动关闭响应或EventSource，否则HTTP连接会一直占用，导致测试卡住
• 设置超时：必须给测试加超时时间，避免因为服务器异常导致测试无限等待
• SSE格式解析：SSE的消息是按行分隔的，每行以event:、data:、id:开头，空行表示一个事件结束，解析时要对应实际的返回格式

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