Pub/Sub: update how to test with mock

pubsub/cloud-client/iam_test.py

@@ -13,15 +13,17 @@
 # limitations under the License.
 
 import os
+import uuid
 
 from google.cloud import pubsub_v1
 import pytest
 
 import iam
 
+UUID = uuid.uuid4().hex
 PROJECT = os.environ['GCLOUD_PROJECT']
-TOPIC = 'iam-test-topic'
-SUBSCRIPTION = 'iam-test-subscription'
+TOPIC = 'iam-test-topic-' + UUID
+SUBSCRIPTION = 'iam-test-subscription-' + UUID
 
 
 @pytest.fixture(scope='module')
@@ -63,6 +65,15 @@ def subscription(subscriber_client, topic):
     yield subscription_path
 
 
+def _to_delete():
+    publisher_client = pubsub_v1.PublisherClient()
+    subscriber_client = pubsub_v1.SubscriberClient()
+    publisher_client.delete_topic(
+        'projects/{}/topics/{}'.format(PROJECT, TOPIC))
+    subscriber_client.delete_subscription(
+        'projects/{}/subscriptions/{}'.format(PROJECT, SUBSCRIPTION))
+
+
 def test_get_topic_policy(topic, capsys):
     iam.get_topic_policy(PROJECT, TOPIC)
 
@@ -109,3 +120,6 @@ def test_check_subscription_permissions(subscription, capsys):
 
     assert subscription in out
     assert 'pubsub.subscriptions.consume' in out
+
+    # Clean up resources.
+    _to_delete()

pubsub/cloud-client/publisher_test.py

@@ -14,6 +14,7 @@
 
 import os
 import time
+import uuid
 
 from gcp_devrel.testing import eventually_consistent
 from google.cloud import pubsub_v1
@@ -22,8 +23,9 @@
 
 import publisher
 
+UUID = uuid.uuid4().hex
 PROJECT = os.environ['GCLOUD_PROJECT']
-TOPIC = 'publisher-test-topic'
+TOPIC = 'publisher-test-topic-' + UUID
 
 
 @pytest.fixture
@@ -56,6 +58,12 @@ def new_sleep(period):
     return mock.patch('time.sleep', new=new_sleep)
 
 
+def _to_delete():
+    publisher_client = pubsub_v1.PublisherClient()
+    publisher_client.delete_topic('projects/{}/topics/{}'.format(
+        PROJECT, TOPIC))
+
+
 def test_list(client, topic, capsys):
     @eventually_consistent.call
     def _():
@@ -127,3 +135,5 @@ def test_publish_with_futures(topic, capsys):
 
     out, _ = capsys.readouterr()
     assert 'Published' in out
+
+    _to_delete()

pubsub/cloud-client/quickstart/sub_test.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import mock
 import os
 import pytest
 
@@ -68,31 +69,31 @@ def _publish_messages(topic_path):
     publish_future.result()
 
 
-def _sub_timeout(project_id, subscription_name):
-    # This is an exactly copy of `sub.py` except
-    # StreamingPullFuture.result() will time out after 10s.
-    client = pubsub_v1.SubscriberClient()
-    subscription_path = client.subscription_path(
-        project_id, subscription_name)
+def test_sub(monkeypatch, topic_path, subscription_path, capsys):
 
-    def callback(message):
-        print('Received message {} of message ID {}\n'.format(
-            message, message.message_id))
-        message.ack()
-        print('Acknowledged message {}\n'.format(message.message_id))
+    real_client = pubsub_v1.SubscriberClient()
+    mock_client = mock.Mock(spec=pubsub_v1.SubscriberClient,
+                            wraps=real_client)
 
-    streaming_pull_future = client.subscribe(
-        subscription_path, callback=callback)
-    print('Listening for messages on {}..\n'.format(subscription_path))
+    # Attributes on mock_client_constructor uses the corresponding
+    # attributes on pubsub_v1.SubscriberClient.
+    mock_client_constructor = mock.create_autospec(pubsub_v1.SubscriberClient)
+    mock_client_constructor.return_value = mock_client
 
-    try:
-        streaming_pull_future.result(timeout=10)
-    except:  # noqa
-        streaming_pull_future.cancel()
+    monkeypatch.setattr(pubsub_v1, 'SubscriberClient', mock_client_constructor)
 
+    def mock_subscribe(subscription_path, callback=None):
+        real_future = real_client.subscribe(subscription_path,
+                                            callback=callback)
+        mock_future = mock.Mock(spec=real_future, wraps=real_future)
 
-def test_sub(monkeypatch, topic_path, subscription_path, capsys):
-    monkeypatch.setattr(sub, 'sub', _sub_timeout)
+        def mock_result():
+            return real_future.result(timeout=10)
+
+        mock_future.result.side_effect = mock_result
+        return mock_future
+
+    mock_client.subscribe.side_effect = mock_subscribe
 
     _publish_messages(topic_path)
 

pubsub/cloud-client/quickstart_test.py

@@ -15,14 +15,16 @@
 # limitations under the License.
 
 import os
+import uuid
 
 from google.cloud import pubsub_v1
 import pytest
 import quickstart
 
+UUID = uuid.uuid4().hex
 PROJECT = os.environ['GCLOUD_PROJECT']
-TOPIC = 'end-to-end-test-topic'
-SUBSCRIPTION = 'end-to-end-test-topic-sub'
+TOPIC = 'end-to-end-test-topic-' + UUID
+SUBSCRIPTION = 'end-to-end-test-topic-sub-' + UUID
 N = 10
 
 
@@ -61,6 +63,15 @@ def subscription(subscriber_client, topic):
     yield SUBSCRIPTION
 
 
+def _to_delete():
+    publisher_client = pubsub_v1.PublisherClient()
+    subscriber_client = pubsub_v1.SubscriberClient()
+    publisher_client.delete_topic(
+        'projects/{}/topics/{}'.format(PROJECT, TOPIC))
+    subscriber_client.delete_subscription(
+        'projects/{}/subscriptions/{}'.format(PROJECT, SUBSCRIPTION))
+
+
 def test_end_to_end(topic, subscription, capsys):
 
     quickstart.end_to_end(PROJECT, topic, subscription, N)
@@ -69,3 +80,6 @@ def test_end_to_end(topic, subscription, capsys):
     assert "Received all messages" in out
     assert "Publish time lapsed" in out
     assert "Subscribe time lapsed" in out
+
+    # Clean up resources.
+    _to_delete()

pubsub/cloud-client/subscriber_test.py

@@ -14,6 +14,7 @@
 
 import os
 import time
+import uuid
 
 from gcp_devrel.testing import eventually_consistent
 from google.cloud import pubsub_v1
@@ -22,11 +23,12 @@
 
 import subscriber
 
+UUID = uuid.uuid4().hex
 PROJECT = os.environ['GCLOUD_PROJECT']
-TOPIC = 'subscription-test-topic'
-SUBSCRIPTION_ONE = 'subscription-test-subscription-one'
-SUBSCRIPTION_TWO = 'subscription-test-subscription-two'
-SUBSCRIPTION_THREE = 'subscription-test-subscription-three'
+TOPIC = 'subscription-test-topic-' + UUID
+SUBSCRIPTION_ONE = 'subscription-test-subscription-one-' + UUID
+SUBSCRIPTION_TWO = 'subscription-test-subscription-two-' + UUID
+SUBSCRIPTION_THREE = 'subscription-test-subscription-three-' + UUID
 ENDPOINT = 'https://{}.appspot.com/push'.format(PROJECT)
 NEW_ENDPOINT = 'https://{}.appspot.com/push2'.format(PROJECT)
 
@@ -185,6 +187,20 @@ def new_sleep(period):
     return mock.patch('time.sleep', new=new_sleep)
 
 
+def _to_delete():
+    publisher_client = pubsub_v1.PublisherClient()
+    subscriber_client = pubsub_v1.SubscriberClient()
+    resources = [TOPIC, SUBSCRIPTION_TWO, SUBSCRIPTION_THREE]
+
+    for item in resources:
+        if 'subscription-test-topic' in item:
+            publisher_client.delete_topic(
+                'projects/{}/topics/{}'.format(PROJECT, item))
+        if 'subscription-test-subscription' in item:
+            subscriber_client.delete_subscription(
+                'projects/{}/subscriptions/{}'.format(PROJECT, item))
+
+
 def test_receive(publisher_client, topic, subscription_two, capsys):
     _publish_messages(publisher_client, topic)
 
@@ -195,7 +211,7 @@ def test_receive(publisher_client, topic, subscription_two, capsys):
     out, _ = capsys.readouterr()
     assert 'Listening' in out
     assert subscription_two in out
-    assert 'Message 1' in out
+    assert 'Message' in out
 
 
 def test_receive_with_custom_attributes(
@@ -227,7 +243,7 @@ def test_receive_with_flow_control(
     out, _ = capsys.readouterr()
     assert 'Listening' in out
     assert subscription_two in out
-    assert 'Message 1' in out
+    assert 'Message' in out
 
 
 def test_receive_synchronously(
@@ -249,3 +265,6 @@ def test_receive_synchronously_with_lease(
 
     out, _ = capsys.readouterr()
     assert 'Done.' in out
+
+    # Clean up resources after all the tests.
+    _to_delete()

pubsub/streaming-analytics/PubSubToGCS_test.py

@@ -26,8 +26,8 @@
 
 PROJECT = os.environ['GCLOUD_PROJECT']
 BUCKET = os.environ['CLOUD_STORAGE_BUCKET']
-TOPIC = 'test-topic'
-UUID = uuid.uuid4().hex
+UUID = uuid.uuid1().hex
+TOPIC = 'test-topic-' + UUID
 
 
 @pytest.fixture
@@ -95,6 +95,10 @@ def test_run(publisher_client, topic_path):
     files = gcs_client.list_prefix('gs://{}/pubsub/{}'.format(BUCKET, UUID))
     assert len(files) > 0
 
-    # Clean up. Delete topic. Delete files.
+    # Clean up. Delete subscription. Delete topic. Delete GCS files.
+    subscriber_client = pubsub_v1.SubscriberClient()
+    subscriptions = publisher_client.list_topic_subscriptions(topic_path)
+    for subscription in subscriptions:
+        subscriber_client.delete_subscription(subscription)
     publisher_client.delete_topic(topic_path)
     gcs_client.delete_batch(list(files))