From 666bb396825ab91b81f8cb700bbc2d51f04884ce Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sat, 4 Oct 2025 14:18:25 +0900 Subject: [PATCH 01/11] Add doc on testing Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 526 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 526 insertions(+) create mode 100644 docs/testing.md diff --git a/docs/testing.md b/docs/testing.md new file mode 100644 index 0000000..c7d093c --- /dev/null +++ b/docs/testing.md @@ -0,0 +1,526 @@ +# Testing Guide + +This guide covers testing connect-python services and clients. + +> **Note:** The examples in this guide use a fictional `GreetService` for demonstration purposes. In your actual project, replace these with your own service definitions. + +## Setup + +Install the required testing dependencies: + +```bash +pip install pytest pytest-asyncio httpx +``` + +Or if using uv: + +```bash +uv add --dev pytest pytest-asyncio httpx +``` + +## Recommended approach: In-memory testing + +The recommended approach is **in-memory testing** using httpx's ASGI/WSGI transports. This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts. + +## Testing servers + +### In-memory testing + +Test services using httpx's ASGI/WSGI transport, which tests your full application stack while remaining fast and isolated: + +=== "ASGI" + + ```python + import pytest + import httpx + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class TestGreetService(GreetService): + async def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest.mark.asyncio + async def test_greet(): + # Create the ASGI application + app = GreetServiceASGIApplication(TestGreetService()) + + # Test using httpx with ASGI transport + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + response = await client.greet(GreetRequest(name="Alice")) + + assert response.greeting == "Hello, Alice!" + ``` + +=== "WSGI" + + ```python + import httpx + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class TestGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + + def test_greet(): + # Create the WSGI application + app = GreetServiceWSGIApplication(TestGreetServiceSync()) + + # Test using httpx with WSGI transport + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + response = client.greet(GreetRequest(name="Alice")) + + assert response.greeting == "Hello, Alice!" + ``` + +This approach: + +- Tests your full application stack (routing, serialization, error handling) +- Runs fast without network overhead +- Provides isolation between tests +- Works with all streaming types + +For integration tests with actual servers over TCP/HTTP, see standard pytest patterns for [server fixtures](https://docs.pytest.org/en/stable/how-to/fixtures.html). + +### Using fixtures for reusable test setup + +For cleaner tests, use pytest fixtures to set up clients and services: + +=== "ASGI" + + ```python + import pytest + import pytest_asyncio + import httpx + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class TestGreetService(GreetService): + async def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest_asyncio.fixture + async def greet_client(): + app = GreetServiceASGIApplication(TestGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + + @pytest.mark.asyncio + async def test_greet(greet_client): + response = await greet_client.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + + @pytest.mark.asyncio + async def test_greet_multiple_names(greet_client): + response = await greet_client.greet(GreetRequest(name="Bob")) + assert response.greeting == "Hello, Bob!" + ``` + +=== "WSGI" + + ```python + import pytest + import httpx + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class TestGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest.fixture + def greet_client(): + app = GreetServiceWSGIApplication(TestGreetServiceSync()) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClientSync("http://test", session=session) + + def test_greet(greet_client): + response = greet_client.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + + def test_greet_multiple_names(greet_client): + response = greet_client.greet(GreetRequest(name="Bob")) + assert response.greeting == "Hello, Bob!" + ``` + +This pattern: + +- Reduces code duplication across multiple tests +- Makes tests more readable and focused on behavior +- Follows pytest best practices +- Matches the pattern used in connect-python's own test suite + +### Testing error handling + +Test that your service returns appropriate errors: + +```python +from connectrpc.code import Code +from connectrpc.errors import ConnectError + +class TestGreetService(GreetService): + async def greet(self, request, ctx): + if not request.name: + raise ConnectError(Code.INVALID_ARGUMENT, "name is required") + return GreetResponse(greeting=f"Hello, {request.name}!") + +@pytest.mark.asyncio +async def test_greet_error(): + app = GreetServiceASGIApplication(TestGreetService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + with pytest.raises(ConnectError) as exc_info: + await client.greet(GreetRequest(name="")) + + assert exc_info.value.code == Code.INVALID_ARGUMENT + assert "name is required" in exc_info.value.message +``` + +### Testing streaming services + +For server streaming: + +```python +@pytest.mark.asyncio +async def test_server_streaming(): + class StreamingGreetService(GreetService): + async def greet_stream(self, request, ctx): + for i in range(3): + yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") + + app = GreetServiceASGIApplication(StreamingGreetService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + responses = [] + async for response in client.greet_stream(GreetRequest(name="Alice")): + responses.append(response) + + assert len(responses) == 3 + assert responses[0].greeting == "Hello Alice #1!" +``` + +For client streaming: + +```python +@pytest.mark.asyncio +async def test_client_streaming(): + class ClientStreamingService(GreetService): + async def greet_many(self, request_stream, ctx): + names = [] + async for req in request_stream: + names.append(req.name) + return GreetResponse(greeting=f"Hello, {', '.join(names)}!") + + app = GreetServiceASGIApplication(ClientStreamingService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + async def request_stream(): + yield GreetRequest(name="Alice") + yield GreetRequest(name="Bob") + + response = await client.greet_many(request_stream()) + + assert "Alice" in response.greeting + assert "Bob" in response.greeting +``` + +### Testing with context (headers and trailers) + +Test code that uses request headers: + +```python +class AuthGreetService(GreetService): + async def greet(self, request, ctx): + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing token") + + ctx.response_headers()["greet-version"] = "v1" + return GreetResponse(greeting=f"Hello, {request.name}!") + +@pytest.mark.asyncio +async def test_greet_with_headers(): + app = GreetServiceASGIApplication(AuthGreetService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + response = await client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer token123"} + ) + + assert response.greeting == "Hello, Alice!" +``` + + +## Testing clients + +For testing client code that calls Connect services, use the same in-memory testing approach shown above. Create a test service implementation and use httpx transports to test your client logic without network overhead. + +## Testing interceptors + +### Testing with interceptors + +The recommended approach is to test interceptors as part of your full application stack: + +```python +class LoggingInterceptor: + def __init__(self): + self.requests = [] + + async def on_start(self, ctx): + method_name = ctx.method().name + self.requests.append(method_name) + return method_name + + async def on_end(self, token, ctx): + # token is the value returned from on_start + pass + +@pytest.mark.asyncio +async def test_service_with_interceptor(): + interceptor = LoggingInterceptor() + app = GreetServiceASGIApplication( + TestGreetService(), + interceptors=[interceptor] + ) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + await client.greet(GreetRequest(name="Alice")) + + # Verify interceptor was called + assert "Greet" in interceptor.requests +``` + +## Test organization + +### Project structure + +Organize your tests in a `test/` directory at the root of your project: + +``` +my-project/ +├── greet/ +│ └── v1/ +│ ├── greet_connect.py +│ └── greet_pb2.py +├── test/ +│ ├── __init__.py +│ ├── conftest.py # Shared fixtures +│ ├── test_greet.py # Service tests +│ └── test_integration.py # Integration tests +└── pyproject.toml +``` + +### Shared fixtures with conftest.py + +Use `conftest.py` to share fixtures across multiple test files: + +```python +# test/conftest.py +import pytest +import pytest_asyncio +import httpx +from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient +from greet.v1.greet_pb2 import GreetResponse + +class TestGreetService(GreetService): + async def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + +@pytest_asyncio.fixture +async def greet_client(): + """Shared client fixture available to all tests.""" + app = GreetServiceASGIApplication(TestGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) +``` + +Then use it in any test file: + +```python +# test/test_greet.py +import pytest +from greet.v1.greet_pb2 import GreetRequest + +@pytest.mark.asyncio +async def test_greet(greet_client): + """Test basic greeting.""" + response = await greet_client.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" +``` + +### Running tests + +Run all tests: + +```bash +pytest +``` + +Run tests in a specific file: + +```bash +pytest test/test_greet.py +``` + +Run a specific test: + +```bash +pytest test/test_greet.py::test_greet +``` + +Run with verbose output: + +```bash +pytest -v +``` + +## Practical examples + +### Testing with mock external dependencies + +Use fixtures to mock external services: + +```python +import pytest +import pytest_asyncio +from unittest.mock import AsyncMock +from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient +from greet.v1.greet_pb2 import GreetRequest, GreetResponse + +class DatabaseGreetService(GreetService): + def __init__(self, db): + self.db = db + + async def greet(self, request, ctx): + # Fetch greeting from database + greeting_template = await self.db.get_greeting_template() + return GreetResponse(greeting=greeting_template.format(name=request.name)) + +@pytest.fixture +def mock_db(): + """Mock database for testing.""" + db = AsyncMock() + db.get_greeting_template.return_value = "Hello, {name}!" + return db + +@pytest_asyncio.fixture +async def greet_client_with_db(mock_db): + app = GreetServiceASGIApplication(DatabaseGreetService(mock_db)) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + +@pytest.mark.asyncio +async def test_greet_with_database(greet_client_with_db, mock_db): + response = await greet_client_with_db.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + mock_db.get_greeting_template.assert_called_once() +``` + +### Testing authentication flows + +Test services that require authentication: + +```python +import pytest +import pytest_asyncio +from connectrpc.code import Code +from connectrpc.errors import ConnectError +from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient +from greet.v1.greet_pb2 import GreetRequest, GreetResponse + +class AuthGreetService(GreetService): + async def greet(self, request, ctx): + # Check for authorization header + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") + + # Validate token (simplified) + token = auth[7:] # Remove "Bearer " prefix + if token != "valid-token": + raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") + + return GreetResponse(greeting=f"Hello, {request.name}!") + +@pytest_asyncio.fixture +async def auth_greet_client(): + app = GreetServiceASGIApplication(AuthGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + +@pytest.mark.asyncio +async def test_greet_with_valid_token(auth_greet_client): + response = await auth_greet_client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer valid-token"} + ) + assert response.greeting == "Hello, Alice!" + +@pytest.mark.asyncio +async def test_greet_without_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + await auth_greet_client.greet(GreetRequest(name="Alice")) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Missing or invalid token" in exc_info.value.message + +@pytest.mark.asyncio +async def test_greet_with_invalid_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + await auth_greet_client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer invalid-token"} + ) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Invalid token" in exc_info.value.message +``` From a29ee46c3f310e42efa8cc5e96e3d6bc764f38e6 Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 5 Oct 2025 21:17:38 +0900 Subject: [PATCH 02/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 893 ++++++++++++++++++++++++++++++++++-------------- 1 file changed, 642 insertions(+), 251 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index c7d093c..4169c81 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -169,166 +169,419 @@ This pattern: Test that your service returns appropriate errors: -```python -from connectrpc.code import Code -from connectrpc.errors import ConnectError - -class TestGreetService(GreetService): - async def greet(self, request, ctx): - if not request.name: - raise ConnectError(Code.INVALID_ARGUMENT, "name is required") - return GreetResponse(greeting=f"Hello, {request.name}!") - -@pytest.mark.asyncio -async def test_greet_error(): - app = GreetServiceASGIApplication(TestGreetService()) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) +=== "ASGI" - with pytest.raises(ConnectError) as exc_info: - await client.greet(GreetRequest(name="")) + ```python + import pytest + import httpx + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse - assert exc_info.value.code == Code.INVALID_ARGUMENT - assert "name is required" in exc_info.value.message -``` + class TestGreetService(GreetService): + async def greet(self, request, ctx): + if not request.name: + raise ConnectError(Code.INVALID_ARGUMENT, "name is required") + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest.mark.asyncio + async def test_greet_error(): + app = GreetServiceASGIApplication(TestGreetService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + with pytest.raises(ConnectError) as exc_info: + await client.greet(GreetRequest(name="")) + + assert exc_info.value.code == Code.INVALID_ARGUMENT + assert "name is required" in exc_info.value.message + ``` + +=== "WSGI" + + ```python + import pytest + import httpx + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class TestGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + if not request.name: + raise ConnectError(Code.INVALID_ARGUMENT, "name is required") + return GreetResponse(greeting=f"Hello, {request.name}!") + + def test_greet_error(): + app = GreetServiceWSGIApplication(TestGreetServiceSync()) + + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + + with pytest.raises(ConnectError) as exc_info: + client.greet(GreetRequest(name="")) + + assert exc_info.value.code == Code.INVALID_ARGUMENT + assert "name is required" in exc_info.value.message + ``` ### Testing streaming services -For server streaming: +For server streaming (assumes your service has a `greet_stream` method defined in your proto): -```python -@pytest.mark.asyncio -async def test_server_streaming(): - class StreamingGreetService(GreetService): - async def greet_stream(self, request, ctx): - for i in range(3): - yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") +=== "ASGI" - app = GreetServiceASGIApplication(StreamingGreetService()) + ```python + @pytest.mark.asyncio + async def test_server_streaming(): + class StreamingGreetService(GreetService): + async def greet_stream(self, request, ctx): + for i in range(3): + yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) + app = GreetServiceASGIApplication(StreamingGreetService()) - responses = [] - async for response in client.greet_stream(GreetRequest(name="Alice")): - responses.append(response) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) - assert len(responses) == 3 - assert responses[0].greeting == "Hello Alice #1!" -``` + responses = [] + async for response in client.greet_stream(GreetRequest(name="Alice")): + responses.append(response) -For client streaming: + assert len(responses) == 3 + assert responses[0].greeting == "Hello Alice #1!" + ``` -```python -@pytest.mark.asyncio -async def test_client_streaming(): - class ClientStreamingService(GreetService): - async def greet_many(self, request_stream, ctx): - names = [] - async for req in request_stream: - names.append(req.name) - return GreetResponse(greeting=f"Hello, {', '.join(names)}!") +=== "WSGI" - app = GreetServiceASGIApplication(ClientStreamingService()) + ```python + def test_server_streaming(): + class StreamingGreetServiceSync(GreetServiceSync): + def greet_stream(self, request, ctx): + for i in range(3): + yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) + app = GreetServiceWSGIApplication(StreamingGreetServiceSync()) - async def request_stream(): - yield GreetRequest(name="Alice") - yield GreetRequest(name="Bob") + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) - response = await client.greet_many(request_stream()) + responses = [] + for response in client.greet_stream(GreetRequest(name="Alice")): + responses.append(response) - assert "Alice" in response.greeting - assert "Bob" in response.greeting -``` + assert len(responses) == 3 + assert responses[0].greeting == "Hello Alice #1!" + ``` + +For client streaming (assumes your service has a `greet_many` method defined in your proto): + +=== "ASGI" + + ```python + @pytest.mark.asyncio + async def test_client_streaming(): + class ClientStreamingService(GreetService): + async def greet_many(self, request_stream, ctx): + names = [] + async for req in request_stream: + names.append(req.name) + return GreetResponse(greeting=f"Hello, {', '.join(names)}!") + + app = GreetServiceASGIApplication(ClientStreamingService()) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + async def request_stream(): + yield GreetRequest(name="Alice") + yield GreetRequest(name="Bob") + + response = await client.greet_many(request_stream()) + + assert "Alice" in response.greeting + assert "Bob" in response.greeting + ``` + +=== "WSGI" + + ```python + def test_client_streaming(): + class ClientStreamingServiceSync(GreetServiceSync): + def greet_many(self, request_stream, ctx): + names = [] + for req in request_stream: + names.append(req.name) + return GreetResponse(greeting=f"Hello, {', '.join(names)}!") + + app = GreetServiceWSGIApplication(ClientStreamingServiceSync()) + + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + + def request_stream(): + yield GreetRequest(name="Alice") + yield GreetRequest(name="Bob") + + response = client.greet_many(request_stream()) + + assert "Alice" in response.greeting + assert "Bob" in response.greeting + ``` ### Testing with context (headers and trailers) Test code that uses request headers: -```python -class AuthGreetService(GreetService): - async def greet(self, request, ctx): - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing token") +=== "ASGI" - ctx.response_headers()["greet-version"] = "v1" - return GreetResponse(greeting=f"Hello, {request.name}!") + ```python + class AuthGreetService(GreetService): + async def greet(self, request, ctx): + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing token") -@pytest.mark.asyncio -async def test_greet_with_headers(): - app = GreetServiceASGIApplication(AuthGreetService()) + ctx.response_headers()["greet-version"] = "v1" + return GreetResponse(greeting=f"Hello, {request.name}!") - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) + @pytest.mark.asyncio + async def test_greet_with_headers(): + app = GreetServiceASGIApplication(AuthGreetService()) - response = await client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer token123"} - ) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) - assert response.greeting == "Hello, Alice!" -``` + response = await client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer token123"} + ) + + assert response.greeting == "Hello, Alice!" + ``` + +=== "WSGI" + + ```python + class AuthGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing token") + + ctx.response_headers()["greet-version"] = "v1" + return GreetResponse(greeting=f"Hello, {request.name}!") + + def test_greet_with_headers(): + app = GreetServiceWSGIApplication(AuthGreetServiceSync()) + + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + + response = client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer token123"} + ) + + assert response.greeting == "Hello, Alice!" + ``` ## Testing clients For testing client code that calls Connect services, use the same in-memory testing approach shown above. Create a test service implementation and use httpx transports to test your client logic without network overhead. +### Example: Testing client error handling + +=== "Async" + + ```python + import pytest + import httpx + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + async def fetch_user_greeting(user_id: str, client: GreetServiceClient): + """Client code that handles errors.""" + try: + response = await client.greet(GreetRequest(name=user_id)) + return response.greeting + except ConnectError as e: + if e.code == Code.NOT_FOUND: + return "User not found" + elif e.code == Code.UNAUTHENTICATED: + return "Please login" + raise + + @pytest.mark.asyncio + async def test_client_error_handling(): + class TestGreetService(GreetService): + async def greet(self, request, ctx): + if request.name == "unknown": + raise ConnectError(Code.NOT_FOUND, "User not found") + return GreetResponse(greeting=f"Hello, {request.name}!") + + app = GreetServiceASGIApplication(TestGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + + # Test successful case + result = await fetch_user_greeting("Alice", client) + assert result == "Hello, Alice!" + + # Test error handling + result = await fetch_user_greeting("unknown", client) + assert result == "User not found" + ``` + +=== "Sync" + + ```python + import httpx + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + def fetch_user_greeting(user_id: str, client: GreetServiceClientSync): + """Client code that handles errors.""" + try: + response = client.greet(GreetRequest(name=user_id)) + return response.greeting + except ConnectError as e: + if e.code == Code.NOT_FOUND: + return "User not found" + elif e.code == Code.UNAUTHENTICATED: + return "Please login" + raise + + def test_client_error_handling(): + class TestGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + if request.name == "unknown": + raise ConnectError(Code.NOT_FOUND, "User not found") + return GreetResponse(greeting=f"Hello, {request.name}!") + + app = GreetServiceWSGIApplication(TestGreetServiceSync()) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + + # Test successful case + result = fetch_user_greeting("Alice", client) + assert result == "Hello, Alice!" + + # Test error handling + result = fetch_user_greeting("unknown", client) + assert result == "User not found" + ``` + ## Testing interceptors -### Testing with interceptors - -The recommended approach is to test interceptors as part of your full application stack: - -```python -class LoggingInterceptor: - def __init__(self): - self.requests = [] - - async def on_start(self, ctx): - method_name = ctx.method().name - self.requests.append(method_name) - return method_name - - async def on_end(self, token, ctx): - # token is the value returned from on_start - pass - -@pytest.mark.asyncio -async def test_service_with_interceptor(): - interceptor = LoggingInterceptor() - app = GreetServiceASGIApplication( - TestGreetService(), - interceptors=[interceptor] - ) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) - await client.greet(GreetRequest(name="Alice")) - - # Verify interceptor was called - assert "Greet" in interceptor.requests -``` +Test interceptors as part of your full application stack: + +=== "ASGI" + + ```python + class LoggingInterceptor: + def __init__(self): + self.requests = [] + + async def on_start(self, ctx): + method_name = ctx.method().name + self.requests.append(method_name) + return method_name + + async def on_end(self, token, ctx): + # token is the value returned from on_start + pass + + @pytest.mark.asyncio + async def test_service_with_interceptor(): + interceptor = LoggingInterceptor() + app = GreetServiceASGIApplication( + TestGreetService(), + interceptors=[interceptor] + ) + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + await client.greet(GreetRequest(name="Alice")) + + # Verify interceptor was called + assert "Greet" in interceptor.requests + ``` + +=== "WSGI" + + ```python + class LoggingInterceptorSync: + def __init__(self): + self.requests = [] + + def on_start(self, ctx): + method_name = ctx.method().name + self.requests.append(method_name) + return method_name + + def on_end(self, token, ctx): + # token is the value returned from on_start + pass + + def test_service_with_interceptor(): + interceptor = LoggingInterceptorSync() + app = GreetServiceWSGIApplication( + TestGreetServiceSync(), + interceptors=[interceptor] + ) + + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + client.greet(GreetRequest(name="Alice")) + + # Verify interceptor was called + assert "Greet" in interceptor.requests + ``` ## Test organization @@ -354,42 +607,80 @@ my-project/ Use `conftest.py` to share fixtures across multiple test files: -```python -# test/conftest.py -import pytest -import pytest_asyncio -import httpx -from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient -from greet.v1.greet_pb2 import GreetResponse - -class TestGreetService(GreetService): - async def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") - -@pytest_asyncio.fixture -async def greet_client(): - """Shared client fixture available to all tests.""" - app = GreetServiceASGIApplication(TestGreetService()) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) -``` +=== "ASGI" + + ```python + # test/conftest.py + import pytest + import pytest_asyncio + import httpx + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetResponse -Then use it in any test file: + class TestGreetService(GreetService): + async def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") -```python -# test/test_greet.py -import pytest -from greet.v1.greet_pb2 import GreetRequest + @pytest_asyncio.fixture + async def greet_client(): + """Shared client fixture available to all tests.""" + app = GreetServiceASGIApplication(TestGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + ``` -@pytest.mark.asyncio -async def test_greet(greet_client): - """Test basic greeting.""" - response = await greet_client.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" -``` + Then use it in any test file: + + ```python + # test/test_greet.py + import pytest + from greet.v1.greet_pb2 import GreetRequest + + @pytest.mark.asyncio + async def test_greet(greet_client): + """Test basic greeting.""" + response = await greet_client.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + ``` + +=== "WSGI" + + ```python + # test/conftest.py + import pytest + import httpx + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetResponse + + class TestGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest.fixture + def greet_client(): + """Shared client fixture available to all tests.""" + app = GreetServiceWSGIApplication(TestGreetServiceSync()) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClientSync("http://test", session=session) + ``` + + Then use it in any test file: + + ```python + # test/test_greet.py + from greet.v1.greet_pb2 import GreetRequest + + def test_greet(greet_client): + """Test basic greeting.""" + response = greet_client.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + ``` ### Running tests @@ -423,104 +714,204 @@ pytest -v Use fixtures to mock external services: -```python -import pytest -import pytest_asyncio -from unittest.mock import AsyncMock -from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient -from greet.v1.greet_pb2 import GreetRequest, GreetResponse - -class DatabaseGreetService(GreetService): - def __init__(self, db): - self.db = db - - async def greet(self, request, ctx): - # Fetch greeting from database - greeting_template = await self.db.get_greeting_template() - return GreetResponse(greeting=greeting_template.format(name=request.name)) - -@pytest.fixture -def mock_db(): - """Mock database for testing.""" - db = AsyncMock() - db.get_greeting_template.return_value = "Hello, {name}!" - return db - -@pytest_asyncio.fixture -async def greet_client_with_db(mock_db): - app = GreetServiceASGIApplication(DatabaseGreetService(mock_db)) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) - -@pytest.mark.asyncio -async def test_greet_with_database(greet_client_with_db, mock_db): - response = await greet_client_with_db.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" - mock_db.get_greeting_template.assert_called_once() -``` +=== "ASGI" + + ```python + import pytest + import pytest_asyncio + from unittest.mock import AsyncMock + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class DatabaseGreetService(GreetService): + def __init__(self, db): + self.db = db + + async def greet(self, request, ctx): + # Fetch greeting from database + greeting_template = await self.db.get_greeting_template() + return GreetResponse(greeting=greeting_template.format(name=request.name)) + + @pytest.fixture + def mock_db(): + """Mock database for testing.""" + db = AsyncMock() + db.get_greeting_template.return_value = "Hello, {name}!" + return db + + @pytest_asyncio.fixture + async def greet_client_with_db(mock_db): + app = GreetServiceASGIApplication(DatabaseGreetService(mock_db)) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + + @pytest.mark.asyncio + async def test_greet_with_database(greet_client_with_db, mock_db): + response = await greet_client_with_db.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + mock_db.get_greeting_template.assert_called_once() + ``` + +=== "WSGI" + + ```python + import pytest + from unittest.mock import Mock + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class DatabaseGreetServiceSync(GreetServiceSync): + def __init__(self, db): + self.db = db + + def greet(self, request, ctx): + # Fetch greeting from database + greeting_template = self.db.get_greeting_template() + return GreetResponse(greeting=greeting_template.format(name=request.name)) + + @pytest.fixture + def mock_db(): + """Mock database for testing.""" + db = Mock() + db.get_greeting_template.return_value = "Hello, {name}!" + return db + + @pytest.fixture + def greet_client_with_db(mock_db): + app = GreetServiceWSGIApplication(DatabaseGreetServiceSync(mock_db)) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClientSync("http://test", session=session) + + def test_greet_with_database(greet_client_with_db, mock_db): + response = greet_client_with_db.greet(GreetRequest(name="Alice")) + assert response.greeting == "Hello, Alice!" + mock_db.get_greeting_template.assert_called_once() + ``` ### Testing authentication flows Test services that require authentication: -```python -import pytest -import pytest_asyncio -from connectrpc.code import Code -from connectrpc.errors import ConnectError -from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient -from greet.v1.greet_pb2 import GreetRequest, GreetResponse - -class AuthGreetService(GreetService): - async def greet(self, request, ctx): - # Check for authorization header - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") - - # Validate token (simplified) - token = auth[7:] # Remove "Bearer " prefix - if token != "valid-token": - raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") - - return GreetResponse(greeting=f"Hello, {request.name}!") - -@pytest_asyncio.fixture -async def auth_greet_client(): - app = GreetServiceASGIApplication(AuthGreetService()) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) - -@pytest.mark.asyncio -async def test_greet_with_valid_token(auth_greet_client): - response = await auth_greet_client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer valid-token"} - ) - assert response.greeting == "Hello, Alice!" - -@pytest.mark.asyncio -async def test_greet_without_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - await auth_greet_client.greet(GreetRequest(name="Alice")) - - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Missing or invalid token" in exc_info.value.message - -@pytest.mark.asyncio -async def test_greet_with_invalid_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - await auth_greet_client.greet( +=== "ASGI" + + ```python + import pytest + import pytest_asyncio + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class AuthGreetService(GreetService): + async def greet(self, request, ctx): + # Check for authorization header + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") + + # Validate token (simplified) + token = auth[7:] # Remove "Bearer " prefix + if token != "valid-token": + raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") + + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest_asyncio.fixture + async def auth_greet_client(): + app = GreetServiceASGIApplication(AuthGreetService()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClient("http://test", session=session) + + @pytest.mark.asyncio + async def test_greet_with_valid_token(auth_greet_client): + response = await auth_greet_client.greet( GreetRequest(name="Alice"), - headers={"authorization": "Bearer invalid-token"} + headers={"authorization": "Bearer valid-token"} ) + assert response.greeting == "Hello, Alice!" - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Invalid token" in exc_info.value.message -``` + @pytest.mark.asyncio + async def test_greet_without_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + await auth_greet_client.greet(GreetRequest(name="Alice")) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Missing or invalid token" in exc_info.value.message + + @pytest.mark.asyncio + async def test_greet_with_invalid_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + await auth_greet_client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer invalid-token"} + ) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Invalid token" in exc_info.value.message + ``` + +=== "WSGI" + + ```python + import pytest + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest, GreetResponse + + class AuthGreetServiceSync(GreetServiceSync): + def greet(self, request, ctx): + # Check for authorization header + auth = ctx.request_headers().get("authorization") + if not auth or not auth.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") + + # Validate token (simplified) + token = auth[7:] # Remove "Bearer " prefix + if token != "valid-token": + raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") + + return GreetResponse(greeting=f"Hello, {request.name}!") + + @pytest.fixture + def auth_greet_client(): + app = GreetServiceWSGIApplication(AuthGreetServiceSync()) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + yield GreetServiceClientSync("http://test", session=session) + + def test_greet_with_valid_token(auth_greet_client): + response = auth_greet_client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer valid-token"} + ) + assert response.greeting == "Hello, Alice!" + + def test_greet_without_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + auth_greet_client.greet(GreetRequest(name="Alice")) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Missing or invalid token" in exc_info.value.message + + def test_greet_with_invalid_token(auth_greet_client): + with pytest.raises(ConnectError) as exc_info: + auth_greet_client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer invalid-token"} + ) + + assert exc_info.value.code == Code.UNAUTHENTICATED + assert "Invalid token" in exc_info.value.message + ``` From 0e41b3a5fd43e43db08ddde202c80864885a659f Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Mon, 13 Oct 2025 17:05:54 +0900 Subject: [PATCH 03/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/docs/testing.md b/docs/testing.md index 4169c81..043cbba 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -20,7 +20,7 @@ uv add --dev pytest pytest-asyncio httpx ## Recommended approach: In-memory testing -The recommended approach is **in-memory testing** using httpx's ASGI/WSGI transports. This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts. +The recommended approach is **in-memory testing** using httpx's ASGI/WSGI transports (provided by httpx, not connect-python). This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts. ## Testing servers @@ -413,6 +413,8 @@ Test code that uses request headers: assert response.greeting == "Hello, Alice!" ``` +> **Note:** For accessing response headers and trailers from clients, see [Headers and trailers](headers-and-trailers.md). + ## Testing clients @@ -719,6 +721,7 @@ Use fixtures to mock external services: ```python import pytest import pytest_asyncio + import httpx from unittest.mock import AsyncMock from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient from greet.v1.greet_pb2 import GreetRequest, GreetResponse @@ -759,6 +762,7 @@ Use fixtures to mock external services: ```python import pytest + import httpx from unittest.mock import Mock from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync from greet.v1.greet_pb2 import GreetRequest, GreetResponse @@ -803,6 +807,7 @@ Test services that require authentication: ```python import pytest import pytest_asyncio + import httpx from connectrpc.code import Code from connectrpc.errors import ConnectError from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient @@ -863,6 +868,7 @@ Test services that require authentication: ```python import pytest + import httpx from connectrpc.code import Code from connectrpc.errors import ConnectError from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync From 3cdd3c4a8d9413c6554b2312dae6929a7a41a4ed Mon Sep 17 00:00:00 2001 From: Yasushi Itoh Date: Sun, 19 Oct 2025 13:30:41 +0900 Subject: [PATCH 04/11] Update docs/testing.md Co-authored-by: Anuraag (Rag) Agrawal Signed-off-by: Yasushi Itoh --- docs/testing.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/testing.md b/docs/testing.md index 043cbba..8697e8c 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -31,8 +31,8 @@ Test services using httpx's ASGI/WSGI transport, which tests your full applicati === "ASGI" ```python - import pytest import httpx + import pytest from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient from greet.v1.greet_pb2 import GreetRequest, GreetResponse From b978c40fcdafebd0d87acff71df31938f3881d06 Mon Sep 17 00:00:00 2001 From: Yasushi Itoh Date: Sun, 19 Oct 2025 13:30:55 +0900 Subject: [PATCH 05/11] Update docs/testing.md Co-authored-by: Anuraag (Rag) Agrawal Signed-off-by: Yasushi Itoh --- docs/testing.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/testing.md b/docs/testing.md index 8697e8c..b41abad 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -1,4 +1,4 @@ -# Testing Guide +# Testing This guide covers testing connect-python services and clients. From 3b47abf6f2cda96b4ff2a969a123407937f09065 Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 19 Oct 2025 13:57:37 +0900 Subject: [PATCH 06/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 706 +++++++++++------------------------------------- 1 file changed, 159 insertions(+), 547 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index b41abad..eaaf4c9 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -22,28 +22,73 @@ uv add --dev pytest pytest-asyncio httpx The recommended approach is **in-memory testing** using httpx's ASGI/WSGI transports (provided by httpx, not connect-python). This tests your full application stack (routing, serialization, error handling, interceptors) while remaining fast and isolated - no network overhead or port conflicts. +Here's a minimal example without any test framework: + +=== "ASGI" + + ```python + import httpx + from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest + from server import Greeter # Your service implementation + + # Create ASGI app with your service + app = GreetServiceASGIApplication(Greeter()) + + # Connect client to service using in-memory transport + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" # URL is ignored for in-memory transport + ) as session: + client = GreetServiceClient("http://test", session=session) + response = await client.greet(GreetRequest(name="Alice")) + + print(response.greeting) # "Hello, Alice!" + ``` + +=== "WSGI" + + ```python + import httpx + from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest + from server import GreeterSync # Your service implementation + + # Create WSGI app with your service + app = GreetServiceWSGIApplication(GreeterSync()) + + # Connect client to service using in-memory transport + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" # URL is ignored for in-memory transport + ) as session: + client = GreetServiceClientSync("http://test", session=session) + response = client.greet(GreetRequest(name="Alice")) + + print(response.greeting) # "Hello, Alice!" + ``` + +This pattern works with any test framework (pytest, unittest) or none at all. The examples below show how to integrate with pytest, but the core testing approach remains the same. + ## Testing servers ### In-memory testing -Test services using httpx's ASGI/WSGI transport, which tests your full application stack while remaining fast and isolated: +Testing the service we created in the [Getting Started](getting-started.md) guide looks like this: === "ASGI" ```python import httpx import pytest - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetService(GreetService): - async def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") + from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest + from server import Greeter # Import your actual service implementation @pytest.mark.asyncio async def test_greet(): - # Create the ASGI application - app = GreetServiceASGIApplication(TestGreetService()) + # Create the ASGI application with your service + app = GreetServiceASGIApplication(Greeter()) # Test using httpx with ASGI transport async with httpx.AsyncClient( @@ -60,16 +105,13 @@ Test services using httpx's ASGI/WSGI transport, which tests your full applicati ```python import httpx - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") + from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest + from server import GreeterSync # Import your actual service implementation def test_greet(): - # Create the WSGI application - app = GreetServiceWSGIApplication(TestGreetServiceSync()) + # Create the WSGI application with your service + app = GreetServiceWSGIApplication(GreeterSync()) # Test using httpx with WSGI transport with httpx.Client( @@ -98,19 +140,16 @@ For cleaner tests, use pytest fixtures to set up clients and services: === "ASGI" ```python + import httpx import pytest import pytest_asyncio - import httpx - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetService(GreetService): - async def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") + from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest + from server import Greeter @pytest_asyncio.fixture async def greet_client(): - app = GreetServiceASGIApplication(TestGreetService()) + app = GreetServiceASGIApplication(Greeter()) async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" @@ -123,26 +162,23 @@ For cleaner tests, use pytest fixtures to set up clients and services: assert response.greeting == "Hello, Alice!" @pytest.mark.asyncio - async def test_greet_multiple_names(greet_client): - response = await greet_client.greet(GreetRequest(name="Bob")) - assert response.greeting == "Hello, Bob!" + async def test_greet_empty_name(greet_client): + response = await greet_client.greet(GreetRequest(name="")) + assert response.greeting == "Hello, !" ``` === "WSGI" ```python - import pytest import httpx - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") + import pytest + from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest + from server import GreeterSync @pytest.fixture def greet_client(): - app = GreetServiceWSGIApplication(TestGreetServiceSync()) + app = GreetServiceWSGIApplication(GreeterSync()) with httpx.Client( transport=httpx.WSGITransport(app=app), base_url="http://test" @@ -153,9 +189,9 @@ For cleaner tests, use pytest fixtures to set up clients and services: response = greet_client.greet(GreetRequest(name="Alice")) assert response.greeting == "Hello, Alice!" - def test_greet_multiple_names(greet_client): - response = greet_client.greet(GreetRequest(name="Bob")) - assert response.greeting == "Hello, Bob!" + def test_greet_empty_name(greet_client): + response = greet_client.greet(GreetRequest(name="")) + assert response.greeting == "Hello, !" ``` This pattern: @@ -165,255 +201,16 @@ This pattern: - Follows pytest best practices - Matches the pattern used in connect-python's own test suite -### Testing error handling - -Test that your service returns appropriate errors: - -=== "ASGI" - - ```python - import pytest - import httpx - from connectrpc.code import Code - from connectrpc.errors import ConnectError - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetService(GreetService): - async def greet(self, request, ctx): - if not request.name: - raise ConnectError(Code.INVALID_ARGUMENT, "name is required") - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest.mark.asyncio - async def test_greet_error(): - app = GreetServiceASGIApplication(TestGreetService()) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) - - with pytest.raises(ConnectError) as exc_info: - await client.greet(GreetRequest(name="")) - - assert exc_info.value.code == Code.INVALID_ARGUMENT - assert "name is required" in exc_info.value.message - ``` - -=== "WSGI" - - ```python - import pytest - import httpx - from connectrpc.code import Code - from connectrpc.errors import ConnectError - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class TestGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - if not request.name: - raise ConnectError(Code.INVALID_ARGUMENT, "name is required") - return GreetResponse(greeting=f"Hello, {request.name}!") - - def test_greet_error(): - app = GreetServiceWSGIApplication(TestGreetServiceSync()) - - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClientSync("http://test", session=session) - - with pytest.raises(ConnectError) as exc_info: - client.greet(GreetRequest(name="")) - - assert exc_info.value.code == Code.INVALID_ARGUMENT - assert "name is required" in exc_info.value.message - ``` - -### Testing streaming services - -For server streaming (assumes your service has a `greet_stream` method defined in your proto): - -=== "ASGI" - - ```python - @pytest.mark.asyncio - async def test_server_streaming(): - class StreamingGreetService(GreetService): - async def greet_stream(self, request, ctx): - for i in range(3): - yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") - - app = GreetServiceASGIApplication(StreamingGreetService()) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) - - responses = [] - async for response in client.greet_stream(GreetRequest(name="Alice")): - responses.append(response) - - assert len(responses) == 3 - assert responses[0].greeting == "Hello Alice #1!" - ``` - -=== "WSGI" - - ```python - def test_server_streaming(): - class StreamingGreetServiceSync(GreetServiceSync): - def greet_stream(self, request, ctx): - for i in range(3): - yield GreetResponse(greeting=f"Hello {request.name} #{i + 1}!") - - app = GreetServiceWSGIApplication(StreamingGreetServiceSync()) - - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClientSync("http://test", session=session) - - responses = [] - for response in client.greet_stream(GreetRequest(name="Alice")): - responses.append(response) - - assert len(responses) == 3 - assert responses[0].greeting == "Hello Alice #1!" - ``` - -For client streaming (assumes your service has a `greet_many` method defined in your proto): - -=== "ASGI" - - ```python - @pytest.mark.asyncio - async def test_client_streaming(): - class ClientStreamingService(GreetService): - async def greet_many(self, request_stream, ctx): - names = [] - async for req in request_stream: - names.append(req.name) - return GreetResponse(greeting=f"Hello, {', '.join(names)}!") - - app = GreetServiceASGIApplication(ClientStreamingService()) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) - - async def request_stream(): - yield GreetRequest(name="Alice") - yield GreetRequest(name="Bob") - - response = await client.greet_many(request_stream()) - - assert "Alice" in response.greeting - assert "Bob" in response.greeting - ``` - -=== "WSGI" - - ```python - def test_client_streaming(): - class ClientStreamingServiceSync(GreetServiceSync): - def greet_many(self, request_stream, ctx): - names = [] - for req in request_stream: - names.append(req.name) - return GreetResponse(greeting=f"Hello, {', '.join(names)}!") - - app = GreetServiceWSGIApplication(ClientStreamingServiceSync()) - - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClientSync("http://test", session=session) - - def request_stream(): - yield GreetRequest(name="Alice") - yield GreetRequest(name="Bob") - - response = client.greet_many(request_stream()) - - assert "Alice" in response.greeting - assert "Bob" in response.greeting - ``` - -### Testing with context (headers and trailers) - -Test code that uses request headers: - -=== "ASGI" - - ```python - class AuthGreetService(GreetService): - async def greet(self, request, ctx): - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing token") - - ctx.response_headers()["greet-version"] = "v1" - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest.mark.asyncio - async def test_greet_with_headers(): - app = GreetServiceASGIApplication(AuthGreetService()) - - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClient("http://test", session=session) - - response = await client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer token123"} - ) - - assert response.greeting == "Hello, Alice!" - ``` - -=== "WSGI" - - ```python - class AuthGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing token") - - ctx.response_headers()["greet-version"] = "v1" - return GreetResponse(greeting=f"Hello, {request.name}!") - - def test_greet_with_headers(): - app = GreetServiceWSGIApplication(AuthGreetServiceSync()) - - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - client = GreetServiceClientSync("http://test", session=session) +With your test client setup, you can use any connect code for interacting with the service under test including [streaming](streaming.md), reading [headers and trailers](headers-and-trailers.md), or checking [errors](errors.md). For example, to test error handling: - response = client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer token123"} - ) +```python +with pytest.raises(ConnectError) as exc_info: + await client.greet(GreetRequest(name="")) - assert response.greeting == "Hello, Alice!" - ``` +assert exc_info.value.code == Code.INVALID_ARGUMENT +``` -> **Note:** For accessing response headers and trailers from clients, see [Headers and trailers](headers-and-trailers.md). +See the [Errors](errors.md) guide for more details on error handling. ## Testing clients @@ -514,29 +311,38 @@ For testing client code that calls Connect services, use the same in-memory test ## Testing interceptors -Test interceptors as part of your full application stack: +Test interceptors as part of your full application stack. For example, testing an authentication interceptor: === "ASGI" ```python - class LoggingInterceptor: - def __init__(self): - self.requests = [] + import httpx + import pytest + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest + from server import Greeter + + class AuthInterceptor: + def __init__(self, valid_token: str): + self.valid_token = valid_token + self.authenticated_calls = 0 async def on_start(self, ctx): - method_name = ctx.method().name - self.requests.append(method_name) - return method_name + auth = ctx.request_headers().get("authorization") + if not auth or auth != f"Bearer {self.valid_token}": + raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") + self.authenticated_calls += 1 async def on_end(self, token, ctx): - # token is the value returned from on_start pass @pytest.mark.asyncio - async def test_service_with_interceptor(): - interceptor = LoggingInterceptor() + async def test_auth_interceptor(): + interceptor = AuthInterceptor("secret-token") app = GreetServiceASGIApplication( - TestGreetService(), + Greeter(), interceptors=[interceptor] ) @@ -545,32 +351,53 @@ Test interceptors as part of your full application stack: base_url="http://test" ) as session: client = GreetServiceClient("http://test", session=session) - await client.greet(GreetRequest(name="Alice")) - # Verify interceptor was called - assert "Greet" in interceptor.requests + # Valid token succeeds + response = await client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer secret-token"} + ) + assert response.greeting == "Hello, Alice!" + assert interceptor.authenticated_calls == 1 + + # Invalid token fails + with pytest.raises(ConnectError) as exc_info: + await client.greet( + GreetRequest(name="Bob"), + headers={"authorization": "Bearer wrong-token"} + ) + assert exc_info.value.code == Code.UNAUTHENTICATED ``` === "WSGI" ```python - class LoggingInterceptorSync: - def __init__(self): - self.requests = [] + import httpx + import pytest + from connectrpc.code import Code + from connectrpc.errors import ConnectError + from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest + from server import GreeterSync + + class AuthInterceptorSync: + def __init__(self, valid_token: str): + self.valid_token = valid_token + self.authenticated_calls = 0 def on_start(self, ctx): - method_name = ctx.method().name - self.requests.append(method_name) - return method_name + auth = ctx.request_headers().get("authorization") + if not auth or auth != f"Bearer {self.valid_token}": + raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") + self.authenticated_calls += 1 def on_end(self, token, ctx): - # token is the value returned from on_start pass - def test_service_with_interceptor(): - interceptor = LoggingInterceptorSync() + def test_auth_interceptor(): + interceptor = AuthInterceptorSync("secret-token") app = GreetServiceWSGIApplication( - TestGreetServiceSync(), + GreeterSync(), interceptors=[interceptor] ) @@ -579,17 +406,29 @@ Test interceptors as part of your full application stack: base_url="http://test" ) as session: client = GreetServiceClientSync("http://test", session=session) - client.greet(GreetRequest(name="Alice")) - # Verify interceptor was called - assert "Greet" in interceptor.requests + # Valid token succeeds + response = client.greet( + GreetRequest(name="Alice"), + headers={"authorization": "Bearer secret-token"} + ) + assert response.greeting == "Hello, Alice!" + assert interceptor.authenticated_calls == 1 + + # Invalid token fails + with pytest.raises(ConnectError) as exc_info: + client.greet( + GreetRequest(name="Bob"), + headers={"authorization": "Bearer wrong-token"} + ) + assert exc_info.value.code == Code.UNAUTHENTICATED ``` -## Test organization +See the [Interceptors](interceptors.md) guide for more details on implementing interceptors. -### Project structure +## Test organization -Organize your tests in a `test/` directory at the root of your project: +Organize your tests in a `tests/` directory at the root of your project: ``` my-project/ @@ -597,118 +436,15 @@ my-project/ │ └── v1/ │ ├── greet_connect.py │ └── greet_pb2.py -├── test/ +├── server.py +├── tests/ │ ├── __init__.py -│ ├── conftest.py # Shared fixtures -│ ├── test_greet.py # Service tests -│ └── test_integration.py # Integration tests +│ ├── test_greet.py +│ └── test_integration.py └── pyproject.toml ``` -### Shared fixtures with conftest.py - -Use `conftest.py` to share fixtures across multiple test files: - -=== "ASGI" - - ```python - # test/conftest.py - import pytest - import pytest_asyncio - import httpx - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetResponse - - class TestGreetService(GreetService): - async def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest_asyncio.fixture - async def greet_client(): - """Shared client fixture available to all tests.""" - app = GreetServiceASGIApplication(TestGreetService()) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) - ``` - - Then use it in any test file: - - ```python - # test/test_greet.py - import pytest - from greet.v1.greet_pb2 import GreetRequest - - @pytest.mark.asyncio - async def test_greet(greet_client): - """Test basic greeting.""" - response = await greet_client.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" - ``` - -=== "WSGI" - - ```python - # test/conftest.py - import pytest - import httpx - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetResponse - - class TestGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest.fixture - def greet_client(): - """Shared client fixture available to all tests.""" - app = GreetServiceWSGIApplication(TestGreetServiceSync()) - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClientSync("http://test", session=session) - ``` - - Then use it in any test file: - - ```python - # test/test_greet.py - from greet.v1.greet_pb2 import GreetRequest - - def test_greet(greet_client): - """Test basic greeting.""" - response = greet_client.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" - ``` - -### Running tests - -Run all tests: - -```bash -pytest -``` - -Run tests in a specific file: - -```bash -pytest test/test_greet.py -``` - -Run a specific test: - -```bash -pytest test/test_greet.py::test_greet -``` - -Run with verbose output: - -```bash -pytest -v -``` +For test framework-specific patterns like shared fixtures or test discovery, consult your test framework's documentation ([pytest](https://docs.pytest.org/), [unittest](https://docs.python.org/3/library/unittest.html)). ## Practical examples @@ -797,127 +533,3 @@ Use fixtures to mock external services: assert response.greeting == "Hello, Alice!" mock_db.get_greeting_template.assert_called_once() ``` - -### Testing authentication flows - -Test services that require authentication: - -=== "ASGI" - - ```python - import pytest - import pytest_asyncio - import httpx - from connectrpc.code import Code - from connectrpc.errors import ConnectError - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class AuthGreetService(GreetService): - async def greet(self, request, ctx): - # Check for authorization header - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") - - # Validate token (simplified) - token = auth[7:] # Remove "Bearer " prefix - if token != "valid-token": - raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") - - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest_asyncio.fixture - async def auth_greet_client(): - app = GreetServiceASGIApplication(AuthGreetService()) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) - - @pytest.mark.asyncio - async def test_greet_with_valid_token(auth_greet_client): - response = await auth_greet_client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer valid-token"} - ) - assert response.greeting == "Hello, Alice!" - - @pytest.mark.asyncio - async def test_greet_without_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - await auth_greet_client.greet(GreetRequest(name="Alice")) - - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Missing or invalid token" in exc_info.value.message - - @pytest.mark.asyncio - async def test_greet_with_invalid_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - await auth_greet_client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer invalid-token"} - ) - - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Invalid token" in exc_info.value.message - ``` - -=== "WSGI" - - ```python - import pytest - import httpx - from connectrpc.code import Code - from connectrpc.errors import ConnectError - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class AuthGreetServiceSync(GreetServiceSync): - def greet(self, request, ctx): - # Check for authorization header - auth = ctx.request_headers().get("authorization") - if not auth or not auth.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED, "Missing or invalid token") - - # Validate token (simplified) - token = auth[7:] # Remove "Bearer " prefix - if token != "valid-token": - raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") - - return GreetResponse(greeting=f"Hello, {request.name}!") - - @pytest.fixture - def auth_greet_client(): - app = GreetServiceWSGIApplication(AuthGreetServiceSync()) - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClientSync("http://test", session=session) - - def test_greet_with_valid_token(auth_greet_client): - response = auth_greet_client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer valid-token"} - ) - assert response.greeting == "Hello, Alice!" - - def test_greet_without_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - auth_greet_client.greet(GreetRequest(name="Alice")) - - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Missing or invalid token" in exc_info.value.message - - def test_greet_with_invalid_token(auth_greet_client): - with pytest.raises(ConnectError) as exc_info: - auth_greet_client.greet( - GreetRequest(name="Alice"), - headers={"authorization": "Bearer invalid-token"} - ) - - assert exc_info.value.code == Code.UNAUTHENTICATED - assert "Invalid token" in exc_info.value.message - ``` From 395a23446e8f6119904b4d7994ddd8991d07f268 Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 19 Oct 2025 14:23:30 +0900 Subject: [PATCH 07/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 162 ++++++++++++++++-------------------------------- 1 file changed, 52 insertions(+), 110 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index eaaf4c9..b9cd21b 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -68,11 +68,11 @@ Here's a minimal example without any test framework: print(response.greeting) # "Hello, Alice!" ``` -This pattern works with any test framework (pytest, unittest) or none at all. The examples below show how to integrate with pytest, but the core testing approach remains the same. +This pattern works with any test framework (pytest, unittest) or none at all. The examples below show how to integrate with both pytest and unittest. ## Testing servers -### In-memory testing +### Using pytest Testing the service we created in the [Getting Started](getting-started.md) guide looks like this: @@ -124,6 +124,56 @@ Testing the service we created in the [Getting Started](getting-started.md) guid assert response.greeting == "Hello, Alice!" ``` +### Using unittest + +The same in-memory testing approach works with unittest: + +=== "ASGI" + + ```python + import asyncio + import httpx + import unittest + from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient + from greet.v1.greet_pb2 import GreetRequest + from server import Greeter + + class TestGreet(unittest.TestCase): + def test_greet(self): + async def run_test(): + app = GreetServiceASGIApplication(Greeter()) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClient("http://test", session=session) + response = await client.greet(GreetRequest(name="Alice")) + self.assertEqual(response.greeting, "Hello, Alice!") + + asyncio.run(run_test()) + ``` + +=== "WSGI" + + ```python + import httpx + import unittest + from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync + from greet.v1.greet_pb2 import GreetRequest + from server import GreeterSync + + class TestGreet(unittest.TestCase): + def test_greet(self): + app = GreetServiceWSGIApplication(GreeterSync()) + with httpx.Client( + transport=httpx.WSGITransport(app=app), + base_url="http://test" + ) as session: + client = GreetServiceClientSync("http://test", session=session) + response = client.greet(GreetRequest(name="Alice")) + self.assertEqual(response.greeting, "Hello, Alice!") + ``` + This approach: - Tests your full application stack (routing, serialization, error handling) @@ -425,111 +475,3 @@ Test interceptors as part of your full application stack. For example, testing a ``` See the [Interceptors](interceptors.md) guide for more details on implementing interceptors. - -## Test organization - -Organize your tests in a `tests/` directory at the root of your project: - -``` -my-project/ -├── greet/ -│ └── v1/ -│ ├── greet_connect.py -│ └── greet_pb2.py -├── server.py -├── tests/ -│ ├── __init__.py -│ ├── test_greet.py -│ └── test_integration.py -└── pyproject.toml -``` - -For test framework-specific patterns like shared fixtures or test discovery, consult your test framework's documentation ([pytest](https://docs.pytest.org/), [unittest](https://docs.python.org/3/library/unittest.html)). - -## Practical examples - -### Testing with mock external dependencies - -Use fixtures to mock external services: - -=== "ASGI" - - ```python - import pytest - import pytest_asyncio - import httpx - from unittest.mock import AsyncMock - from greet.v1.greet_connect import GreetService, GreetServiceASGIApplication, GreetServiceClient - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class DatabaseGreetService(GreetService): - def __init__(self, db): - self.db = db - - async def greet(self, request, ctx): - # Fetch greeting from database - greeting_template = await self.db.get_greeting_template() - return GreetResponse(greeting=greeting_template.format(name=request.name)) - - @pytest.fixture - def mock_db(): - """Mock database for testing.""" - db = AsyncMock() - db.get_greeting_template.return_value = "Hello, {name}!" - return db - - @pytest_asyncio.fixture - async def greet_client_with_db(mock_db): - app = GreetServiceASGIApplication(DatabaseGreetService(mock_db)) - async with httpx.AsyncClient( - transport=httpx.ASGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClient("http://test", session=session) - - @pytest.mark.asyncio - async def test_greet_with_database(greet_client_with_db, mock_db): - response = await greet_client_with_db.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" - mock_db.get_greeting_template.assert_called_once() - ``` - -=== "WSGI" - - ```python - import pytest - import httpx - from unittest.mock import Mock - from greet.v1.greet_connect import GreetServiceSync, GreetServiceWSGIApplication, GreetServiceClientSync - from greet.v1.greet_pb2 import GreetRequest, GreetResponse - - class DatabaseGreetServiceSync(GreetServiceSync): - def __init__(self, db): - self.db = db - - def greet(self, request, ctx): - # Fetch greeting from database - greeting_template = self.db.get_greeting_template() - return GreetResponse(greeting=greeting_template.format(name=request.name)) - - @pytest.fixture - def mock_db(): - """Mock database for testing.""" - db = Mock() - db.get_greeting_template.return_value = "Hello, {name}!" - return db - - @pytest.fixture - def greet_client_with_db(mock_db): - app = GreetServiceWSGIApplication(DatabaseGreetServiceSync(mock_db)) - with httpx.Client( - transport=httpx.WSGITransport(app=app), - base_url="http://test" - ) as session: - yield GreetServiceClientSync("http://test", session=session) - - def test_greet_with_database(greet_client_with_db, mock_db): - response = greet_client_with_db.greet(GreetRequest(name="Alice")) - assert response.greeting == "Hello, Alice!" - mock_db.get_greeting_template.assert_called_once() - ``` From dd043152f841b1c864e16119aa27e2f7dbe02a6b Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 19 Oct 2025 14:41:38 +0900 Subject: [PATCH 08/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 13 ++++--------- 1 file changed, 4 insertions(+), 9 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index b9cd21b..bf9dc94 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -2,21 +2,16 @@ This guide covers testing connect-python services and clients. -> **Note:** The examples in this guide use a fictional `GreetService` for demonstration purposes. In your actual project, replace these with your own service definitions. - ## Setup -Install the required testing dependencies: +Install httpx for testing connect-python services and clients: ```bash -pip install pytest pytest-asyncio httpx +pip install httpx +# or: uv add --dev httpx ``` -Or if using uv: - -```bash -uv add --dev pytest pytest-asyncio httpx -``` +For pytest examples in this guide, you'll also need pytest (and pytest-asyncio for async tests). unittest requires no additional dependencies. ## Recommended approach: In-memory testing From b44bc2959edc9b8590d69a12d208bceb09e5dee2 Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 19 Oct 2025 15:00:56 +0900 Subject: [PATCH 09/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 94 +++++++++++++++++++++++++++++++------------------ 1 file changed, 59 insertions(+), 35 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index bf9dc94..d1f1ac7 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -356,11 +356,12 @@ For testing client code that calls Connect services, use the same in-memory test ## Testing interceptors -Test interceptors as part of your full application stack. For example, testing an authentication interceptor: +Test interceptors as part of your full application stack. For example, testing the `ServerAuthInterceptor` from the [Interceptors](interceptors.md#metadata-interceptors) guide: === "ASGI" ```python + from contextvars import ContextVar, Token import httpx import pytest from connectrpc.code import Code @@ -369,23 +370,27 @@ Test interceptors as part of your full application stack. For example, testing a from greet.v1.greet_pb2 import GreetRequest from server import Greeter - class AuthInterceptor: - def __init__(self, valid_token: str): - self.valid_token = valid_token - self.authenticated_calls = 0 + _auth_token = ContextVar["auth_token"]("current_auth_token") - async def on_start(self, ctx): - auth = ctx.request_headers().get("authorization") - if not auth or auth != f"Bearer {self.valid_token}": - raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") - self.authenticated_calls += 1 + class ServerAuthInterceptor: + def __init__(self, valid_tokens: list[str]): + self._valid_tokens = valid_tokens - async def on_end(self, token, ctx): - pass + async def on_start(self, ctx) -> Token["auth_token"]: + authorization = ctx.request_headers().get("authorization") + if not authorization or not authorization.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED) + token = authorization[len("Bearer "):] + if token not in self._valid_tokens: + raise ConnectError(Code.PERMISSION_DENIED) + return _auth_token.set(token) + + async def on_end(self, token: Token["auth_token"], ctx): + _auth_token.reset(token) @pytest.mark.asyncio - async def test_auth_interceptor(): - interceptor = AuthInterceptor("secret-token") + async def test_server_auth_interceptor(): + interceptor = ServerAuthInterceptor(["valid-token"]) app = GreetServiceASGIApplication( Greeter(), interceptors=[interceptor] @@ -400,23 +405,31 @@ Test interceptors as part of your full application stack. For example, testing a # Valid token succeeds response = await client.greet( GreetRequest(name="Alice"), - headers={"authorization": "Bearer secret-token"} + headers={"authorization": "Bearer valid-token"} ) assert response.greeting == "Hello, Alice!" - assert interceptor.authenticated_calls == 1 - # Invalid token fails + # Invalid token format fails with UNAUTHENTICATED with pytest.raises(ConnectError) as exc_info: await client.greet( GreetRequest(name="Bob"), - headers={"authorization": "Bearer wrong-token"} + headers={"authorization": "invalid"} ) assert exc_info.value.code == Code.UNAUTHENTICATED + + # Wrong token fails with PERMISSION_DENIED + with pytest.raises(ConnectError) as exc_info: + await client.greet( + GreetRequest(name="Bob"), + headers={"authorization": "Bearer wrong-token"} + ) + assert exc_info.value.code == Code.PERMISSION_DENIED ``` === "WSGI" ```python + from contextvars import ContextVar, Token import httpx import pytest from connectrpc.code import Code @@ -425,22 +438,26 @@ Test interceptors as part of your full application stack. For example, testing a from greet.v1.greet_pb2 import GreetRequest from server import GreeterSync - class AuthInterceptorSync: - def __init__(self, valid_token: str): - self.valid_token = valid_token - self.authenticated_calls = 0 + _auth_token = ContextVar["auth_token"]("current_auth_token") - def on_start(self, ctx): - auth = ctx.request_headers().get("authorization") - if not auth or auth != f"Bearer {self.valid_token}": - raise ConnectError(Code.UNAUTHENTICATED, "Invalid token") - self.authenticated_calls += 1 + class ServerAuthInterceptor: + def __init__(self, valid_tokens: list[str]): + self._valid_tokens = valid_tokens - def on_end(self, token, ctx): - pass + def on_start_sync(self, ctx) -> Token["auth_token"]: + authorization = ctx.request_headers().get("authorization") + if not authorization or not authorization.startswith("Bearer "): + raise ConnectError(Code.UNAUTHENTICATED) + token = authorization[len("Bearer "):] + if token not in self._valid_tokens: + raise ConnectError(Code.PERMISSION_DENIED) + return _auth_token.set(token) - def test_auth_interceptor(): - interceptor = AuthInterceptorSync("secret-token") + def on_end_sync(self, token: Token["auth_token"], ctx): + _auth_token.reset(token) + + def test_server_auth_interceptor(): + interceptor = ServerAuthInterceptor(["valid-token"]) app = GreetServiceWSGIApplication( GreeterSync(), interceptors=[interceptor] @@ -455,18 +472,25 @@ Test interceptors as part of your full application stack. For example, testing a # Valid token succeeds response = client.greet( GreetRequest(name="Alice"), - headers={"authorization": "Bearer secret-token"} + headers={"authorization": "Bearer valid-token"} ) assert response.greeting == "Hello, Alice!" - assert interceptor.authenticated_calls == 1 - # Invalid token fails + # Invalid token format fails with UNAUTHENTICATED with pytest.raises(ConnectError) as exc_info: client.greet( GreetRequest(name="Bob"), - headers={"authorization": "Bearer wrong-token"} + headers={"authorization": "invalid"} ) assert exc_info.value.code == Code.UNAUTHENTICATED + + # Wrong token fails with PERMISSION_DENIED + with pytest.raises(ConnectError) as exc_info: + client.greet( + GreetRequest(name="Bob"), + headers={"authorization": "Bearer wrong-token"} + ) + assert exc_info.value.code == Code.PERMISSION_DENIED ``` See the [Interceptors](interceptors.md) guide for more details on implementing interceptors. From 893a25a15b0a12e09fa67e797c743e161f218e16 Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Sun, 19 Oct 2025 15:14:03 +0900 Subject: [PATCH 10/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 40 ++-------------------------------------- 1 file changed, 2 insertions(+), 38 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index d1f1ac7..1c8dc00 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -361,33 +361,15 @@ Test interceptors as part of your full application stack. For example, testing t === "ASGI" ```python - from contextvars import ContextVar, Token import httpx import pytest from connectrpc.code import Code from connectrpc.errors import ConnectError from greet.v1.greet_connect import GreetServiceASGIApplication, GreetServiceClient from greet.v1.greet_pb2 import GreetRequest + from interceptors import ServerAuthInterceptor from server import Greeter - _auth_token = ContextVar["auth_token"]("current_auth_token") - - class ServerAuthInterceptor: - def __init__(self, valid_tokens: list[str]): - self._valid_tokens = valid_tokens - - async def on_start(self, ctx) -> Token["auth_token"]: - authorization = ctx.request_headers().get("authorization") - if not authorization or not authorization.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED) - token = authorization[len("Bearer "):] - if token not in self._valid_tokens: - raise ConnectError(Code.PERMISSION_DENIED) - return _auth_token.set(token) - - async def on_end(self, token: Token["auth_token"], ctx): - _auth_token.reset(token) - @pytest.mark.asyncio async def test_server_auth_interceptor(): interceptor = ServerAuthInterceptor(["valid-token"]) @@ -429,33 +411,15 @@ Test interceptors as part of your full application stack. For example, testing t === "WSGI" ```python - from contextvars import ContextVar, Token import httpx import pytest from connectrpc.code import Code from connectrpc.errors import ConnectError from greet.v1.greet_connect import GreetServiceWSGIApplication, GreetServiceClientSync from greet.v1.greet_pb2 import GreetRequest + from interceptors import ServerAuthInterceptor from server import GreeterSync - _auth_token = ContextVar["auth_token"]("current_auth_token") - - class ServerAuthInterceptor: - def __init__(self, valid_tokens: list[str]): - self._valid_tokens = valid_tokens - - def on_start_sync(self, ctx) -> Token["auth_token"]: - authorization = ctx.request_headers().get("authorization") - if not authorization or not authorization.startswith("Bearer "): - raise ConnectError(Code.UNAUTHENTICATED) - token = authorization[len("Bearer "):] - if token not in self._valid_tokens: - raise ConnectError(Code.PERMISSION_DENIED) - return _auth_token.set(token) - - def on_end_sync(self, token: Token["auth_token"], ctx): - _auth_token.reset(token) - def test_server_auth_interceptor(): interceptor = ServerAuthInterceptor(["valid-token"]) app = GreetServiceWSGIApplication( From 58d665319954acaa3720b83a109791a8622e068b Mon Sep 17 00:00:00 2001 From: i2y <6240399+i2y@users.noreply.github.com> Date: Mon, 20 Oct 2025 12:54:27 +0900 Subject: [PATCH 11/11] fix Signed-off-by: i2y <6240399+i2y@users.noreply.github.com> --- docs/testing.md | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/docs/testing.md b/docs/testing.md index 1c8dc00..8d5d4b8 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -4,14 +4,7 @@ This guide covers testing connect-python services and clients. ## Setup -Install httpx for testing connect-python services and clients: - -```bash -pip install httpx -# or: uv add --dev httpx -``` - -For pytest examples in this guide, you'll also need pytest (and pytest-asyncio for async tests). unittest requires no additional dependencies. +For pytest examples in this guide, you'll need pytest and pytest-asyncio. unittest requires no additional dependencies. ## Recommended approach: In-memory testing