feat: add clientless Pub/Sub emulator service

.github/ci/provider-groups.json

@@ -41,6 +41,7 @@
     "tests/test_elasticsearch.py::test_plugin_imports_without_elasticsearch_clients",
     "tests/test_mongodb.py::test_plugin_imports_without_pymongo",
     "tests/test_mssql.py::test_plugin_imports_without_pymssql",
+    "tests/test_pubsub.py::test_plugin_imports_without_google_cloud_pubsub",
     "tests/test_spanner.py::test_plugin_imports_without_google_cloud_spanner",
     "tests/test_valkey.py::test_plugin_imports_without_valkey"
   ],
@@ -162,6 +163,14 @@
       ],
       "docs_globs": ["docs/supported-databases/postgres.rst"]
     },
+    "pubsub": {
+      "source_globs": ["src/pytest_databases/docker/pubsub.py"],
+      "test_paths": ["tests/test_pubsub.py"],
+      "images": [
+        "gcr.io/google.com/cloudsdktool/google-cloud-cli:577.0.0-emulators"
+      ],
+      "docs_globs": ["docs/supported-databases/pubsub.rst"]
+    },
     "redis": {
       "source_globs": ["src/pytest_databases/docker/redis.py"],
       "test_paths": ["tests/test_redis.py"],

README.md

@@ -33,6 +33,7 @@ Ready-made database fixtures for your pytest tests.
 - **Google AlloyDB Omni**: Simplified Omni installation for easy testing.
 - **Google Spanner**: The latest cloud-emulator from Google is available
 - **Google BigQuery**: Unofficial BigQuery emulator
+- **Google Pub/Sub**: Official Google Cloud Pub/Sub emulator
 - **CockroachDB**: Version latest is available
 - **Redis**: Latest version
 - **Valkey**: Latest version

docs/changelog.rst

@@ -9,6 +9,8 @@ Next
 Added
 ~~~~~
 
+* Clientless Google Pub/Sub emulator fixtures backed by the official Google Cloud CLI emulator image, with Docker and
+  Podman-compatible lifecycle management, project-scoped xdist isolation, and an in-container protocol smoke check.
 * First-class Docker and Podman runtime selection through one Docker-compatible API transport. Closes
   `gh-146 <https://github.com/litestar-org/pytest-databases/issues/146>`_.
 * Runtime-neutral ``container_client``, ``container_service``, and ``ContainerService`` names while retaining all

docs/getting-started/installation.rst

@@ -7,6 +7,9 @@ First, install the base package using pip:
 
    pip install pytest-databases
 
+Clientless service fixtures such as the Google Pub/Sub emulator are included in the base package. Install the client
+used by your application separately; ``pytest-databases`` does not add a Google Cloud Pub/Sub client dependency.
+
 Optional Database Support
 ~~~~~~~~~~~~~~~~~~~~~~~~~
 

docs/supported-databases/index.rst

@@ -14,6 +14,7 @@ This section provides detailed information on the supported databases, including
    sqlserver
    spanner
    bigquery
+   pubsub
    cockroachdb
    yugabyte
    mongodb

docs/supported-databases/pubsub.rst

@@ -0,0 +1,91 @@
+Google Pub/Sub
+==============
+
+Integration with `Google Cloud Pub/Sub <https://cloud.google.com/pubsub>`_ using Google's official
+`Pub/Sub emulator <https://cloud.google.com/pubsub/docs/emulator>`_. The fixture starts the version-pinned Google Cloud
+CLI emulator image and validates it with the ``gcloud`` tooling already present in that image. The package does not
+depend on ``google-cloud-pubsub`` or mutate Google credential environment variables.
+
+Installation
+------------
+
+Install the base package:
+
+.. code-block:: bash
+
+   pip install pytest-databases
+
+Install the Pub/Sub client used by your application separately, if it needs one.
+
+Usage
+-----
+
+Enable the plugin and pass the service metadata to your application. Standard Google clients recognize
+``PUBSUB_EMULATOR_HOST``; set it in your own fixture or application configuration so the parent test process remains
+under your control.
+
+.. code-block:: python
+
+   import pytest
+
+   from pytest_databases.docker.pubsub import PubSubService
+
+   pytest_plugins = ["pytest_databases.docker.pubsub"]
+
+
+   @pytest.fixture
+   def pubsub_environment(pubsub_service: PubSubService, monkeypatch: pytest.MonkeyPatch) -> PubSubService:
+       monkeypatch.setenv("PUBSUB_EMULATOR_HOST", pubsub_service.emulator_host)
+       monkeypatch.setenv("PUBSUB_PROJECT_ID", pubsub_service.project)
+       return pubsub_service
+
+
+   def test_publish_event(pubsub_environment: PubSubService) -> None:
+       assert pubsub_environment.emulator_host == (
+           f"{pubsub_environment.host}:{pubsub_environment.port}"
+       )
+       # Call application code configured by pubsub_environment here.
+
+The service fixture waits for both the emulator's documented ``Server started`` log and the mapped TCP endpoint. Before
+yielding, the bundled ``gcloud`` CLI creates a topic and subscription inside the emulator container, publishes a payload,
+pulls and acknowledges it, and removes the smoke resources. No host-side Google client or credentials are involved.
+
+Available fixtures
+------------------
+
+* ``pubsub_image``: Google Cloud CLI ``577.0.0-emulators`` image. Override it to test a newer compatible release.
+* ``pubsub_project``: ``pytest-databases`` outside xdist and a worker-suffixed project under xdist.
+* ``xdist_pubsub_isolation_level``: ``database`` by default; override with ``server`` for one container per worker.
+* ``pubsub_service``: :class:`~pytest_databases.docker.pubsub.PubSubService` containing ``host``, ``port``, ``project``,
+  ``emulator_host``, and the underlying container.
+
+Parallel isolation
+------------------
+
+The default ``database`` isolation shares one emulator container and assigns each xdist worker a separate project ID.
+The emulator namespaces resources by project, so workers may use the same topic and subscription names. Override the
+isolation fixture when a test needs a separate emulator process:
+
+.. code-block:: python
+
+   import pytest
+
+
+   @pytest.fixture(scope="session")
+   def xdist_pubsub_isolation_level() -> str:
+       return "server"
+
+Limitations
+-----------
+
+The emulator is intended for local development and can differ from the production service. IAM integration, ACL checks,
+and some production features are not implemented. Data is ephemeral, and tests should not use the emulator to validate
+credentials, permissions, quotas, or production delivery guarantees.
+
+Service API
+-----------
+
+.. automodule:: pytest_databases.docker.pubsub
+   :members:
+   :undoc-members:
+   :show-inheritance:

src/pytest_databases/docker/pubsub.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import socket
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import pytest
+
+from pytest_databases.helpers import get_xdist_worker_id
+from pytest_databases.types import ServiceContainer, XdistIsolationLevel
+
+if TYPE_CHECKING:
+    from collections.abc import Generator
+
+    from docker.models.containers import Container
+
+    from pytest_databases._service import ContainerService
+
+PUBSUB_EMULATOR_IMAGE = "gcr.io/google.com/cloudsdktool/google-cloud-cli:577.0.0-emulators"
+PUBSUB_EMULATOR_PORT = 8085
+
+_PUBSUB_SMOKE_SCRIPT = """\
+set -eu
+topic="pytest-databases-smoke"
+subscription="pytest-databases-smoke"
+cleanup() {
+    gcloud pubsub subscriptions delete "$subscription" --quiet >/dev/null 2>&1 || true
+    gcloud pubsub topics delete "$topic" --quiet >/dev/null 2>&1 || true
+}
+trap cleanup EXIT
+cleanup
+gcloud pubsub topics create "$topic" --quiet
+gcloud pubsub subscriptions create "$subscription" --topic="$topic" --quiet
+gcloud pubsub topics publish "$topic" --message=pytest-databases --quiet
+payload="$(timeout 30 gcloud pubsub subscriptions pull "$subscription" --auto-ack --limit=1 --format='value(message.data)' --quiet)"
+test "$payload" = "pytest-databases"
+"""
+
+
+@dataclass
+class PubSubService(ServiceContainer):
+    project: str
+    emulator_host: str
+
+
+@pytest.fixture(scope="session")
+def pubsub_image() -> str:
+    return PUBSUB_EMULATOR_IMAGE
+
+
+@pytest.fixture(scope="session")
+def pubsub_project() -> str:
+    worker_id = get_xdist_worker_id()
+    if worker_id is None or worker_id == "master":
+        return "pytest-databases"
+    return f"pytest-databases-{worker_id}"
+
+
+@pytest.fixture(scope="session")
+def xdist_pubsub_isolation_level() -> XdistIsolationLevel:
+    return "database"
+
+
+def _pubsub_start_command(project: str) -> list[str]:
+    return [
+        "gcloud",
+        "beta",
+        "emulators",
+        "pubsub",
+        "start",
+        f"--host-port=0.0.0.0:{PUBSUB_EMULATOR_PORT}",
+        f"--project={project}",
+    ]
+
+
+def _is_pubsub_responsive(service: ServiceContainer) -> bool:
+    try:
+        connection = socket.create_connection((service.host, service.port), timeout=1)
+    except OSError:
+        return False
+    connection.close()
+    return True
+
+
+def _smoke_pubsub_emulator(
+    emulator_container: Container,
+    *,
+    project: str,
+) -> None:
+    result = emulator_container.exec_run(
+        ["bash", "-c", _PUBSUB_SMOKE_SCRIPT],
+        environment={
+            "CLOUDSDK_API_ENDPOINT_OVERRIDES_PUBSUB": f"http://localhost:{PUBSUB_EMULATOR_PORT}/",
+            "CLOUDSDK_AUTH_DISABLE_CREDENTIALS": "true",
+            "CLOUDSDK_CORE_PROJECT": project,
+        },
+    )
+    if result.exit_code != 0:
+        output = result.output.decode(errors="replace") if isinstance(result.output, bytes) else str(result.output)
+        msg = f"Pub/Sub emulator smoke check failed with exit code {result.exit_code}: {output[-2000:]}"
+        raise RuntimeError(msg)
+
+
+@pytest.fixture(scope="session")
+def pubsub_service(
+    container_service: ContainerService,
+    pubsub_image: str,
+    pubsub_project: str,
+    xdist_pubsub_isolation_level: XdistIsolationLevel,
+) -> Generator[PubSubService, None, None]:
+    worker_id = get_xdist_worker_id()
+    container_name = "pubsub"
+    if xdist_pubsub_isolation_level == "server" and worker_id not in {None, "master"}:
+        container_name = f"{container_name}_{worker_id}"
+
+    with container_service.run(
+        image=pubsub_image,
+        command=_pubsub_start_command(pubsub_project),
+        name=container_name,
+        container_port=PUBSUB_EMULATOR_PORT,
+        wait_for_log="Server started",
+        check=_is_pubsub_responsive,
+        timeout=60,
+        transient=xdist_pubsub_isolation_level == "server",
+    ) as service:
+        _smoke_pubsub_emulator(
+            service.container,
+            project=pubsub_project,
+        )
+        yield PubSubService(
+            container=service.container,
+            host=service.host,
+            port=service.port,
+            project=pubsub_project,
+            emulator_host=f"{service.host}:{service.port}",
+        )