500 Internal error from Core - while using pytest

[mbrancato]

When running tests for async streaming under pytest, I get the following gRPC error:

    async for response in stream:
  File "/Users/mike/.pyenv/versions/3.11.9/envs/global-3.11/lib/python3.11/site-packages/google/api_core/grpc_helpers_async.py", line 109, in _wrapped_aiter
    raise exceptions.from_grpc_error(rpc_error) from rpc_error
google.api_core.exceptions.InternalServerError: 500 Internal error from Core

In some cases, the PubSub emulator gets a NullPointerException as well, but its a little more difficult to reproduce.

Environment details

• OS type and version: MacOS Sequoia
• Python version: Python 3.11.9
• pip version: pip 24.1.1
• google-cloud-pubsub version:

Name: google-cloud-pubsub
Version: 2.25.2
Summary: Google Cloud Pub/Sub API client library
Home-page: https://github.com/googleapis/python-pubsub
Author: Google LLC
Author-email: [email protected]
License: Apache 2.0
Location: /Users/mike/.pyenv/versions/3.11.9/envs/global-3.11/lib/python3.11/site-packages
Requires: google-api-core, google-auth, grpc-google-iam-v1, grpcio, grpcio-status, opentelemetry-api, opentelemetry-sdk, proto-plus, protobuf

Steps to reproduce

1. Start the pubsub sumulator
2. Create an asyncio test using unittest
3. Perform an async streaming pull

Code example

Running the following code will fail with and RPC error 500 Internal error from Core when run using pytest:

import os
import sys
import traceback
from unittest import IsolatedAsyncioTestCase
from testcontainers.google import PubSubContainer
class TestPubSubService(IsolatedAsyncioTestCase):

async def test_stuff(self) -> None:

test_project_id = "test-project"
    pubsub_emulator = PubSubContainer(project=test_project_id)
    print("Starting PubSub emulator", flush=True, file=sys.stderr)
    try:
        pubsub_emulator.start()
        host = pubsub_emulator.get_pubsub_emulator_host()
        os.environ["PUBSUB_EMULATOR_HOST"] = host
        os.environ["PUBSUB_PROJECT_ID"] = test_project_id

        from google.auth import credentials
        from google.pubsub_v1 import SubscriberAsyncClient, StreamingPullRequest
        from google.pubsub_v1.services.subscriber.transports import (
            SubscriberGrpcAsyncIOTransport,
        )

        pubsub_config = {
            "client_options": {"api_endpoint": pubsub_emulator.get_pubsub_emulator_host()},
            "credentials": credentials.AnonymousCredentials(),
        }

        subscriber_client = SubscriberAsyncClient(**pubsub_config)
        transport = subscriber_client.transport
        if isinstance(transport, SubscriberGrpcAsyncIOTransport):
            await transport.grpc_channel.channel_ready()

        subscription_path = f"projects/{test_project_id}/subscriptions/subscription_value"

        def request_generator():
            while True:
                yield StreamingPullRequest(
                    subscription=subscription_path,
                    stream_ack_deadline_seconds=120,
                )

        try:
            stream = await subscriber_client.streaming_pull(requests=request_generator())
        except Exception as e:
            print(traceback.format_exc(), flush=True, file=sys.stderr)
            raise e

        print("Listening for messages on subscription", flush=True)

        try:
            async for response in stream:
                for received_message in response.received_messages:
                    print(received_message.message)
        except Exception as e:
            print(traceback.format_exc(), flush=True, file=sys.stderr)
            raise e
    finally:
        pubsub_emulator.stop()

So it fails with:

pytest test_foo.py

But it passes with:

python -m unittest test_foo.py

Given there is an RPC error, I wanted to open here, but I plan to open an issue with pytest-asyncio as well.