GCP Testing

GCP Pub/Sub Testing: Integration Testing, Dead Letter Queues, and Message Validation

HelpMeTest

HelpMeTest

— 7 min read
GCP Pub/Sub Testing: Integration Testing, Dead Letter Queues, and Message Validation

Pub/Sub is the messaging backbone for most GCP architectures. It decouples services, enables event-driven patterns, and handles traffic spikes by buffering messages. It also introduces testing complexity that synchronous APIs don't have: messages are asynchronous, ordering isn't guaranteed, delivery can happen more than once, and failures go to dead letter queues rather than returning errors.

This guide covers practical Pub/Sub testing — from unit testing publishers and subscribers to integration testing with the emulator to validating dead letter queue behavior.

What Makes Pub/Sub Testing Different

In a synchronous HTTP API, you call a function, get a response, and check it. In Pub/Sub:

  • Publishing is decoupled from subscribing — you can't check that a subscriber processed a message by looking at the publisher's response
  • At-least-once delivery means your subscriber might receive the same message twice — your tests need to handle this
  • Ordering is only guaranteed if you use ordering keys — otherwise messages arrive in any sequence
  • Failures don't surface immediately — a failed message retry attempt might happen seconds or minutes later

These characteristics mean you need to test publishers and subscribers independently, then test them together with delay and retry assertions.

Unit Testing Pub/Sub Publishers

Publishers are generally simpler to test. You mock the Pub/Sub client and verify:

  • The right topic is used
  • The message data is correctly encoded
  • Attributes are set correctly
  • Errors from the Pub/Sub API are handled
# publisher.py
import json
from google.cloud import pubsub_v1

def publish_order_created(project_id: str, topic_id: str, order: dict) -> str:
    """Publish an order.created event to Pub/Sub."""
    publisher = pubsub_v1.PublisherClient()
    topic_path = publisher.topic_path(project_id, topic_id)

    message_data = json.dumps(order).encode('utf-8')
    future = publisher.publish(
        topic_path,
        message_data,
        event_type='order.created',
        source_service='orders-service',
    )

    return future.result()  # Returns message ID
# test_publisher.py
import json
import pytest
from unittest.mock import patch, MagicMock
from publisher import publish_order_created

def test_publishes_to_correct_topic():
    with patch('publisher.pubsub_v1.PublisherClient') as MockClient:
        mock_publisher = MagicMock()
        MockClient.return_value = mock_publisher
        mock_publisher.publish.return_value.result.return_value = 'msg-id-123'

        publish_order_created('my-project', 'orders-topic', {'id': 'ord-1'})

        # Verify correct topic path
        mock_publisher.publish.assert_called_once()
        call_args = mock_publisher.publish.call_args
        assert 'orders-topic' in call_args[0][0]

def test_message_data_is_json_encoded():
    with patch('publisher.pubsub_v1.PublisherClient') as MockClient:
        mock_publisher = MagicMock()
        MockClient.return_value = mock_publisher
        mock_publisher.publish.return_value.result.return_value = 'msg-id-123'

        order = {'id': 'ord-1', 'total': 99.99, 'items': ['item-a']}
        publish_order_created('my-project', 'orders-topic', order)

        call_args = mock_publisher.publish.call_args
        published_data = json.loads(call_args[0][1].decode('utf-8'))
        assert published_data == order

def test_sets_event_type_attribute():
    with patch('publisher.pubsub_v1.PublisherClient') as MockClient:
        mock_publisher = MagicMock()
        MockClient.return_value = mock_publisher
        mock_publisher.publish.return_value.result.return_value = 'msg-id-123'

        publish_order_created('my-project', 'orders-topic', {'id': 'ord-1'})

        call_kwargs = mock_publisher.publish.call_args[1]
        assert call_kwargs.get('event_type') == 'order.created'

def test_handles_publish_failure():
    with patch('publisher.pubsub_v1.PublisherClient') as MockClient:
        mock_publisher = MagicMock()
        MockClient.return_value = mock_publisher
        mock_publisher.publish.return_value.result.side_effect = Exception('API error')

        with pytest.raises(Exception, match='API error'):
            publish_order_created('my-project', 'orders-topic', {'id': 'ord-1'})

Unit Testing Pub/Sub Subscribers

Subscriber testing is more complex because you need to simulate the Pub/Sub message envelope:

# subscriber.py
import json
import base64
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/pubsub/push', methods=['POST'])
def handle_pubsub_push():
    """Cloud Run push subscription endpoint."""
    envelope = request.get_json()

    if not envelope or 'message' not in envelope:
        return jsonify({'error': 'Bad Request'}), 400

    pubsub_message = envelope['message']
    data = base64.b64decode(pubsub_message.get('data', '')).decode('utf-8')
    message = json.loads(data)
    attributes = pubsub_message.get('attributes', {})

    try:
        process_order(message, attributes)
        return jsonify({}), 200  # 200 = acknowledge message
    except ValueError as e:
        return jsonify({'error': str(e)}), 400  # 400 = nack, will retry
    except Exception as e:
        return jsonify({'error': 'Internal error'}), 500  # 500 = nack, will retry

def process_order(order: dict, attributes: dict):
    if 'id' not in order:
        raise ValueError('Order must have an id field')
    # ... processing logic
# test_subscriber.py
import base64
import json
import pytest
from unittest.mock import patch
from subscriber import app

@pytest.fixture
def client():
    app.config['TESTING'] = True
    with app.test_client() as c:
        yield c

def make_pubsub_envelope(data: dict, attributes: dict = None) -> dict:
    """Helper to build a valid Pub/Sub push envelope."""
    encoded = base64.b64encode(json.dumps(data).encode()).decode()
    return {
        'message': {
            'data': encoded,
            'attributes': attributes or {},
            'messageId': 'test-message-id',
        },
        'subscription': 'projects/test/subscriptions/test-sub',
    }

def test_acknowledges_valid_message(client):
    envelope = make_pubsub_envelope({'id': 'ord-1', 'total': 99.99})
    response = client.post('/pubsub/push', json=envelope)
    assert response.status_code == 200

def test_nacks_invalid_message_schema(client):
    """Validation error → 400 → Pub/Sub retries or sends to DLQ."""
    envelope = make_pubsub_envelope({'total': 99.99})  # Missing 'id'
    response = client.post('/pubsub/push', json=envelope)
    assert response.status_code == 400

def test_rejects_missing_message_envelope(client):
    response = client.post('/pubsub/push', json={})
    assert response.status_code == 400

def test_handles_malformed_json_in_data(client):
    """Malformed JSON in message data shouldn't crash the service."""
    bad_envelope = {
        'message': {
            'data': base64.b64encode(b'not-json{').decode(),
            'attributes': {},
        }
    }
    response = client.post('/pubsub/push', json=bad_envelope)
    # Should return 400 (nack) not 500 (unexpected error)
    assert response.status_code in [400, 500]
    # Ideally 400 — update this test to enforce 400 and fix the handler

Integration Testing With the Pub/Sub Emulator

The Pub/Sub emulator runs locally and behaves like the real service:

# Start the emulator
gcloud emulators pubsub start --host-port=localhost:8085

# In another terminal, set the environment variable
export PUBSUB_EMULATOR_HOST=localhost:8085
# test_pubsub_integration.py
import os
import json
import time
import pytest
from google.cloud import pubsub_v1

# Emulator is set via PUBSUB_EMULATOR_HOST env var
PROJECT = 'test-project'
TOPIC_ID = 'orders'
SUBSCRIPTION_ID = 'orders-sub'

@pytest.fixture(scope='module')
def publisher():
    return pubsub_v1.PublisherClient()

@pytest.fixture(scope='module')
def subscriber():
    return pubsub_v1.SubscriberClient()

@pytest.fixture(scope='module')
def topic(publisher):
    topic_path = publisher.topic_path(PROJECT, TOPIC_ID)
    try:
        publisher.create_topic(request={'name': topic_path})
    except Exception:
        pass  # Already exists
    return topic_path

@pytest.fixture(scope='module')
def subscription(subscriber, topic):
    sub_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION_ID)
    try:
        subscriber.create_subscription(request={'name': sub_path, 'topic': topic})
    except Exception:
        pass  # Already exists
    return sub_path

def test_publish_and_subscribe_roundtrip(publisher, subscriber, topic, subscription):
    """Full roundtrip: publish a message and verify it's received."""
    order = {'id': 'ord-integration-001', 'total': 150.00}

    # Publish
    future = publisher.publish(topic, json.dumps(order).encode('utf-8'))
    message_id = future.result(timeout=5)
    assert message_id is not None

    # Pull with timeout
    received_messages = []

    def callback(message):
        received_messages.append(json.loads(message.data.decode('utf-8')))
        message.ack()

    streaming_pull = subscriber.subscribe(subscription, callback=callback)

    # Wait for message
    timeout = time.time() + 10
    while not received_messages and time.time() < timeout:
        time.sleep(0.1)

    streaming_pull.cancel()

    assert len(received_messages) >= 1
    assert received_messages[0]['id'] == 'ord-integration-001'
    assert received_messages[0]['total'] == 150.00

def test_at_least_once_delivery(publisher, subscriber, topic, subscription):
    """Pub/Sub guarantees at-least-once: subscriber must be idempotent."""
    order = {'id': 'ord-idempotency-test'}

    publisher.publish(topic, json.dumps(order).encode('utf-8'))

    # Pull and NACK (don't acknowledge) — message will be redelivered
    response = subscriber.pull(
        request={'subscription': subscription, 'max_messages': 1},
        timeout=5,
    )

    if response.received_messages:
        # Nack: don't call ack(), just let it expire
        # In a real test, verify your subscriber handles duplicates correctly
        ack_id = response.received_messages[0].ack_id
        subscriber.modify_ack_deadline(
            request={
                'subscription': subscription,
                'ack_ids': [ack_id],
                'ack_deadline_seconds': 0,  # Immediately redeliver
            }
        )

Testing Dead Letter Queue Behavior

Dead letter queues (DLQs) capture messages that fail repeatedly. Testing DLQ behavior requires:

  1. Publishing a message
  2. Verifying the subscriber nacks it (by returning a non-200 status or throwing)
  3. Waiting for the retry count to be exhausted
  4. Verifying the message appears in the DLQ subscription
# test_dead_letter.py
import pytest
import json
import time
from google.cloud import pubsub_v1

def test_invalid_message_reaches_dead_letter_queue():
    """Messages that fail validation should end up in the DLQ."""
    publisher = pubsub_v1.PublisherClient()
    subscriber = pubsub_v1.SubscriberClient()

    # Publish a deliberately malformed message
    bad_message = {'invalid': 'no id field'}
    publisher.publish(
        publisher.topic_path(PROJECT, TOPIC_ID),
        json.dumps(bad_message).encode('utf-8'),
    )

    # In a push subscription test, the subscriber handler returns 400
    # We verify the DLQ subscription receives it after max_delivery_attempts

    dlq_subscription = subscriber.subscription_path(PROJECT, 'orders-dlq-sub')

    # Wait for DLQ delivery (depends on retry configuration)
    # In emulator: retries happen quickly; in production: can be minutes
    received_in_dlq = []
    deadline = time.time() + 30

    while not received_in_dlq and time.time() < deadline:
        response = subscriber.pull(
            request={'subscription': dlq_subscription, 'max_messages': 10},
            timeout=2,
        )
        for msg in response.received_messages:
            data = json.loads(msg.message.data.decode('utf-8'))
            if data.get('invalid') == 'no id field':
                received_in_dlq.append(data)
                subscriber.acknowledge(
                    request={'subscription': dlq_subscription, 'ack_ids': [msg.ack_id]}
                )

    assert received_in_dlq, "Message did not reach DLQ within 30 seconds"

Message Schema Validation

Pub/Sub doesn't enforce message schemas by default. Add schema validation to catch malformed messages early:

# schema_validator.py
import json
import jsonschema

ORDER_SCHEMA = {
    'type': 'object',
    'required': ['id', 'total', 'items'],
    'properties': {
        'id': {'type': 'string', 'minLength': 1},
        'total': {'type': 'number', 'minimum': 0},
        'items': {
            'type': 'array',
            'minItems': 1,
            'items': {'type': 'string'},
        },
    },
    'additionalProperties': False,
}

def validate_order_message(data: dict):
    try:
        jsonschema.validate(data, ORDER_SCHEMA)
    except jsonschema.ValidationError as e:
        raise ValueError(f"Invalid message schema: {e.message}")
# test_schema_validator.py
import pytest
from schema_validator import validate_order_message

def test_valid_order_passes():
    validate_order_message({'id': 'ord-1', 'total': 50.0, 'items': ['item-a']})

def test_missing_required_field_raises():
    with pytest.raises(ValueError, match='Invalid message schema'):
        validate_order_message({'total': 50.0, 'items': ['item-a']})  # No 'id'

def test_negative_total_raises():
    with pytest.raises(ValueError):
        validate_order_message({'id': 'ord-1', 'total': -10.0, 'items': ['item-a']})

def test_empty_items_raises():
    with pytest.raises(ValueError):
        validate_order_message({'id': 'ord-1', 'total': 50.0, 'items': []})

def test_unknown_fields_raise():
    with pytest.raises(ValueError):
        validate_order_message({
            'id': 'ord-1', 'total': 50.0, 'items': ['item-a'],
            'unexpected_field': 'value',
        })

CI Configuration for Pub/Sub Testing

# .github/workflows/pubsub-tests.yml
name: Pub/Sub Integration Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Install Google Cloud SDK
        uses: google-github-actions/setup-gcloud@v1

      - name: Start Pub/Sub emulator
        run: |
          gcloud components install pubsub-emulator --quiet
          gcloud emulators pubsub start --host-port=localhost:8085 &
          sleep 3  # Give emulator time to start
          echo "PUBSUB_EMULATOR_HOST=localhost:8085" >> $GITHUB_ENV

      - name: Unit tests
        run: pytest tests/unit/ -v

      - name: Integration tests
        env:
          PUBSUB_EMULATOR_HOST: localhost:8085
        run: pytest tests/integration/ -v -k "pubsub"

Key Testing Principles for Pub/Sub

Test acknowledgment behavior explicitly. A 200 response acknowledges; non-200 causes retry. Your tests should verify both paths work correctly.

Test idempotency. Pub/Sub can deliver the same message twice. Your subscriber must produce the same result for duplicate messages. Write tests that send the same message twice and verify state is correct.

Don't test Pub/Sub itself. Test your publisher and subscriber code. Use mocks for unit tests and the emulator for integration tests. You're not verifying GCP's reliability — you're verifying your code.

Test DLQ configuration. A DLQ that's misconfigured (wrong permissions, wrong subscription) will silently drop failed messages. Test end-to-end DLQ delivery in staging before production.

HelpMeTest monitors live applications end-to-end, including event-driven systems. Use it to verify that messages published by one service are correctly processed by subscribers in your actual GCP environment. Start free →

Read more

Start now free