Metadata-Version: 2.4
Name: pytest-grpc-aio
Version: 0.3.0
Summary: pytest plugin for grpc.aio
Author-email: Conrad Bzura <conradbzura@gmail.com>
Project-URL: Homepage, https://github.com/conradbzura/pytest-grpc-aio
Classifier: Framework :: Pytest
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: POSIX
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Topic :: Software Development :: Testing
Classifier: Topic :: Software Development :: Quality Assurance
Classifier: Topic :: Utilities
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.6
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pytest>=3.6.0
Dynamic: license-file

# pytest-grpc-aio

Write tests for gRPC services with pytest using `grpc.aio` (asyncio-based gRPC).

## Installation

```bash
pip install pytest-grpc-aio
```

## Features

- **Async/await support** - Built on `grpc.aio` for native asyncio integration
- **Flexible servicer configuration** - Provide servicers via fixtures or at test time
- **Real and fake servers** - Run tests against actual gRPC servers or direct Python calls
- **Context tracking** - Access server/channel details via `grpc_context` fixture
- **Type-safe** - Full type hints with Protocol-based interfaces

## Quick Start

### 1. Define Your Service

Given a proto file:

```proto
syntax = "proto3";

package example.v1;

service EchoService {
    rpc SayHello(HelloRequest) returns (HelloResponse) {}
}

message HelloRequest {
    string name = 1;
}

message HelloResponse {
    string message = 1;
}
```

Implement your servicer:

```python
from example_pb2 import HelloRequest, HelloResponse
from example_pb2_grpc import EchoServiceServicer


class EchoServicer(EchoServiceServicer):
    async def SayHello(self, request: HelloRequest, context) -> HelloResponse:
        return HelloResponse(message=f"Hello, {request.name}!")
```

### 2. Configure Test Fixtures

```python
import pytest
from example_pb2_grpc import add_EchoServiceServicer_to_server, EchoServiceStub
from servicer import EchoServicer


@pytest.fixture
def grpc_add_to_server():
    """Required: Tell pytest how to register your servicer."""
    return add_EchoServiceServicer_to_server


@pytest.fixture
def grpc_stub_cls():
    """Required: Tell pytest which stub class to use."""
    return EchoServiceStub


# Option 1: Provide servicer via fixture (module/session scope)
@pytest.fixture(scope="module")
def grpc_servicer():
    """Optional: Provide a default servicer for all tests."""
    return EchoServicer()
```

### 3. Write Tests

#### Using the Fixture-Provided Servicer

```python
import pytest
from example_pb2 import HelloRequest


@pytest.mark.asyncio
async def test_say_hello(grpc_aio_stub):
    """Test using the servicer from grpc_servicer fixture."""
    async with grpc_aio_stub() as stub:
        request = HelloRequest(name="World")
        response = await stub.SayHello(request)
        assert response.message == "Hello, World!"
```

#### Providing Servicer Per-Test

```python
@pytest.mark.asyncio
async def test_with_custom_servicer(grpc_aio_stub):
    """Override the servicer for a specific test."""
    custom_servicer = EchoServicer()

    async with grpc_aio_stub(servicer=custom_servicer) as stub:
        request = HelloRequest(name="Custom")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Custom!"
```

#### Testing Error Handling

```python
import grpc


class ErrorServicer(EchoServiceServicer):
    async def SayHello(self, request: HelloRequest, context):
        await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid name")


@pytest.mark.asyncio
async def test_error_handling(grpc_aio_stub):
    """Test error scenarios."""
    error_servicer = ErrorServicer()

    async with grpc_aio_stub(servicer=error_servicer) as stub:
        request = HelloRequest(name="")

        with pytest.raises(grpc.RpcError) as exc_info:
            await stub.SayHello(request)

        assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT
```

## Advanced Usage

### Working with Credentials

```python
import grpc
from pathlib import Path


@pytest.fixture
def my_channel_credentials():
    """Provide SSL credentials for secure channels."""
    cert_path = Path("/path/to/cert.pem")
    return grpc.ssl_channel_credentials(
        root_certificates=cert_path.read_bytes()
    )


@pytest.mark.asyncio
async def test_secure_connection(grpc_aio_stub, my_channel_credentials):
    """Test with SSL/TLS credentials."""
    async with grpc_aio_stub(credentials=my_channel_credentials) as stub:
        request = HelloRequest(name="Secure")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Secure!"
```

### Using Channel Options

```python
@pytest.mark.asyncio
async def test_with_options(grpc_aio_stub):
    """Test with custom channel options."""
    options = [
        ("grpc.max_receive_message_length", 1024 * 1024 * 10),
        ("grpc.max_send_message_length", 1024 * 1024 * 10),
    ]

    async with grpc_aio_stub(options=options) as stub:
        request = HelloRequest(name="Options")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Options!"
```

### Accessing gRPC Context

The `grpc_context` fixture provides access to:
- `grpc_context.addr`: server address
- `grpc_context.server`: the gRPC server instance
- `grpc_context.channel`: the gRPC channel instance
- `grpc_context.servicer`: the servicer being tested
- `grpc_context.add_to_server`: the `add_to_server` function
- `grpc_context.interceptors`: any configured interceptors

```python
@pytest.mark.asyncio
async def test_inspect_context(grpc_aio_stub, grpc_context):
    """Access server and channel details via grpc_context."""
    async with grpc_aio_stub() as stub:
        print(f"Testing against server at {grpc_context.addr}")

        request = HelloRequest(name="Context")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Context!"
```

### Using Interceptors

```python
import grpc.aio


class LoggingInterceptor(grpc.aio.ServerInterceptor):
    async def intercept_service(self, continuation, handler_call_details):
        print(f"Handling RPC: {handler_call_details.method}")
        return await continuation(handler_call_details)


@pytest.fixture
def grpc_interceptors():
    """Add server interceptors."""
    return [LoggingInterceptor()]
```

### Lower-Level Access

If you need more control, you can use the channel or server fixtures directly:

```python
@pytest.mark.asyncio
async def test_with_channel(grpc_aio_channel, grpc_stub_cls):
    """Use the channel fixture directly."""
    servicer = EchoServicer()

    async with grpc_aio_channel(servicer) as channel:
        stub = grpc_stub_cls(channel)
        request = HelloRequest(name="Channel")
        response = await stub.SayHello(request)
        assert response.message == "Hello, Channel!"


@pytest.mark.asyncio
async def test_with_server(grpc_aio_server, grpc_addr, grpc_stub_cls):
    """Use the server fixture directly."""
    servicer = EchoServicer()

    async with grpc_aio_server(servicer):
        async with grpc.aio.insecure_channel(grpc_addr) as channel:
            stub = grpc_stub_cls(channel)
            request = HelloRequest(name="Server")
            response = await stub.SayHello(request)
            assert response.message == "Hello, Server!"
```

## Command-Line Options

The plugin provides several command-line options:

### Fake Server Mode

Run tests by calling service handlers directly (no real gRPC server):

```bash
pytest --grpc-fake-server
```

**Benefits:**
- Faster test execution
- Direct exception propagation (easier debugging)
- No network overhead

**Trade-offs:**
- Doesn't test actual gRPC serialization/networking
- May miss integration issues

**Example output with fake server:**

```
def test_error_handling(grpc_aio_stub):
    async with grpc_aio_stub(servicer=ErrorServicer()) as stub:
        request = HelloRequest(name="")
>       response = await stub.SayHello(request)

test_example.py:45:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

    async def SayHello(self, request: HelloRequest, context):
>       await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Invalid name")
E       FakeRpcError: Invalid name

servicer.py:12: FakeRpcError
```

### Configuring Worker Threads

Control the ThreadPoolExecutor used by gRPC servers:

```bash
# Set maximum workers via command line
pytest --grpc-max-workers=10

# Or in a test module
grpc_max_workers = 10
```

The effective value is the maximum of the CLI option and module variable.

**Use cases:**
- Test thread-safety of servicers
- Simulate concurrent client requests
- Stress test resource locking

## Available Fixtures

### Required User Fixtures

| Fixture | Scope | Description |
|---------|-------|-------------|
| `grpc_add_to_server` | Any | Function that registers your servicer with a gRPC server |
| `grpc_stub_cls` | Any | Your stub class (e.g., from `*_pb2_grpc.py`) |

### Optional User Fixtures

| Fixture | Scope | Description |
|---------|-------|-------------|
| `grpc_servicer` | Any | Default servicer instance to use in tests |
| `grpc_interceptors` | Any | List of gRPC interceptors |

### Provided Fixtures

| Fixture | Description |
|---------|-------------|
| `grpc_aio_stub` | **Most useful** - Returns a factory for creating async stubs with context managers |
| `grpc_aio_channel` | Returns a factory for creating async channels with context managers |
| `grpc_aio_server` | Returns a factory for creating async servers with context managers |
| `grpc_context` | Provides access to server/channel details during tests |
| `grpc_addr` | The address where the test server is listening |

## Synchronous gRPC (grpc.Server)

For legacy/sync gRPC code, use the non-aio fixtures:

```python
def test_sync(grpc_stub):
    """Use synchronous gRPC (no async/await)."""
    with grpc_stub() as stub:
        request = HelloRequest(name="Sync")
        response = stub.SayHello(request)
        assert response.message == "Hello, Sync!"
```

Available sync fixtures: `grpc_stub`, `grpc_channel`, `grpc_server`

## Type Safety

All fixtures and factories are fully typed using `typing.Protocol`:

```python
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from pytest_grpc_aio import GrpcAioStubFactory, GrpcContext

    # Your IDE will provide autocomplete and type checking
    stub_factory: GrpcAioStubFactory[EchoServiceStub, EchoServicer]
    context: GrpcContext
```

## Migration from pytest-grpc

If you're migrating from the original `pytest-grpc`:

1. **Use async fixtures**: Replace `grpc_stub` with `grpc_aio_stub`
2. **Use context managers**: Stubs are now created via `async with grpc_aio_stub() as stub:`
3. **Pass servicers explicitly**: Can override `grpc_servicer` by passing `servicer=` argument
4. **Add `@pytest.mark.asyncio`**: Required for async test functions

## Examples

See the `example/` directory for complete working examples including:
- Basic echo service tests
- Error handling
- Secure connections
- Custom interceptors
- Thread-safety testing

## Contributing

Contributions welcome! Please open an issue or PR on GitHub.

## License

MIT
