Merge pull request #124 from QueryaHub/issue-102-testclient

feat(testing): ship oxyroute.testing.TestClient

Author: ZhuchkaTriplesix

docs/development.md

@@ -19,6 +19,29 @@ The `make test` and `make pytest` commands automatically run tests from a tempor
 - `make fix` — Auto-format code with `ruff format` and `cargo fmt`.
 - `make develop` — Build the Rust extension into `.venv` without running tests.
 
+## Writing Application Tests
+
+OxyRoute ships with an integrated `TestClient` for writing synchronous HTTP tests against your application without needing to start a real server.
+
+```python
+from oxyroute import App
+from oxyroute.testing import TestClient
+
+app = App()
+
+@app.get("/")
+def home():
+    return {"status": "ok"}
+
+def test_home():
+    with TestClient(app) as client:
+        resp = client.get("/")
+        assert resp.status_code == 200
+        assert resp.json() == {"status": "ok"}
+```
+
+Using `with TestClient(app)` ensures that the application's `__rsgi_init__` and `__rsgi_del__` lifespan hooks are run synchronously.
+
 ## Granian RSGI (end-to-end)
 
 `tests/test_granian_e2e.py` starts a real **Granian** subprocess with `--interface rsgi`, sends HTTP requests with **httpx**, then stops the server. It is part of the normal **pytest** run when `granian` is installed (`oxyroute[dev]` includes it). The same file runs in **CI** on every matrix combination (Linux, macOS, Windows), so the native RSGI path is exercised against a real server, not only the in-process httpx test transport.

tests/_rsgi_test_transport.py oxyroute/testing.pytests/_rsgi_test_transport.py renamed to oxyroute/testing.py

@@ -18,6 +18,8 @@
 from collections.abc import Callable
 from typing import Any
 
+import httpx
+
 _BLOCKING_LOOP_LOCAL = threading.local()
 
 
@@ -409,3 +411,62 @@ async def asgi3(scope: dict[str, Any], receive: Any, send: Any) -> None:
 
 asgi_test_app = build_test_app
 """Alias: ``asgi_test_app(app)`` returns an ASGI3 callable for httpx.ASGITransport."""
+
+
+class TestClient(httpx.Client):
+    """Synchronous test client for OxyRoute apps.
+
+    Wraps the RSGI testing transport and an async httpx client in a background
+    thread so it can be used in fully synchronous tests.
+    """
+
+    def __init__(self, app: Any, base_url: str = "http://testserver", **kwargs: Any) -> None:
+        self.app = app
+        self._loop = asyncio.new_event_loop()
+        self._thread = threading.Thread(target=self._run_loop, daemon=True)
+        self._thread.start()
+
+        transport = httpx.ASGITransport(app=asgi_test_app(app), client=("127.0.0.1", 12345))
+        self.async_client = httpx.AsyncClient(transport=transport, base_url=base_url, **kwargs)
+
+        super().__init__(
+            transport=httpx.MockTransport(lambda r: httpx.Response(200)),
+            base_url=base_url,
+            **kwargs,
+        )
+
+    def _run_loop(self) -> None:
+        asyncio.set_event_loop(self._loop)
+        self._loop.run_forever()
+
+    def _run_sync(self, coro: Any) -> Any:
+        return asyncio.run_coroutine_threadsafe(coro, self._loop).result()
+
+    def send(self, request: httpx.Request, **kwargs: Any) -> httpx.Response:
+        resp = self._run_sync(self.async_client.send(request, **kwargs))
+        self._run_sync(resp.aread())
+        return resp
+
+    def close(self) -> None:
+        self._run_sync(self.async_client.aclose())
+        if self._loop.is_running():
+            self._loop.call_soon_threadsafe(self._loop.stop)
+            self._thread.join()
+        super().close()
+
+    def __enter__(self) -> TestClient:
+        self._run_sync(self.async_client.__aenter__())
+        if hasattr(self.app, "__rsgi_init__"):
+            init = self.app.__rsgi_init__()
+            if asyncio.iscoroutine(init):
+                self._run_sync(init)
+        return self
+
+    def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+        self._run_sync(self.async_client.__aexit__(exc_type, exc_value, traceback))
+        if hasattr(self.app, "__rsgi_del__"):
+            dele = self.app.__rsgi_del__()
+            if asyncio.iscoroutine(dele):
+                self._run_sync(dele)
+        self.close()
+        super().__exit__(exc_type, exc_value, traceback)

tests/test_405.py

@@ -6,7 +6,7 @@
 
 import httpx
 from oxyroute import App
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 def test_405_get_on_post_only_path() -> None:

tests/test_api_router.py

@@ -9,7 +9,7 @@
 import pytest
 from oxyroute import APIRouter, App
 from oxyroute.router import join_path
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 def test_join_path() -> None:

tests/test_cors.py

@@ -6,7 +6,7 @@
 
 import httpx
 from oxyroute import App, CORSConfig, apply_cors
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 def test_cors_preflight_204_allows_post() -> None:

tests/test_cors_units.py

@@ -8,7 +8,7 @@
 import httpx
 from oxyroute import App
 from oxyroute.cors import CORSConfig, apply_cors
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 @dataclass

tests/test_csrf.py

@@ -8,7 +8,7 @@
 import httpx
 from oxyroute import App
 from oxyroute.csrf import CSRFConfig, apply_csrf
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 _HDR = "X-CSRF-Token"
 CK = "oxyroute_csrf"

tests/test_db_query.py

@@ -1,7 +1,7 @@
 import httpx
 import pytest
 from oxyroute import App, DBQuery, Depends
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 @pytest.mark.anyio

tests/test_dep_chain.py

@@ -6,7 +6,7 @@
 
 import httpx
 from oxyroute import App
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 def test_dep_second_receives_first_by_name() -> None:

tests/test_exception_handlers.py

@@ -2,7 +2,7 @@
 
 import httpx
 from oxyroute import App, Response
-from tests._rsgi_test_transport import asgi_test_app
+from oxyroute.testing import asgi_test_app
 
 
 def test_exception_handlers():