pytest-gcpsecretmanager 0.2.0
A PyTest plugin for mocking GCP's Secret Manager
License: MIT
Requires Python: >=3.12

# pytest-gcpsecretmanager

A pytest plugin that provides an in-memory fake for Google Cloud Secret Manager. No Docker, no emulator, no GCP credentials required.

Both `SecretManagerServiceClient` (sync) and `SecretManagerServiceAsyncClient` (async) are transparently patched for the duration of each test.

## Installation

```bash
pip install pytest-gcpsecretmanager
```

The plugin requires Python 3.12+ and pytest 7.0+. It does **not** require `google-cloud-secret-manager` to be installed — if the SDK is absent, fake modules are injected so that `from google.cloud.secretmanager import SecretManagerServiceClient` still works in your code under test.

## Quick start

```python
import pytest
from google.cloud.secretmanager import SecretManagerServiceClient

@pytest.mark.secret("my-api-key", "s3cret")
def test_reads_secret(secret_manager):
    client = SecretManagerServiceClient()
    response = client.access_secret_version(
        name="projects/test-project/secrets/my-api-key/versions/latest"
    )
    assert response.payload.data == b"s3cret"
```

The `secret_manager` fixture activates patching and returns the underlying `SecretStore`. The `@pytest.mark.secret` marker pre-populates secrets before the test runs.

## Fixtures

### `secret_manager`

Function-scoped. Activates patching and returns a `SecretStore` instance. Any `SecretManagerServiceClient` or `SecretManagerServiceAsyncClient` instantiated during the test will use this in-memory store.

```python
def test_programmatic_setup(secret_manager):
    secret_manager.set_secret("db-password", "hunter2")
    secret_manager.set_secret("binary-key", b"\x00\x01\x02")

    client = SecretManagerServiceClient()
    resp = client.access_secret_version(
        name="projects/test-project/secrets/db-password/versions/latest"
    )
    assert resp.payload.data == b"hunter2"
```

The store also supports multiple versions:

```python
def test_versioned_secrets(secret_manager):
    secret_manager.set_secret_sequence("rotating-key", ["v1", "v2", "v3"])

    client = SecretManagerServiceClient()
    resp = client.access_secret_version(
        name="projects/test-project/secrets/rotating-key/versions/2"
    )
    assert resp.payload.data == b"v2"
```

### `secret_manager_client`

Returns a `FakeSecretManagerServiceClient` instance backed by the current test's store.

```python
def test_with_client(secret_manager_client):
    secret_manager_client.create_secret(
        parent="projects/test-project", secret_id="new-secret"
    )
    secret_manager_client.add_secret_version(
        parent="projects/test-project/secrets/new-secret",
        payload={"data": b"payload"},
    )
    resp = secret_manager_client.access_secret_version(
        name="projects/test-project/secrets/new-secret/versions/1"
    )
    assert resp.payload.data == b"payload"
```

### `secret_manager_async_client`

Returns a `FakeSecretManagerServiceAsyncClient` — same API, but all methods are `async`.

```python
async def test_async_client(secret_manager_async_client):
    await secret_manager_async_client.create_secret(
        parent="projects/test-project", secret_id="async-secret"
    )
    await secret_manager_async_client.add_secret_version(
        parent="projects/test-project/secrets/async-secret",
        payload={"data": b"hello"},
    )
    resp = await secret_manager_async_client.access_secret_version(
        name="projects/test-project/secrets/async-secret/versions/1"
    )
    assert resp.payload.data == b"hello"
```

### `secret_failure_injector`

Returns the active `_FailureInjector` for programmatic failure injection (see [Failure injection](#failure-injection) below).

## Markers

### `@pytest.mark.secret(secret_id, value, *, project=None)`

Pre-populate a secret before the test runs. The default project is `"test-project"`.

```python
# Single version
@pytest.mark.secret("api-key", "my-key")
def test_single(secret_manager): ...

# Multiple versions (pass a list)
@pytest.mark.secret("rotating", ["v1", "v2", "v3"])
def test_versions(secret_manager): ...

# Custom project
@pytest.mark.secret("api-key", "my-key", project="my-project")
def test_custom_project(secret_manager): ...

# Stack multiple markers
@pytest.mark.secret("key-a", "value-a")
@pytest.mark.secret("key-b", "value-b")
def test_multiple(secret_manager): ...
```

### `@pytest.mark.secret_failure(method, exception, *, transient=False, count=1)`

Inject a failure into a specific client method.

```python
from pytest_gcpsecretmanager import NotFound

# Permanent failure — every call raises
@pytest.mark.secret_failure("access_secret_version", NotFound("gone"))
def test_not_found(secret_manager):
    client = SecretManagerServiceClient()
    with pytest.raises(NotFound):
        client.access_secret_version(name="projects/p/secrets/s/versions/1")

# Transient failure — fails `count` times, then succeeds
@pytest.mark.secret_failure(
    "access_secret_version", NotFound("retry me"), transient=True, count=2
)
@pytest.mark.secret("key", "val")
def test_transient(secret_manager):
    client = SecretManagerServiceClient()
    for _ in range(2):
        with pytest.raises(NotFound):
            client.access_secret_version(
                name="projects/test-project/secrets/key/versions/latest"
            )
    # Third call succeeds
    resp = client.access_secret_version(
        name="projects/test-project/secrets/key/versions/latest"
    )
    assert resp.payload.data == b"val"
```

## Failure injection

In addition to the marker, you can inject failures programmatically via the `secret_failure_injector` fixture:

```python
from pytest_gcpsecretmanager import PermissionDenied

def test_programmatic_failure(secret_manager, secret_failure_injector):
    secret_failure_injector.add_permanent_failure(
        "create_secret", PermissionDenied("nope")
    )
    client = SecretManagerServiceClient()
    with pytest.raises(PermissionDenied):
        client.create_secret(parent="projects/test-project", secret_id="x")
```

Available exception types: `NotFound`, `AlreadyExists`, `PermissionDenied`, `FailedPrecondition`, `InvalidArgument`, `ResourceExhausted`, `DeadlineExceeded`.

## Supported API methods

Both sync and async clients support:

| Method | Description |
|---|---|
| `create_secret` | Create a new secret |
| `get_secret` | Get secret metadata |
| `delete_secret` | Delete a secret and all its versions |
| `list_secrets` | List secrets in a project |
| `add_secret_version` | Add a new version to a secret |
| `get_secret_version` | Get version metadata |
| `access_secret_version` | Access the secret payload |
| `destroy_secret_version` | Permanently destroy a version |
| `disable_secret_version` | Disable a version |
| `enable_secret_version` | Re-enable a disabled version |
| `list_secret_versions` | List all versions of a secret |
| `secret_path` | Static helper: build a secret resource name |
| `secret_version_path` | Static helper: build a version resource name |

## How patching works

When the `secret_manager` fixture is active:

1. If `google-cloud-secret-manager` is installed, the real client classes are patched via `unittest.mock.patch` across all known import paths (`google.cloud.secretmanager`, `google.cloud.secretmanager_v1`, etc.).
2. If the SDK is **not** installed, fake modules are injected into `sys.modules` so that `from google.cloud.secretmanager import SecretManagerServiceClient` resolves to the fake client.

This means your application code doesn't need any special imports or conditional logic — the fake is transparent.

## Response types

Responses use lightweight dataclasses that mirror the GCP protobuf shapes:

- `AccessSecretVersionResponse` — has `.name` and `.payload` (a `SecretPayload` with `.data` bytes and `.data_crc32c`)
- `Secret` — has `.name`, `.replication`, `.create_time`, `.labels`
- `SecretVersion` — has `.name`, `.create_time`, `.state`

## License

MIT