Hi I'm trying to test an SSE (Server-Sent Events) endpoint implemented with FastAPI using pytest.
In the following sample code, the endpoint works. But the test stops at client.get('/') maybe because TestClient does not SSE.
# test_main.py
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
@app.get('/')
async def main(request: Request) -> StreamingResponse:
async def gen():
while True:
yield 'data: ' + json.dumps({'msg': 'Hello World!'}) + '\n\n'
await asyncio.sleep(0.5)
return StreamingResponse(gen(), media_type='text/event-stream')
client = TestClient(app)
def test_read_stream():
response = client.get("/")
assert response.status_code == 200
for line in response.iter_lines():
assert line == 'data: {"msg": "Hello World"}'
How can I test the API using pytest?
