FastAPI TestClient测试SSE接口多次请求失败问题求助

问题场景 #

使用pytest对FastAPI服务做单元测试时，发现fastapi.testclient.TestClient调用返回EventSourceResponse的POST接口，仅能成功执行一次，后续调用会抛出RuntimeError(f'{self!r} is bound to a different event loop')错误。但接口返回纯文本响应时无此问题，最小复现代码如下：

from fastapi.testclient import TestClient
from fastapi import FastAPI
from sse_starlette import EventSourceResponse


app = FastAPI()


@app.post("/hello")
def query_hello_1():
    def hello():
        yield "hello"
    return EventSourceResponse(
        hello(),
    )


@app.post("/hello_2")
def query_hello_2():
    def hello():
        yield "hello_2"
    return EventSourceResponse(
        hello(),
    )


client = TestClient(app)


def test_hello_1():
    print(list(client.post("/hello").iter_lines()))


def test_hello_2():
    print(list(client.post("/hello_2").iter_lines()))

解决方法 #

方法1：每个测试函数内创建新的TestClient #

不要复用全局的TestClient实例，而是在每个测试函数内部初始化客户端，确保每次测试都使用全新的实例和对应的事件循环：

from fastapi.testclient import TestClient
from fastapi import FastAPI
from sse_starlette import EventSourceResponse


app = FastAPI()


@app.post("/hello")
def query_hello_1():
    def hello():
        yield "hello"
    return EventSourceResponse(hello())


@app.post("/hello_2")
def query_hello_2():
    def hello():
        yield "hello_2"
    return EventSourceResponse(hello())


def test_hello_1():
    client = TestClient(app)
    print(list(client.post("/hello").iter_lines()))


def test_hello_2():
    client = TestClient(app)
    print(list(client.post("/hello_2").iter_lines()))

方法2：用pytest fixture管理TestClient生命周期 #

将TestClient定义为函数级别的pytest fixture，让pytest自动为每个测试函数注入全新的客户端实例：

import pytest
from fastapi.testclient import TestClient
from fastapi import FastAPI
from sse_starlette import EventSourceResponse


app = FastAPI()


@app.post("/hello")
def query_hello_1():
    def hello():
        yield "hello"
    return EventSourceResponse(hello())


@app.post("/hello_2")
def query_hello_2():
    def hello():
        yield "hello_2"
    return EventSourceResponse(hello())


@pytest.fixture(scope="function")
def client():
    return TestClient(app)


def test_hello_1(client):
    print(list(client.post("/hello").iter_lines()))


def test_hello_2(client):
    print(list(client.post("/hello_2").iter_lines()))

问题根源 #

EventSourceResponse依赖异步事件循环，全局TestClient会在首次请求时绑定到某个事件循环；而pytest默认会为每个测试函数创建新的事件循环，后续测试中客户端的旧循环绑定与当前测试的循环不匹配，因此抛出错误。为每个测试创建新的TestClient实例，就能保证每次请求都使用当前测试的事件循环，避免冲突。

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