test/refactor: use subprocess client

Author: winstxnhdw

tests/unit/test_response/demo.py

@@ -0,0 +1,33 @@
+import asyncio
+from collections.abc import AsyncIterator
+
+import anyio
+
+from litestar import Litestar, get
+from litestar.response import ServerSentEvent
+from litestar.response.streaming import ClientDisconnectError
+
+
+@get("/notify/{topic:str}")
+async def get_notified(topic: str) -> ServerSentEvent:
+    file = await anyio.NamedTemporaryFile("w", suffix=topic, prefix="test").__aenter__()
+    await file.write(topic)
+
+    async def generator() -> AsyncIterator[str]:
+        try:
+            yield topic
+        except ClientDisconnectError:
+            await file.__aexit__(None, None, None)
+        while True:
+            await asyncio.sleep(0.1)
+
+    return ServerSentEvent(generator(), event_type="Notifier")
+
+
+def create_test_app() -> Litestar:
+    return Litestar(
+        route_handlers=[get_notified],
+    )
+
+
+app = create_test_app()

tests/unit/test_response/test_sse.py

@@ -1,6 +1,8 @@
+import pathlib
 from collections.abc import AsyncIterator, Iterator
 
 import anyio
+import httpx
 import pytest
 from httpx_sse import ServerSentEvent as HTTPXServerSentEvent
 from httpx_sse import aconnect_sse
@@ -9,9 +11,18 @@
 from litestar.exceptions import ImproperlyConfiguredException
 from litestar.response import ServerSentEvent
 from litestar.response.sse import ServerSentEventMessage
-from litestar.testing import create_async_test_client
+from litestar.testing import create_async_test_client, subprocess_async_client
 from litestar.types import SSEData
 
+ROOT = pathlib.Path(__file__).parent
+APP = "demo:app"
+
+
+@pytest.fixture(name="async_client")
+async def fx_async_client() -> AsyncIterator[httpx.AsyncClient]:
+    async with subprocess_async_client(workdir=ROOT, app=APP) as client:
+        yield client
+
 
 async def test_sse_steaming_response() -> None:
     @get(
@@ -96,28 +107,21 @@ async def numbers() -> AsyncIterator[SSEData]:
                 assert events[i].retry == expected_events[i].retry
 
 
-async def test_sse_cleanup() -> None:
-    shared_state = []
+async def test_sse_cleanup(async_client: httpx.AsyncClient) -> None:
+    topic = "cleanup"
 
-    @get("/testme")
-    async def handler() -> ServerSentEvent:
-        async def numbers() -> AsyncIterator[SSEData]:
-            try:
-                yield 0
-                await anyio.sleep(5)
-                yield 1
-            finally:
-                shared_state.append(42)
-
-        return ServerSentEvent(numbers(), event_type="special", event_id="123", retry_duration=1000)
-
-    async with create_async_test_client(handler) as client:
-        async with aconnect_sse(client, "GET", f"{client.base_url}/testme") as event_source:
-            async for sse in event_source.aiter_sse():
-                assert sse.data == "0"
-                break
+    async with (
+        anyio.NamedTemporaryFile("r", suffix=topic, prefix="test") as file,
+        async_client as client,
+        aconnect_sse(client, "GET", f"notify/{topic}") as event_source,
+    ):
+        file_path = pathlib.Path(str(file.name))
+        assert not file_path.exists()
+        async for sse in event_source.aiter_sse():
+            assert sse.data == topic
+            assert file_path.read_text() == topic
 
-    assert shared_state.pop() == 42
+    assert not file_path.exists()
 
 
 def test_invalid_content_type_raises() -> None: