# test_endless_sse.py
import pytest
import httpx
import asyncio

from fastapi import FastAPI
from sse_starlette.sse import EventSourceResponse

app = FastAPI()

@app.get("/endless_sse")
async def endless_sse():
    async def event_publisher():
        i = 0
        try:
            while True:
                i += 1
                yield dict(data=i)
                await asyncio.sleep(0.2)
        except asyncio.CancelledError as e:
            raise e

    return EventSourceResponse(event_publisher())

# This test should pass but it hangs forever and prints no output
@pytest.mark.asyncio
async def test_endless_sse():
    async with httpx.AsyncClient(app=app, base_url="http://test") as aclient:
        async with aclient.stream("GET", "/endless_sse") as response:
            async for line in response.aiter_lines():
                line_stripped = line.strip()
                print("line_stripped: ", line_stripped)
                if line_stripped == "data: 10":
                    break


if __name__ == "__main__":
    # Run: uvicorn test_endless_sse:app --reload

    # Check that we get output from the app running in uvicorn
    async def fetch_endless_sse():
        async with httpx.AsyncClient(base_url="http://127.0.0.1:8000") as aclient:
            async with aclient.stream("GET", "/endless_sse") as response:
                async for line in response.aiter_lines():
                    line_stripped = line.strip()
                    print("line_stripped: ", line_stripped)
                    if line_stripped == "data: 10":
                        break


    asyncio.run(fetch_endless_sse())

python 3.8.0

httpx==0.22.0
pytest==7.1.1
pytest-asyncio==0.18.3
fastapi==0.75.2
uvicorn==0.17.6
sse-starlette==0.10.3