From df185b9129cf38e2a48927a62e9d2e255cc5f115 Mon Sep 17 00:00:00 2001 From: Peter Lamut Date: Thu, 31 Oct 2019 22:22:49 +0200 Subject: [PATCH] chore(pubsub): add subscriber role test for streaming (#9507) Pulling the messages using a streaming pull should work with accounts having only the pubsub.subscriber role. This commits add a test that covers this aspect. --- pubsub/tests/system.py | 40 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/pubsub/tests/system.py b/pubsub/tests/system.py index cb00a4b91ecd..fd7473e1e53b 100644 --- a/pubsub/tests/system.py +++ b/pubsub/tests/system.py @@ -17,6 +17,7 @@ import datetime import itertools import operator as op +import os import threading import time @@ -488,6 +489,45 @@ def test_streaming_pull_max_messages( finally: subscription_future.cancel() # trigger clean shutdown + @pytest.mark.skipif( + "KOKORO_GFILE_DIR" not in os.environ, + reason="Requires Kokoro environment with a limited subscriber service account.", + ) + def test_streaming_pull_subscriber_permissions_sufficient( + self, publisher, topic_path, subscriber, subscription_path, cleanup + ): + + # Make sure the topic and subscription get deleted. + cleanup.append((publisher.delete_topic, topic_path)) + cleanup.append((subscriber.delete_subscription, subscription_path)) + + # create a topic and subscribe to it + publisher.create_topic(topic_path) + subscriber.create_subscription(subscription_path, topic_path) + + # A service account granting only the pubsub.subscriber role must be used. + filename = os.path.join( + os.environ["KOKORO_GFILE_DIR"], "pubsub-subscriber-service-account.json" + ) + streaming_pull_subscriber = type(subscriber).from_service_account_file(filename) + + # Subscribe to the topic, publish a message, and verify that subscriber + # successfully pulls and processes it. + callback = StreamingPullCallback(processing_time=0.01, resolve_at_msg_count=1) + future = streaming_pull_subscriber.subscribe(subscription_path, callback) + self._publish_messages(publisher, topic_path, batch_sizes=[1]) + + try: + callback.done_future.result(timeout=10) + except exceptions.TimeoutError: + pytest.fail( + "Timeout: receiving/processing streamed messages took too long." + ) + else: + assert 1 in callback.seen_message_ids + finally: + future.cancel() + def _publish_messages(self, publisher, topic_path, batch_sizes): """Publish ``count`` messages in batches and wait until completion.""" publish_futures = []