tests: mitigate flaky snippets tests (#432)

* test: add additional debug info to snippet test

* Reduce the sleep timeout in message callback

* Auto retry flaky delete schema test

* More flaky tests whack-a-mole

samples/snippets/schema_test.py

@@ -17,6 +17,7 @@
 import os
 import uuid
 
+from flaky import flaky
 from google.api_core.exceptions import NotFound
 from google.cloud.pubsub import PublisherClient, SchemaServiceClient, SubscriberClient
 from google.pubsub_v1.types import Encoding
@@ -251,6 +252,7 @@ def test_subscribe_with_proto_schema(
     assert "Received a binary-encoded message" in out
 
 
+@flaky(max_runs=3, min_passes=1)
 def test_delete_schema(proto_schema, capsys):
     schema.delete_schema(PROJECT_ID, PROTO_SCHEMA_ID)
     out, _ = capsys.readouterr()

samples/snippets/subscriber.py

@@ -499,7 +499,7 @@ def receive_messages_with_blocking_shutdown(project_id, subscription_id, timeout
 
     def callback(message):
         print(f"Received {message.data}.")
-        time.sleep(timeout + 5.0)  # Pocess longer than streaming pull future timeout.
+        time.sleep(timeout + 3.0)  # Pocess longer than streaming pull future timeout.
         message.ack()
         print(f"Done processing the message {message.data}.")
 

samples/snippets/subscriber_test.py

@@ -425,21 +425,26 @@ def test_receive_with_blocking_shutdown(
         if re.search(r".*done waiting.*stream shutdown.*", line, flags=re.IGNORECASE)
     ]
 
-    assert "Listening" in out
-    assert subscription_async in out
+    try:
+        assert "Listening" in out
+        assert subscription_async in out
 
-    assert len(stream_canceled_lines) == 1
-    assert len(shutdown_done_waiting_lines) == 1
-    assert len(msg_received_lines) == 3
-    assert len(msg_done_lines) == 3
+        assert len(stream_canceled_lines) == 1
+        assert len(shutdown_done_waiting_lines) == 1
+        assert len(msg_received_lines) == 3
+        assert len(msg_done_lines) == 3
 
-    # The stream should have been canceled *after* receiving messages, but before
-    # message processing was done.
-    assert msg_received_lines[-1] < stream_canceled_lines[0] < msg_done_lines[0]
+        # The stream should have been canceled *after* receiving messages, but before
+        # message processing was done.
+        assert msg_received_lines[-1] < stream_canceled_lines[0] < msg_done_lines[0]
 
-    # Yet, waiting on the stream shutdown should have completed *after* the processing
-    # of received messages has ended.
-    assert msg_done_lines[-1] < shutdown_done_waiting_lines[0]
+        # Yet, waiting on the stream shutdown should have completed *after*
+        # the processing of received messages has ended.
+        assert msg_done_lines[-1] < shutdown_done_waiting_lines[0]
+    except AssertionError:  # pragma: NO COVER
+        from pprint import pprint
+        pprint(out_lines)  # To make possible flakiness debugging easier.
+        raise
 
 
 def test_listen_for_errors(publisher_client, topic, subscription_async, capsys):
@@ -464,6 +469,7 @@ def test_receive_synchronously(publisher_client, topic, subscription_sync, capsy
     assert f"{subscription_sync}" in out
 
 
+@flaky(max_runs=3, min_passes=1)
 def test_receive_synchronously_with_lease(
     publisher_client, topic, subscription_sync, capsys
 ):