Repository: mercadona/rele
Pull request #207 (2021-09-08 15:09:52)

Add filter expressions to subscriptions
### :tophat: What?

Add filter expressions to subscriptions. 

Filter expressions are applied by pubsub before sending the message to a specific subscriber. It's use is defined [here](https://cloud.google.com/pubsub/docs/filtering#filtering_syntax) and it prevents the messages to be sent and filtered by the workers.

### :thinking: Why?

To give support to this pubsub feature.

### :link: Related issue

N/A



Patch:
diff --git a/.gitignore b/.gitignore
index 1e208b4..5653318 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,6 +10,7 @@
 /dist/
 /*.egg-info/
 /env/
+/venv/
 MANIFEST
 coverage.*
 docs/_build
diff --git a/AUTHORS.md b/AUTHORS.md
index d93234c..26a1bde 100644
--- a/AUTHORS.md
+++ b/AUTHORS.md
@@ -13,3 +13,5 @@ Thank you to all the contributors to Relé!
 * Craig Mulligan (@hobochild)
 * Daniel Demmel (@daaain)
 * Luis Garcia Cuesta (@luisgc93)
+* Chirag Jain (@chirgjin-les)
+* Jordi Chulia (@jorchube)
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 6788948..8308385 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,6 +1,25 @@
 Changelog
 =========
 
+1.4.1 (2022-04-19)
+-------------------
+* [Modified] Fixed bug in the post-publish-failure VerboseLoggingMiddleware hook. (#220)
+
+1.4.0 (2022-04-13)
+-------------------
+* [Added] Added a VerboseLoggingMiddleware that does not truncate mesage payload. (#218)
+
+1.3.0 (2022-04-04)
+-------------------
+* GC Project Id & Windows support (#215)
+
+1.2.0 (2021-12-10)
+-------------------
+* [CHANGED] TimeotError from publisher (#212)
+* Added filter_subs_by setting in documentation (#208)
+* Automatic topic creation (#206)
+* Log post publish success (#204)
+
 1.1.1 (2021-6-28)
 -------------------
 * Do not define default_app_config, it's deprecated (#199)
diff --git a/docs/conf.py b/docs/conf.py
index 87669b1..c34eab0 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -75,7 +75,7 @@ html_sidebars = {"**": ["sidebar.html", "navigation.html", "searchbox.html"]}
 
 # Setup function
 def setup(app):
-    app.add_stylesheet("style.css")
+    app.add_css_file("style.css")
 
 
 # -- Extension configuration -------------------------------------------------
diff --git a/docs/guides/filters.rst b/docs/guides/filters.rst
index f69df7a..8bbd82c 100644
--- a/docs/guides/filters.rst
+++ b/docs/guides/filters.rst
@@ -2,8 +2,9 @@ Filtering Messages
 ==================
 
 Filter can be used to execute a subscription with specific parameters.
-There's two types of filters, global or passing a filter_by parameter in the
-subscription.
+There are three types of filters, global, by passing a filter_by parameter in the
+subscription (this applies the filter locally) or by passing a backend_filter_by
+parameter in the subscription (this applies the filter on pubsub).
 
 
 `filter_by` parameter
@@ -26,6 +27,26 @@ is passed as parameter ``filter_by`` in the subscription.
         print(f'Received a photo of type {kwargs.get("type")}')
 
 
+`backend_filter_by` parameter
+_____________________________
+
+This filter is an expression that is applied to the subscription creation. This filter
+expression is applied by pubsub before passing the message to the subscriber. More info
+about filter expressions `here <https://cloud.google.com/pubsub/docs/filtering#filtering_syntax>`_.
+
+.. note::
+   Filter expressions are only applied on the subscription creation, they are not updated
+   if changed if you do not recreate the subscription on pubsub.
+
+.. code:: python
+
+    # This subscription is going to be called if in the kwargs
+    # has a key type with value landscape
+
+    @sub(topic='photo-updated', backend_filter_by='attributes.type = "landscape"')
+    def sub_process_landscape_photos(data, **kwargs):
+        print(f'Received a photo of type {kwargs.get("type")}')
+
 
 Global Filter
 _____________
diff --git a/docs/settings.rst b/docs/settings.rst
index b6e5c64..38f9047 100644
--- a/docs/settings.rst
+++ b/docs/settings.rst
@@ -28,9 +28,18 @@ Example::
         'APP_NAME': 'myappname',
         'ENCODER_PATH': 'rest_framework.utils.encoders.JSONEncoder',
         'ACK_DEADLINE': 120,
-        'PUBLISHER_TIMEOUT': 3.0
+        'PUBLISHER_TIMEOUT': 3.0,
+        'FILTER_SUBS_BY': boolean_function
     }
 
+``GC_PROJECT_ID``
+------------------
+
+**Optional**
+
+GCP project id to use. If this is not provided then it is inferred via either
+service account's project id or quota project id if using Application Default Credentials (ADC)
+
 
 ``GC_CREDENTIALS_PATH``
 -----------------------
@@ -143,3 +152,11 @@ Default behavior of the Google Cloud PubSub library is to use 10 threads per sub
 We thought this was too much for a default setting and have taken the liberty of
 reducing the thread count to 2. If you would like to maintain the default Google PubSub
 library behavior, please set this value to 10.
+
+``FILTER_SUBS_BY``
+----------------------------
+
+**Optional**
+
+Boolean function that applies a global filter on all subscriptions.
+For more information, please see `Filtering Messages section <https://mercadonarele.readthedocs.io/en/latest/guides/filters.html#global-filter>`.
diff --git a/rele/__init__.py b/rele/__init__.py
index 098fb90..99653e4 100644
--- a/rele/__init__.py
+++ b/rele/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "1.1.1"
+__version__ = "1.4.1"
 
 try:
     import django
diff --git a/rele/client.py b/rele/client.py
index 146e348..0b5fa92 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -2,12 +2,12 @@ import json
 import logging
 import os
 import time
+from concurrent.futures import TimeoutError
 from contextlib import suppress
 
 import google.auth
 from google.api_core import exceptions
 from google.cloud import pubsub_v1
-from google.cloud.pubsub_v1.exceptions import TimeoutError
 
 from rele.middleware import run_middleware_hook
 
@@ -43,7 +43,7 @@ class Subscriber:
         self.credentials = credentials if not USE_EMULATOR else None
         self._client = pubsub_v1.SubscriberClient(credentials=credentials)
 
-    def create_subscription(self, subscription, topic):
+    def create_subscription(self, subscription):
         """Handles creating the subscription when it does not exists.
 
         This makes it easier to deploy a worker and forget about the
@@ -55,13 +55,13 @@ class Subscriber:
         :param topic: str Topic name to subscribe
         """
         subscription_path = self._client.subscription_path(
-            self._gc_project_id, subscription
+            self._gc_project_id, subscription.name
         )
-        topic_path = self._client.topic_path(self._gc_project_id, topic)
+        topic_path = self._client.topic_path(self._gc_project_id, subscription.topic)
 
         with suppress(exceptions.AlreadyExists):
             try:
-                self._create_subscription(subscription_path, topic_path)
+                self._create_subscription(subscription_path, topic_path, subscription)
             except exceptions.NotFound:
                 logger.warning(
                     "Cannot subscribe to a topic that does not exist."
@@ -69,18 +69,23 @@ class Subscriber:
                 )
                 topic = self._create_topic(topic_path)
                 logger.info(f"Topic {topic.name} created.")
-                self._create_subscription(subscription_path, topic_path)
+                self._create_subscription(subscription_path, topic_path, subscription)
 
     def _create_topic(self, topic_path):
         publisher_client = pubsub_v1.PublisherClient(credentials=self.credentials)
         return publisher_client.create_topic(request={"name": topic_path})
 
-    def _create_subscription(self, name, topic):
-        self._client.create_subscription(
-            name=name,
-            topic=topic,
-            ack_deadline_seconds=self._ack_deadline,
-        )
+    def _create_subscription(self, subscription_path, topic_path, subscription):
+        request = {
+            "name": subscription_path,
+            "topic": topic_path,
+            "ack_deadline_seconds": self._ack_deadline,
+        }
+
+        if subscription.backend_filter_by:
+            request["filter"] = subscription.backend_filter_by
+
+        self._client.create_subscription(request=request)
 
     def consume(self, subscription_name, callback, scheduler):
         """Begin listening to topic from the SubscriberClient.
diff --git a/rele/config.py b/rele/config.py
index da3ae8e..4ad4648 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -22,6 +22,7 @@ class Config:
     """
 
     def __init__(self, setting):
+        self._gc_project_id = setting.get("GC_PROJECT_ID")
         self.gc_credentials_path = setting.get("GC_CREDENTIALS_PATH")
         self.app_name = setting.get("APP_NAME")
         self.sub_prefix = setting.get("SUB_PREFIX")
@@ -49,13 +50,17 @@ class Config:
                 self.gc_credentials_path
             )
         else:
-            credentials, __ = get_google_defaults()
+            credentials, project_id = get_google_defaults()
             self._credentials = credentials
+            if not self._gc_project_id:
+                self._gc_project_id = project_id
         return self._credentials
 
     @property
     def gc_project_id(self):
-        if self.credentials:
+        if self._gc_project_id:
+            return self._gc_project_id
+        elif self.credentials:
             return self.credentials.project_id
         else:
             return None
diff --git a/rele/contrib/__init__.py b/rele/contrib/__init__.py
index 8bea450..b47f00f 100644
--- a/rele/contrib/__init__.py
+++ b/rele/contrib/__init__.py
@@ -1,6 +1,7 @@
 from .flask_middleware import FlaskMiddleware  # noqa
 from .logging_middleware import LoggingMiddleware  # noqa
 from .unrecoverable_middleware import UnrecoverableMiddleWare  # noqa
+from .verbose_logging_middleware import VerboseLoggingMiddleware  # noqa
 
 try:
     from .django_db_middleware import DjangoDBMiddleware  # noqa
diff --git a/rele/contrib/verbose_logging_middleware.py b/rele/contrib/verbose_logging_middleware.py
new file mode 100644
index 0000000..e45f522
--- /dev/null
+++ b/rele/contrib/verbose_logging_middleware.py
@@ -0,0 +1,46 @@
+import json
+
+from rele.contrib.logging_middleware import LoggingMiddleware
+
+
+class VerboseLoggingMiddleware(LoggingMiddleware):
+    def post_process_message_failure(
+        self, subscription, exception, start_time, message
+    ):
+        super().post_process_message_failure(
+            subscription, exception, start_time, _VerboseMessage(message)
+        )
+
+
+class _VerboseMessage:
+    def __init__(self, message):
+        self._message = message
+        self.attributes = message.attributes
+
+    def __repr__(self):
+        _MESSAGE_REPR = """\
+Message {{
+  data: {!r}
+  ordering_key: {!r}
+  attributes: {}
+}}"""
+
+        data = self._message._message.data
+        attrs = self._message_attrs_repr()
+        ordering_key = str(self._message.ordering_key)
+
+        return _MESSAGE_REPR.format(data, ordering_key, attrs)
+
+    def _message_attrs_repr(self):
+        message_attrs = json.dumps(
+            dict(self.attributes), indent=2, separators=(",", ": "), sort_keys=True
+        )
+
+        indented = []
+        for line in message_attrs.split("\n"):
+            indented.append("  " + line)
+
+        message_attrs = "\n".join(indented)
+        message_attrs = message_attrs.lstrip()
+
+        return message_attrs
diff --git a/rele/subscription.py b/rele/subscription.py
index 5325791..7080d45 100644
--- a/rele/subscription.py
+++ b/rele/subscription.py
@@ -34,12 +34,15 @@ class Subscription:
 
     """
 
-    def __init__(self, func, topic, prefix="", suffix="", filter_by=None):
+    def __init__(
+        self, func, topic, prefix="", suffix="", filter_by=None, backend_filter_by=None
+    ):
         self._func = func
         self.topic = topic
         self._prefix = prefix
         self._suffix = suffix
         self._filters = self._init_filters(filter_by)
+        self.backend_filter_by = backend_filter_by
 
     def _init_filters(self, filter_by):
         if filter_by and not (
@@ -142,7 +145,7 @@ class Callback:
             run_middleware_hook("post_process_message")
 
 
-def sub(topic, prefix=None, suffix=None, filter_by=None):
+def sub(topic, prefix=None, suffix=None, filter_by=None, backend_filter_by=None):
     """Decorator function that makes declaring a PubSub Subscription simple.
 
     The Subscriber returned will automatically create and name
diff --git a/rele/worker.py b/rele/worker.py
index 6ad6901..5536b55 100644
--- a/rele/worker.py
+++ b/rele/worker.py
@@ -42,7 +42,7 @@ class Worker:
         re-created. Therefore, it is idempotent.
         """
         for subscription in self._subscriptions:
-            self._subscriber.create_subscription(subscription.name, subscription.topic)
+            self._subscriber.create_subscription(subscription)
 
     def start(self):
         """Begin consuming all subscriptions.
@@ -111,6 +111,18 @@ class Worker:
             time.sleep(sleep_interval)
 
 
+def _get_stop_signal():
+    """
+    Get stop signal for worker.
+    Returns `SIGBREAK` on windows because `SIGSTP` doesn't exist on it
+    """
+    if sys.platform.startswith("win"):
+        # SIGSTP doesn't exist on windows, so we use SIGBREAK instead
+        return signal.SIGBREAK
+
+    return signal.SIGTSTP
+
+
 def create_and_run(subs, config):
     """
     Create and run a worker from a list of Subscription objects and a config
@@ -134,7 +146,9 @@ def create_and_run(subs, config):
         config.threads_per_subscription,
     )
 
-    signal.signal(signal.SIGINT, signal.SIG_IGN)
+    # to allow killing runrele worker via ctrl+c
+    signal.signal(signal.SIGINT, worker.stop)
     signal.signal(signal.SIGTERM, worker.stop)
-    signal.signal(signal.SIGTSTP, worker.stop)
+    signal.signal(_get_stop_signal(), worker.stop)
+
     worker.run_forever()


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 120493d..62e9ca3 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -6,6 +6,7 @@ from unittest.mock import MagicMock, patch
 import pytest
 from google.cloud.pubsub_v1 import PublisherClient
 from google.cloud.pubsub_v1.exceptions import TimeoutError
+from google.protobuf import timestamp_pb2
 
 from rele import Publisher
 from rele.client import Subscriber
@@ -94,3 +95,10 @@ def mock_post_publish_failure():
         "rele.contrib.logging_middleware.LoggingMiddleware.post_publish_failure"
     ) as mock:
         yield mock
+
+
+@pytest.fixture
+def publish_time():
+    timestamp = timestamp_pb2.Timestamp()
+    timestamp.GetCurrentTime()
+    return timestamp
diff --git a/tests/contrib/test_verbose_logging_middleware.py b/tests/contrib/test_verbose_logging_middleware.py
new file mode 100644
index 0000000..92e0ec4
--- /dev/null
+++ b/tests/contrib/test_verbose_logging_middleware.py
@@ -0,0 +1,97 @@
+import queue
+from unittest.mock import MagicMock
+
+import pytest
+from google.cloud import pubsub_v1
+from tests.subs import sub_stub
+
+from rele.contrib.logging_middleware import LoggingMiddleware
+from rele.contrib.verbose_logging_middleware import VerboseLoggingMiddleware
+
+
+@pytest.fixture
+def long_message_wrapper(published_at, publish_time):
+    long_string = "A" * 100
+    rele_message = pubsub_v1.types.PubsubMessage(
+        data=long_string.encode("utf-8"),
+        attributes={"lang": "es", "published_at": str(published_at)},
+        message_id="1",
+        publish_time=publish_time,
+    )
+
+    message = pubsub_v1.subscriber.message.Message(
+        rele_message._pb,
+        "ack-id",
+        delivery_attempt=1,
+        request_queue=queue.Queue(),
+    )
+    message.ack = MagicMock(autospec=True)
+    return message
+
+
+@pytest.fixture
+def message_wrapper(published_at, publish_time):
+    rele_message = pubsub_v1.types.PubsubMessage(
+        data="ABCDE".encode("utf-8"),
+        attributes={"lang": "es", "published_at": str(published_at)},
+        message_id="1",
+        publish_time=publish_time,
+    )
+
+    message = pubsub_v1.subscriber.message.Message(
+        rele_message._pb,
+        "ack-id",
+        delivery_attempt=1,
+        request_queue=queue.Queue(),
+    )
+    message.ack = MagicMock(autospec=True)
+    return message
+
+
+@pytest.fixture
+def expected_message_log():
+    return 'Message {\n  data: b\'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA\'\n  ordering_key: \'\'\n  attributes: {\n    "lang": "es",\n    "published_at": "1560244246.863829"\n  }\n}'
+
+
+class TestVerboseLoggingMiddleware:
+    @pytest.fixture
+    def verbose_logging_middleware(self, config):
+        verbose_logging_middleware = VerboseLoggingMiddleware()
+        verbose_logging_middleware.setup(config)
+        return verbose_logging_middleware
+
+    @pytest.fixture
+    def logging_middleware(self, config):
+        logging_middleware = LoggingMiddleware()
+        logging_middleware.setup(config)
+        return logging_middleware
+
+    def test_message_payload_log_is_not_truncated_on_post_process_failure(
+        self,
+        verbose_logging_middleware,
+        caplog,
+        long_message_wrapper,
+        expected_message_log,
+    ):
+        verbose_logging_middleware.post_process_message_failure(
+            sub_stub, RuntimeError("💩"), 1, long_message_wrapper
+        )
+
+        message_log = caplog.records[0].subscription_message.__repr__()
+
+        assert message_log == expected_message_log
+
+    def test_post_process_failure_message_payload_format_matches_logging_middleware_format(
+        self, verbose_logging_middleware, logging_middleware, caplog, message_wrapper
+    ):
+        verbose_logging_middleware.post_process_message_failure(
+            sub_stub, RuntimeError("💩"), 1, message_wrapper
+        )
+        logging_middleware.post_process_message_failure(
+            sub_stub, RuntimeError("💩"), 1, message_wrapper
+        )
+
+        verbose_message_log = caplog.records[0].subscription_message.__repr__()
+        message_log = caplog.records[1].subscription_message.__repr__()
+
+        assert verbose_message_log == message_log
diff --git a/tests/test_client.py b/tests/test_client.py
index c88969c..249a50f 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,12 +1,14 @@
 import concurrent
 import decimal
 import logging
+from concurrent.futures import TimeoutError
 from unittest.mock import ANY, patch
 
 import pytest
 from google.api_core import exceptions
 from google.cloud.pubsub_v1 import PublisherClient, SubscriberClient
-from google.cloud.pubsub_v1.exceptions import TimeoutError
+
+from rele.subscription import Subscription
 
 
 @pytest.mark.usefixtures("publisher", "time_mock")
@@ -149,14 +151,20 @@ class TestSubscriber:
         project_id,
         subscriber,
     ):
-        expected_subscription = f"projects/{project_id}/subscriptions/" f"test-topic"
+        expected_subscription = (
+            f"projects/{project_id}/subscriptions/" f"{project_id}-test-topic"
+        )
         expected_topic = f"projects/{project_id}/topics/" f"{project_id}-test-topic"
 
-        subscriber.create_subscription("test-topic", f"{project_id}-test-topic")
+        subscriber.create_subscription(
+            Subscription(None, topic=f"{project_id}-test-topic")
+        )
         _mocked_client.assert_called_once_with(
-            ack_deadline_seconds=60,
-            name=expected_subscription,
-            topic=expected_topic,
+            request={
+                "ack_deadline_seconds": 60,
+                "name": expected_subscription,
+                "topic": expected_topic,
+            }
         )
         assert subscriber._gc_project_id == "rele-test"
 
@@ -164,17 +172,47 @@ class TestSubscriber:
     def test_creates_subscription_with_custom_ack_deadline_when_provided(
         self, _mocked_client, project_id, subscriber
     ):
-        expected_subscription = f"projects/{project_id}/subscriptions/" f"test-topic"
+        expected_subscription = (
+            f"projects/{project_id}/subscriptions/" f"{project_id}-test-topic"
+        )
         expected_topic = f"projects/{project_id}/topics/" f"{project_id}-test-topic"
         subscriber._ack_deadline = 100
         subscriber.create_subscription(
-            subscription="test-topic", topic=f"{project_id}-test-topic"
+            Subscription(None, topic=f"{project_id}-test-topic")
+        )
+
+        _mocked_client.assert_called_once_with(
+            request={
+                "ack_deadline_seconds": 100,
+                "name": expected_subscription,
+                "topic": expected_topic,
+            }
+        )
+
+    @patch.object(SubscriberClient, "create_subscription")
+    def test_creates_subscription_with_backend_filter_by_when_provided(
+        self, _mocked_client, project_id, subscriber
+    ):
+        expected_subscription = (
+            f"projects/{project_id}/subscriptions/" f"{project_id}-test-topic"
+        )
+        expected_topic = f"projects/{project_id}/topics/" f"{project_id}-test-topic"
+        backend_filter_by = "attributes:domain"
+        subscriber.create_subscription(
+            Subscription(
+                None,
+                topic=f"{project_id}-test-topic",
+                backend_filter_by=backend_filter_by,
+            )
         )
 
         _mocked_client.assert_called_once_with(
-            ack_deadline_seconds=100,
-            name=expected_subscription,
-            topic=expected_topic,
+            request={
+                "ack_deadline_seconds": 60,
+                "name": expected_subscription,
+                "topic": expected_topic,
+                "filter": backend_filter_by,
+            }
         )
 
     @patch.object(
@@ -186,7 +224,7 @@ class TestSubscriber:
         self, _mocked_client, project_id, subscriber
     ):
         subscriber.create_subscription(
-            subscription="test-topic", topic=f"{project_id}-test-topic"
+            Subscription(None, topic=f"{project_id}-test-topic")
         )
 
         _mocked_client.assert_called()
@@ -199,11 +237,29 @@ class TestSubscriber:
     def test_creates_topic_when_subscription_topic_does_not_exist(
         self, _mocked_client, project_id, subscriber, mock_create_topic
     ):
+        expected_subscription = (
+            f"projects/{project_id}/subscriptions/" f"{project_id}-test-topic"
+        )
+        expected_topic = f"projects/{project_id}/topics/" f"{project_id}-test-topic"
+        backend_filter_by = "attributes:domain"
         subscriber.create_subscription(
-            subscription="test-topic", topic=f"{project_id}-test-topic"
+            Subscription(
+                None,
+                topic=f"{project_id}-test-topic",
+                backend_filter_by=backend_filter_by,
+            )
         )
 
         assert _mocked_client.call_count == 2
         mock_create_topic.assert_called_with(
             request={"name": f"projects/rele-test/topics/{project_id}-test-topic"}
         )
+
+        _mocked_client.assert_called_with(
+            request={
+                "ack_deadline_seconds": 60,
+                "name": expected_subscription,
+                "topic": expected_topic,
+                "filter": backend_filter_by,
+            }
+        )
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 51f0fa8..3a63b6f 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -53,24 +53,13 @@ class TestWorker:
         assert isinstance(scheduler, ThreadScheduler)
         assert isinstance(scheduler._executor, futures.ThreadPoolExecutor)
 
-    def test_setup_creates_subscription_when_topic_given(
-        self, mock_create_subscription, worker
-    ):
-        worker.setup()
-
-        topic = "some-cool-topic"
-        subscription = "rele-some-cool-topic"
-        mock_create_subscription.assert_called_once_with(subscription, topic)
-
     @patch.object(Worker, "_wait_forever")
     def test_run_sets_up_and_creates_subscriptions_when_called(
         self, mock_wait_forever, mock_consume, mock_create_subscription, worker
     ):
         worker.run_forever()
 
-        topic = "some-cool-topic"
-        subscription = "rele-some-cool-topic"
-        mock_create_subscription.assert_called_once_with(subscription, topic)
+        mock_create_subscription.assert_called_once_with(sub_stub)
         mock_consume.assert_called_once_with(
             subscription_name="rele-some-cool-topic",
             callback=ANY,