Messaging

Testing Google Cloud Pub/Sub: Emulator, Fake, and Integration Test Patterns

22 Oct 2026 — 4 min read

```tldr Google Cloud Pub/Sub can be tested locally using the official Pub/Sub emulator, which runs as a lightweight process or Docker container. The emulator supports topics, subscriptions, push endpoints, message ordering, and dead letter topics — giving you a realistic test environment without GCP credentials or billing. ```

```takeaways **Use the Pub/Sub emulator for all local and CI tests.** Point PUBSUB_EMULATOR_HOST to localhost:8085 — the client library automatically uses the emulator instead of real GCP.

Testcontainers gcr.io/google.com/cloudsdktool/google-cloud-cli runs the emulator in Docker. No local gcloud install needed in CI.

Create topics and subscriptions programmatically in test setup. The emulator starts empty — your test setup code must create the resources before publishing.

Message ordering requires an ordering key. Test that ordered subscriptions receive messages in sequence when using setOrderingKey().

Pull vs push subscriptions need different test patterns. Pull subscriptions are easy to test synchronously; push subscriptions need an HTTP endpoint or a mock server. ```

Pub/Sub Emulator Setup

Install and run locally

gcloud components install pubsub-emulator
gcloud beta emulators pubsub start --host-port=localhost:8085

Set the environment variable so the client library routes to the emulator:

export PUBSUB_EMULATOR_HOST=localhost:8085

Run with Docker (for CI)

# docker-compose.yml
services:
  pubsub-emulator:
    image: gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators
    command: gcloud beta emulators pubsub start --host-port=0.0.0.0:8085
    ports:
      - "8085:8085"

Testcontainers (Java)

@Container
static GenericContainer<?> pubsubEmulator = new GenericContainer<>(
    "gcr.io/google.com/cloudsdktool/google-cloud-cli:emulators"
)
.withCommand("gcloud beta emulators pubsub start --host-port=0.0.0.0:8085")
.withExposedPorts(8085)
.waitingFor(Wait.forLogMessage(".*Server started.*", 1));

Client Setup for Tests

import com.google.api.gax.core.NoCredentialsProvider;
import com.google.api.gax.grpc.GrpcTransportChannel;
import com.google.api.gax.rpc.FixedTransportChannelProvider;
import com.google.cloud.pubsub.v1.*;
import io.grpc.ManagedChannelBuilder;

class PubSubTestBase {
    protected TransportChannelProvider channelProvider;
    protected CredentialsProvider credentialsProvider;
    protected String projectId = "test-project";

    void setupEmulatorClient(String emulatorHost) throws Exception {
        ManagedChannel channel = ManagedChannelBuilder
            .forTarget(emulatorHost)
            .usePlaintext()
            .build();

        channelProvider = FixedTransportChannelProvider.create(
            GrpcTransportChannel.create(channel)
        );
        credentialsProvider = NoCredentialsProvider.create();
    }

    TopicAdminClient topicAdminClient() throws Exception {
        return TopicAdminClient.create(
            TopicAdminSettings.newBuilder()
                .setTransportChannelProvider(channelProvider)
                .setCredentialsProvider(credentialsProvider)
                .build()
        );
    }

    SubscriptionAdminClient subscriptionAdminClient() throws Exception {
        return SubscriptionAdminClient.create(
            SubscriptionAdminSettings.newBuilder()
                .setTransportChannelProvider(channelProvider)
                .setCredentialsProvider(credentialsProvider)
                .build()
        );
    }
}

Publishing and Subscribing

class PubSubPublisherTest extends PubSubTestBase {

    @BeforeEach
    void createResources() throws Exception {
        setupEmulatorClient("localhost:8085");

        // Create topic
        try (TopicAdminClient topicAdmin = topicAdminClient()) {
            topicAdmin.createTopic(TopicName.of(projectId, "test-topic"));
        }

        // Create subscription
        try (SubscriptionAdminClient subAdmin = subscriptionAdminClient()) {
            subAdmin.createSubscription(
                Subscription.newBuilder()
                    .setName(SubscriptionName.of(projectId, "test-sub").toString())
                    .setTopic(TopicName.of(projectId, "test-topic").toString())
                    .setAckDeadlineSeconds(10)
                    .build()
            );
        }
    }

    @Test
    void publisherSendsMessageToSubscriber() throws Exception {
        Publisher publisher = Publisher.newBuilder(TopicName.of(projectId, "test-topic"))
            .setChannelProvider(channelProvider)
            .setCredentialsProvider(credentialsProvider)
            .build();

        PubsubMessage message = PubsubMessage.newBuilder()
            .setData(ByteString.copyFromUtf8("hello pub/sub"))
            .putAttributes("source", "test")
            .build();

        publisher.publish(message).get(10, TimeUnit.SECONDS);
        publisher.shutdown();

        // Pull the message
        Subscriber subscriber = Subscriber.newBuilder(
            SubscriptionName.of(projectId, "test-sub"),
            (msg, ack) -> {
                assertEquals("hello pub/sub", msg.getData().toStringUtf8());
                assertEquals("test", msg.getAttributesOrDefault("source", ""));
                ack.ack();
            }
        )
        .setChannelProvider(channelProvider)
        .setCredentialsProvider(credentialsProvider)
        .build();

        subscriber.startAsync().awaitRunning();
        Thread.sleep(2000);  // wait for delivery
        subscriber.stopAsync().awaitTerminated();
    }
}

Synchronous Pull Testing

For easier test assertions, use synchronous pull:

@Test
void pullMessagesSync() throws Exception {
    // Publish
    publisherFor("orders-topic").publish(
        PubsubMessage.newBuilder()
            .setData(ByteString.copyFromUtf8("{\"orderId\":\"123\"}"))
            .build()
    ).get();

    // Pull synchronously
    SubscriberStubSettings settings = SubscriberStubSettings.newBuilder()
        .setTransportChannelProvider(channelProvider)
        .setCredentialsProvider(credentialsProvider)
        .build();

    try (SubscriberStub subscriberStub = GrpcSubscriberStub.create(settings)) {
        PullRequest pullRequest = PullRequest.newBuilder()
            .setSubscription(SubscriptionName.of(projectId, "orders-sub").toString())
            .setMaxMessages(10)
            .build();

        PullResponse response = subscriberStub.pullCallable().call(pullRequest);

        assertEquals(1, response.getReceivedMessagesList().size());
        String data = response.getReceivedMessages(0).getMessage()
            .getData().toStringUtf8();
        assertEquals("{\"orderId\":\"123\"}", data);

        // Acknowledge
        List<String> ackIds = response.getReceivedMessagesList().stream()
            .map(ReceivedMessage::getAckId)
            .collect(Collectors.toList());
        AcknowledgeRequest ackRequest = AcknowledgeRequest.newBuilder()
            .setSubscription(SubscriptionName.of(projectId, "orders-sub").toString())
            .addAllAckIds(ackIds)
            .build();
        subscriberStub.acknowledgeCallable().call(ackRequest);
    }
}

Message Ordering

@Test
void orderedSubscriptionReceivesMessagesInOrder() throws Exception {
    // Create topic with ordering enabled
    topicAdminClient().createTopic(
        Topic.newBuilder()
            .setName(TopicName.of(projectId, "ordered-topic").toString())
            .build()
    );

    // Create ordered subscription
    subscriptionAdminClient().createSubscription(
        Subscription.newBuilder()
            .setName(SubscriptionName.of(projectId, "ordered-sub").toString())
            .setTopic(TopicName.of(projectId, "ordered-topic").toString())
            .setEnableMessageOrdering(true)
            .build()
    );

    Publisher publisher = Publisher.newBuilder(TopicName.of(projectId, "ordered-topic"))
        .setChannelProvider(channelProvider)
        .setCredentialsProvider(credentialsProvider)
        .setEnableMessageOrdering(true)
        .build();

    String orderKey = "customer-42";
    for (int i = 1; i <= 5; i++) {
        publisher.publish(
            PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8("event-" + i))
                .setOrderingKey(orderKey)
                .build()
        ).get();
    }
    publisher.shutdown();

    List<String> received = Collections.synchronizedList(new ArrayList<>());
    CountDownLatch latch = new CountDownLatch(5);

    Subscriber subscriber = Subscriber.newBuilder(
        SubscriptionName.of(projectId, "ordered-sub"),
        (msg, ack) -> {
            received.add(msg.getData().toStringUtf8());
            ack.ack();
            latch.countDown();
        }
    )
    .setChannelProvider(channelProvider)
    .setCredentialsProvider(credentialsProvider)
    .build();

    subscriber.startAsync().awaitRunning();
    assertTrue(latch.await(15, TimeUnit.SECONDS));
    subscriber.stopAsync().awaitTerminated();

    assertEquals(List.of("event-1","event-2","event-3","event-4","event-5"), received);
}

Dead Letter Topic Testing

@Test
void unacknowledgedMessagesGoToDeadLetterTopic() throws Exception {
    String mainTopic = "payment-events";
    String dlqTopic = "payment-events-dlq";

    // Create dead letter subscription
    subscriptionAdminClient().createSubscription(
        Subscription.newBuilder()
            .setName(SubscriptionName.of(projectId, "payment-sub").toString())
            .setTopic(TopicName.of(projectId, mainTopic).toString())
            .setAckDeadlineSeconds(10)
            .setDeadLetterPolicy(DeadLetterPolicy.newBuilder()
                .setDeadLetterTopic(TopicName.of(projectId, dlqTopic).toString())
                .setMaxDeliveryAttempts(3)
                .build())
            .build()
    );

    // Publish and nack repeatedly to trigger DLQ
    publisherFor(mainTopic).publish(
        PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("bad-event")).build()
    ).get();

    // Nack 3 times
    for (int attempt = 0; attempt < 3; attempt++) {
        PullResponse response = syncPull("payment-sub", 1);
        if (!response.getReceivedMessagesList().isEmpty()) {
            nack("payment-sub", response.getReceivedMessages(0).getAckId());
        }
        Thread.sleep(500);
    }

    // Check DLQ
    Thread.sleep(2000);
    PullResponse dlqResponse = syncPull("payment-events-dlq-sub", 1);
    assertFalse(dlqResponse.getReceivedMessagesList().isEmpty(), "Message should be in DLQ");
    assertEquals(
        "bad-event",
        dlqResponse.getReceivedMessages(0).getMessage().getData().toStringUtf8()
    );
}

Python Example (pytest)

# conftest.py
import pytest
from google.cloud import pubsub_v1
import os

os.environ["PUBSUB_EMULATOR_HOST"] = "localhost:8085"
PROJECT_ID = "test-project"

@pytest.fixture(scope="session")
def publisher():
    return pubsub_v1.PublisherClient()

@pytest.fixture(scope="session")
def subscriber():
    return pubsub_v1.SubscriberClient()

@pytest.fixture
def topic(publisher):
    topic_path = publisher.topic_path(PROJECT_ID, "test-topic")
    publisher.create_topic(request={"name": topic_path})
    yield topic_path
    publisher.delete_topic(request={"topic": topic_path})

@pytest.fixture
def subscription(subscriber, topic):
    sub_path = subscriber.subscription_path(PROJECT_ID, "test-sub")
    subscriber.create_subscription(request={"name": sub_path, "topic": topic})
    yield sub_path
    subscriber.delete_subscription(request={"subscription": sub_path})

def test_publish_subscribe(publisher, subscriber, topic, subscription):
    future = publisher.publish(topic, b"test message", source="pytest")
    message_id = future.result()
    assert message_id

    response = subscriber.pull(
        request={"subscription": subscription, "max_messages": 1}
    )
    assert len(response.received_messages) == 1
    msg = response.received_messages[0].message
    assert msg.data == b"test message"
    assert msg.attributes["source"] == "pytest"

    subscriber.acknowledge(
        request={
            "subscription": subscription,
            "ack_ids": [response.received_messages[0].ack_id]
        }
    )

Common Mistakes

Not setting PUBSUB_EMULATOR_HOST — without this env var, the client connects to real GCP and fails with auth errors.

Forgetting to create topics and subscriptions — the emulator starts empty. Your @BeforeEach must create all resources.

Not acknowledging messages — unacked messages get redelivered after the ack deadline (10s by default in emulator), causing duplicate processing in long-running tests.

Testing push endpoints without an HTTP server — push subscriptions POST to a URL. For unit tests, use pull subscriptions or start a test HTTP server with MockWebServer.