feat: added subprocess test client (#3655)

---------

Co-authored-by: Janek Nouvertné <[email protected]>

docs/examples/testing/subprocess_sse_app.py

@@ -0,0 +1,24 @@
+"""
+Assemble components into an app that shall be tested
+"""
+
+from typing import AsyncGenerator
+
+from litestar import Litestar, get
+from litestar.response import ServerSentEvent
+from litestar.types import SSEData
+
+
+async def generator(topic: str) -> AsyncGenerator[SSEData, None]:
+    count = 0
+    while count < 2:
+        yield topic
+        count += 1
+
+
+@get("/notify/{topic:str}")
+async def get_notified(topic: str) -> ServerSentEvent:
+    return ServerSentEvent(generator(topic), event_type="Notifier")
+
+
+app = Litestar(route_handlers=[get_notified])

docs/examples/testing/test_subprocess_sse.py

@@ -0,0 +1,45 @@
+"""
+Test the app running in a subprocess
+"""
+
+import asyncio
+import pathlib
+import sys
+from typing import AsyncIterator
+
+import httpx
+import httpx_sse
+import pytest
+
+from litestar.testing import subprocess_async_client
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+pytestmark = pytest.mark.anyio
+
+
+@pytest.fixture(scope="session")
+def anyio_backend() -> str:
+    return "asyncio"
+
+
+ROOT = pathlib.Path(__file__).parent
+
+
+@pytest.fixture(name="async_client", scope="session")
+async def fx_async_client() -> AsyncIterator[httpx.AsyncClient]:
+    async with subprocess_async_client(workdir=ROOT, app="subprocess_sse_app:app") as client:
+        yield client
+
+
+async def test_subprocess_async_client(async_client: httpx.AsyncClient) -> None:
+    """Demonstrates functionality of the async client with an infinite SSE source that cannot be tested with the
+    regular async test client.
+    """
+    topic = "demo"
+
+    async with httpx_sse.aconnect_sse(async_client, "GET", f"/notify/{topic}") as event_source:
+        async for event in event_source.aiter_sse():
+            assert event.data == topic
+            break

docs/reference/testing.rst

@@ -3,7 +3,7 @@ testing
 
 
 .. automodule:: litestar.testing
-    :members: RequestFactory, BaseTestClient, TestClient, AsyncTestClient, create_async_test_client, create_test_client
+    :members: RequestFactory, BaseTestClient, TestClient, AsyncTestClient, create_async_test_client, create_test_client, subprocess_sync_client, subprocess_async_client
     :undoc-members: WebSocketTestSession
 
 

docs/usage/testing.rst

@@ -287,6 +287,31 @@ But also this:
             assert response.text == "healthy"
 
 
+Running a live server
+---------------------
+
+The test clients make use of HTTPX's ability to directly call into an ASGI app, without
+having to run an actual server. In most cases this is sufficient but there are some
+exceptions where this won't work, due to the limitations of the emulated client-server
+communication.
+
+For example, when using server-sent events with an infinite generator, it will lock up
+the test client, since HTTPX tries to consume the full response before returning a
+request.
+
+Litestar offers two helper functions,
+:func:`litestar.testing.subprocess_sync_client` and
+:func:`litestar.testing.subprocess_async_client` that will
+launch a Litestar instance with in a subprocess and set up an httpx client for running
+tests. You can either load your actual app file or create subsets from it as you would
+with the regular test client setup:
+
+.. literalinclude:: /examples/testing/subprocess_sse_app.py
+    :language: python
+
+.. literalinclude:: /examples/testing/test_subprocess_sse.py
+    :language: python
+
 RequestFactory
 --------------
 

litestar/testing/__init__.py

@@ -1,5 +1,6 @@
 from litestar.testing.client.async_client import AsyncTestClient
 from litestar.testing.client.base import BaseTestClient
+from litestar.testing.client.subprocess_client import subprocess_async_client, subprocess_sync_client
 from litestar.testing.client.sync_client import TestClient
 from litestar.testing.helpers import create_async_test_client, create_test_client
 from litestar.testing.request_factory import RequestFactory
@@ -13,4 +14,6 @@
     "WebSocketTestSession",
     "create_async_test_client",
     "create_test_client",
+    "subprocess_async_client",
+    "subprocess_sync_client",
 )

litestar/testing/client/subprocess_client.py

@@ -0,0 +1,70 @@
+import pathlib
+import socket
+import subprocess
+import time
+from contextlib import asynccontextmanager, contextmanager
+from typing import AsyncIterator, Iterator
+
+import httpx
+
+
+class StartupError(RuntimeError):
+    pass
+
+
+def _get_available_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        # Bind to a free port provided by the host
+        try:
+            sock.bind(("localhost", 0))
+        except OSError as e:  # pragma: no cover
+            raise StartupError("Could not find an open port") from e
+        else:
+            port: int = sock.getsockname()[1]
+            return port
+
+
+@contextmanager
+def run_app(workdir: pathlib.Path, app: str) -> Iterator[str]:
+    """Launch a litestar application in a subprocess with a random available port."""
+    port = _get_available_port()
+    with subprocess.Popen(
+        args=["litestar", "--app", app, "run", "--port", str(port)],
+        stderr=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        cwd=workdir,
+    ) as proc:
+        url = f"http://127.0.0.1:{port}"
+        for _ in range(100):  # pragma: no cover
+            try:
+                httpx.get(url, timeout=0.1)
+                break
+            except httpx.TransportError:
+                time.sleep(1)
+        yield url
+        proc.kill()
+
+
+@asynccontextmanager
+async def subprocess_async_client(workdir: pathlib.Path, app: str) -> AsyncIterator[httpx.AsyncClient]:
+    """Provides an async httpx client for a litestar app launched in a subprocess.
+
+    Args:
+        workdir: Path to the directory in which the app module resides.
+        app: Uvicorn app string that can be resolved in the provided working directory, e.g.: "app:app"
+    """
+    with run_app(workdir=workdir, app=app) as url:
+        async with httpx.AsyncClient(base_url=url) as client:
+            yield client
+
+
+@contextmanager
+def subprocess_sync_client(workdir: pathlib.Path, app: str) -> Iterator[httpx.Client]:
+    """Provides a sync httpx client for a litestar app launched in a subprocess.
+
+    Args:
+        workdir: Path to the directory in which the app module resides.
+        app: Uvicorn app string that can be resolved in the provided working directory, e.g.: "app:app"
+    """
+    with run_app(workdir=workdir, app=app) as url, httpx.Client(base_url=url) as client:
+        yield client

tests/unit/test_testing/test_sub_client/demo.py

@@ -0,0 +1,28 @@
+"""
+Assemble components into an app that shall be tested
+"""
+
+import asyncio
+from typing import AsyncIterator
+
+from litestar import Litestar, get
+from litestar.response import ServerSentEvent
+
+
+@get("/notify/{topic:str}")
+async def get_notified(topic: str) -> ServerSentEvent:
+    async def generator() -> AsyncIterator[str]:
+        yield topic
+        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_testing/test_sub_client/test_subprocess_client.py

@@ -0,0 +1,54 @@
+"""
+Test the app running in a subprocess
+"""
+
+import asyncio
+import pathlib
+import sys
+from typing import AsyncIterator, Iterator
+
+import httpx
+import httpx_sse
+import pytest
+
+from litestar.testing import subprocess_async_client, subprocess_sync_client
+
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
+
+ROOT = pathlib.Path(__file__).parent
+
+
+@pytest.fixture(name="async_client", scope="session")
+async def fx_async_client() -> AsyncIterator[httpx.AsyncClient]:
+    async with subprocess_async_client(workdir=ROOT, app="demo:app") as client:
+        yield client
+
+
+@pytest.fixture(name="sync_client", scope="session")
+def fx_sync_client() -> Iterator[httpx.Client]:
+    with subprocess_sync_client(workdir=ROOT, app="demo:app") as client:
+        yield client
+
+
+async def test_subprocess_async_client(async_client: httpx.AsyncClient) -> None:
+    """Demonstrates functionality of the async client with an infinite SSE source that cannot be tested with the
+    regular async test client.
+    """
+
+    async with httpx_sse.aconnect_sse(async_client, "GET", "/notify/hello") as event_source:
+        async for event in event_source.aiter_sse():
+            assert event.data == "hello"
+            break
+
+
+def test_subprocess_sync_client(sync_client: httpx.Client) -> None:
+    """Demonstrates functionality of the async client with an infinite SSE source that cannot be tested with the
+    regular async test client.
+    """
+
+    with httpx_sse.connect_sse(sync_client, "GET", "/notify/hello") as event_source:
+        for event in event_source.iter_sse():
+            assert event.data == "hello"
+            break