Pytest Gcpsecretmanager

pytest-gcpsecretmanager

Zero-config pytest plugin that intercepts SecretManagerServiceClient and SecretManagerServiceAsyncClient in-process. No Docker, no emulator binary, no google-cloud-secret-manager SDK required at test time.

Install: pip install pytest-gcpsecretmanager (or add to pyproject.toml dependencies).

Fixtures

All fixtures are function-scoped — state is fully isolated between tests.

secret_manager — primary fixture

Yields a SecretStore instance. Activates patching for the duration of the test. Any code that instantiates SecretManagerServiceClient() or SecretManagerServiceAsyncClient() gets the fake automatically.

def test_reads_config(secret_manager):
    secret_manager.set_secret("database-url", "postgresql://localhost/mydb")
    # Application code under test calls SecretManagerServiceClient() internally
    assert my_app.get_database_url() == "postgresql://localhost/mydb"

secret_manager_client

Returns a FakeSecretManagerServiceClient instance for direct client usage in sync tests. Depends on secret_manager.

def test_create_and_read(secret_manager_client):
    client = secret_manager_client
    client.create_secret(parent="projects/my-proj", secret_id="s", secret={})
    client.add_secret_version(parent="projects/my-proj/secrets/s", payload={"data": b"val"})
    resp = client.access_secret_version(name="projects/my-proj/secrets/s/versions/latest")
    assert resp.payload.data == b"val"

secret_manager_async_client

Returns a FakeSecretManagerServiceAsyncClient for async tests. Depends on secret_manager.

async def test_async_access(secret_manager_async_client):
    client = secret_manager_async_client
    await client.create_secret(parent="projects/p", secret_id="s", secret={})
    await client.add_secret_version(parent="projects/p/secrets/s", payload={"data": b"val"})
    resp = await client.access_secret_version(name="projects/p/secrets/s/versions/latest")
    assert resp.payload.data == b"val"

secret_failure_injector

Returns the _FailureInjector for programmatic failure injection. Depends on secret_manager.

def test_rate_limit_retry(secret_manager, secret_failure_injector):
    secret_manager.set_secret("s", "v")
    secret_failure_injector.add_transient_failure(
        "access_secret_version", ResourceExhausted("rate limited"), count=3
    )
    assert read_with_backoff("s") == "v"  # fails 3 times, then succeeds

Markers

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

Pre-populate secrets declaratively. value accepts str, bytes, or list[str|bytes] (multiple versions).

@pytest.mark.secret("api-key", "sk-test-12345")
@pytest.mark.secret("db-pass", "hunter2")
def test_app_init(secret_manager):
    app = create_app()
    assert app.config["API_KEY"] == "sk-test-12345"

@pytest.mark.secret("rotating-key", ["old-key", "new-key"])
def test_uses_latest(secret_manager):
    assert get_current_key() == "new-key"  # "latest" resolves to version 2

The default project is "test-project". Override with project="other-proj".

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

Inject failures into client methods.

# Permanent failure (all calls raise)
@pytest.mark.secret_failure("access_secret_version", PermissionDenied("denied"))
def test_permission_error(secret_manager):
    assert get_secret_or_default("x", default="fallback") == "fallback"

# Transient failure (first N calls raise, then succeed)
@pytest.mark.secret("my-secret", "value")
@pytest.mark.secret_failure("access_secret_version", DeadlineExceeded("timeout"), transient=True, count=2)
def test_retries(secret_manager):
    assert read_with_retry("my-secret") == "value"

Valid method names: access_secret_version, create_secret, get_secret, delete_secret, list_secrets, add_secret_version, get_secret_version, list_secret_versions, destroy_secret_version, disable_secret_version, enable_secret_version.

SecretStore API

The secret_manager fixture yields a SecretStore. Key methods:

Convenience (use these in most tests):

• set_secret(secret_id, value, *, project=None) — create-or-update with one version. value: str or bytes.
• set_secret_sequence(secret_id, values, *, project=None) — create with N versions. values: list[str|bytes].

Low-level (match GCP API semantics):

• create_secret(project, secret_id, *, labels=None) -> Secret
• get_secret(project, secret_id) -> Secret
• delete_secret(project, secret_id)
• list_secrets(project) -> list[Secret]
• add_secret_version(project, secret_id, data: bytes) -> SecretVersion
• access_secret_version(project, secret_id, version: str) -> AccessSecretVersionResponse
• get_secret_version(project, secret_id, version) -> SecretVersion
• list_secret_versions(project, secret_id) -> list[SecretVersion]
• destroy_secret_version(project, secret_id, version) -> SecretVersion
• disable_secret_version(project, secret_id, version) -> SecretVersion
• enable_secret_version(project, secret_id, version) -> SecretVersion

Version "latest" resolves to the highest-numbered ENABLED version.

Default project: "test-project".

Exceptions

Import from pytest_gcpsecretmanager.exceptions: NotFound, AlreadyExists, PermissionDenied, FailedPrecondition, InvalidArgument, ResourceExhausted, DeadlineExceeded.

These are the real google.api_core.exceptions classes when the SDK is installed, or lightweight stand-ins otherwise. Use them in secret_failure markers and assertions.

Fake Client Calling Conventions

The fake clients accept all three calling conventions the real SDK supports:

# Keyword arguments
client.access_secret_version(name="projects/p/secrets/s/versions/1")

# Dict request
client.access_secret_version(request={"name": "projects/p/secrets/s/versions/1"})

# Object with attributes
client.access_secret_version(request=some_protobuf_object)

Patching Scope

The plugin patches these import paths (sync and async variants):

• google.cloud.secretmanager.SecretManagerServiceClient
• google.cloud.secretmanager_v1.SecretManagerServiceClient
• google.cloud.secretmanager_v1.services.secret_manager_service.SecretManagerServiceClient

If google-cloud-secret-manager is not installed, fake modules are injected into sys.modules so imports still work.

Response Types

resp = client.access_secret_version(...) returns AccessSecretVersionResponse:

• resp.name — full resource name string
• resp.payload.data — bytes secret value
• resp.payload.data_crc32c — CRC32C checksum (auto-computed)

SecretVersion has: name, create_time, state (SecretVersionState.ENABLED/DISABLED/DESTROYED).

Secret has: name, replication, create_time, labels.