Repository: mercadona/rele
Pull request #274 (2024-02-28 13:38:39)

Added a setting to provide client options for Pub/Sub clients
### :tophat: What?

Added `CLIENT_OPTIONS` settings to provide configurations for the Pub/Sub [publisher](https://cloud.google.com/python/docs/reference/pubsub/latest/google.cloud.pubsub_v1.publisher.client.Client#google_cloud_pubsub_v1_publisher_client_Client_api_endpoint) and [subscriber](https://cloud.google.com/python/docs/reference/pubsub/latest/google.cloud.pubsub_v1.subscriber.client.Client#google_cloud_pubsub_v1_subscriber_client_Client) clients.

```
RELE = {
    # other settings
    "CLIENT_OPTIONS": {"api_endpoint": "custom-api.interconnect.example.com"},
}
```
### :thinking: Why?
We want to ability to update the global endpoint for publisher and subscriber clients. 

### :link: Related issue



Patch:
diff --git a/rele/__init__.py b/rele/__init__.py
index 5e43993..3d82b7d 100644
--- a/rele/__init__.py
+++ b/rele/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "1.14.0"
+__version__ = "1.15.0b0"
 
 try:
     import django
diff --git a/rele/client.py b/rele/client.py
index da66622..8601664 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -48,6 +48,7 @@ class Subscriber:
         gc_project_id,
         credentials,
         message_storage_policy,
+        client_options,
         default_ack_deadline=None,
         default_retry_policy=None,
     ):
@@ -55,7 +56,9 @@ class Subscriber:
         self._ack_deadline = default_ack_deadline or DEFAULT_ACK_DEADLINE
         self.credentials = credentials if not USE_EMULATOR else None
         self._message_storage_policy = message_storage_policy
-        self._client = pubsub_v1.SubscriberClient(credentials=credentials)
+        self._client = pubsub_v1.SubscriberClient(
+            credentials=credentials, client_options=client_options
+        )
         self._retry_policy = default_retry_policy
 
     def update_or_create_subscription(self, subscription):
@@ -182,7 +185,15 @@ class Publisher:
     :param blocking: boolean, default None falls back to :ref:`settings_publisher_blocking`
     """
 
-    def __init__(self, gc_project_id, credentials, encoder, timeout, blocking=None):
+    def __init__(
+        self,
+        gc_project_id,
+        credentials,
+        encoder,
+        timeout,
+        client_options,
+        blocking=None,
+    ):
         self._gc_project_id = gc_project_id
         self._timeout = timeout
         self._blocking = blocking
@@ -190,7 +201,9 @@ class Publisher:
         if USE_EMULATOR:
             self._client = pubsub_v1.PublisherClient()
         else:
-            self._client = pubsub_v1.PublisherClient(credentials=credentials)
+            self._client = pubsub_v1.PublisherClient(
+                credentials=credentials, client_options=client_options
+            )
 
     def publish(
         self, topic, data, blocking=None, timeout=None, raise_exception=True, **attrs
diff --git a/rele/config.py b/rele/config.py
index e2301ae..c3eaecd 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -43,6 +43,7 @@ class Config:
         self.filter_by = setting.get("FILTER_SUBS_BY")
         self._credentials = None
         self.retry_policy = setting.get("DEFAULT_RETRY_POLICY")
+        self.client_options = setting.get("CLIENT_OPTIONS")
 
     @property
     def encoder(self):
diff --git a/rele/publishing.py b/rele/publishing.py
index f1aa85f..f251481 100644
--- a/rele/publishing.py
+++ b/rele/publishing.py
@@ -14,6 +14,7 @@ def init_global_publisher(config):
             encoder=config.encoder,
             timeout=config.publisher_timeout,
             blocking=config.publisher_blocking,
+            client_options=config.client_options,
         )
     return _publisher
 
diff --git a/rele/worker.py b/rele/worker.py
index cd8df73..38a102c 100644
--- a/rele/worker.py
+++ b/rele/worker.py
@@ -48,6 +48,7 @@ class Worker:
     def __init__(
         self,
         subscriptions,
+        client_options,
         gc_project_id=None,
         credentials=None,
         gc_storage_region=None,
@@ -59,6 +60,7 @@ class Worker:
             gc_project_id,
             credentials,
             gc_storage_region,
+            client_options,
             default_ack_deadline,
             default_retry_policy,
         )
@@ -228,6 +230,7 @@ def create_and_run(subs, config):
         print(f"Subscription: {sub}")
     worker = Worker(
         subs,
+        config.client_options,
         config.gc_project_id,
         config.credentials,
         config.gc_storage_region,


Test patch:
diff --git a/tests/commands/test_runrele.py b/tests/commands/test_runrele.py
index 70eaf15..357b2b3 100644
--- a/tests/commands/test_runrele.py
+++ b/tests/commands/test_runrele.py
@@ -18,7 +18,7 @@ class TestRunReleCommand:
         call_command("runrele")
 
         mock_worker.assert_called_with(
-            [], "rele-test", ANY, "europe-west1", 60, 2, None
+            [], None, "rele-test", ANY, "europe-west1", 60, 2, None
         )
         mock_worker.return_value.run_forever.assert_called_once_with()
 
@@ -35,6 +35,6 @@ class TestRunReleCommand:
             "be exhausted." in err
         )
         mock_worker.assert_called_with(
-            [], "rele-test", ANY, "europe-west1", 60, 2, None
+            [], None, "rele-test", ANY, "europe-west1", 60, 2, None
         )
         mock_worker.return_value.run_forever.assert_called_once_with()
diff --git a/tests/conftest.py b/tests/conftest.py
index 0c60257..71f5f58 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -29,6 +29,7 @@ def config(project_id):
             "GC_CREDENTIALS_PATH": "tests/dummy-pub-sub-credentials.json",
             "GC_STORAGE_REGION": "some-region",
             "MIDDLEWARE": ["rele.contrib.LoggingMiddleware"],
+            "CLIENT_OPTIONS": {"api_endpoint": "custom-api.interconnect.example.com"},
         }
     )
 
@@ -56,7 +57,11 @@ def mock_worker():
 @pytest.fixture
 def subscriber(project_id, config):
     return Subscriber(
-        config.gc_project_id, config.credentials, config.gc_storage_region, 60
+        config.gc_project_id,
+        config.credentials,
+        config.gc_storage_region,
+        config.client_options,
+        60,
     )
 
 
@@ -73,6 +78,7 @@ def publisher(config, mock_future):
         encoder=config.encoder,
         timeout=config.publisher_timeout,
         blocking=config.publisher_blocking,
+        client_options=config.client_options,
     )
     publisher._client = MagicMock(spec=PublisherClient)
     publisher._client.publish.return_value = mock_future
diff --git a/tests/test_publisher.py b/tests/test_publisher.py
index e1df7a1..bcac394 100644
--- a/tests/test_publisher.py
+++ b/tests/test_publisher.py
@@ -2,13 +2,31 @@ import concurrent
 import decimal
 import logging
 from concurrent.futures import TimeoutError
-from unittest.mock import ANY
+from unittest.mock import ANY, patch
 
 import pytest
 
+from rele import Publisher
+
 
 @pytest.mark.usefixtures("publisher", "time_mock")
 class TestPublisher:
+    @patch("rele.client.pubsub_v1.PublisherClient", autospec=True)
+    def test_initialises_with_correct_parameters(self, mock_publisher_client, config):
+        Publisher(
+            gc_project_id=config.gc_project_id,
+            credentials=config.credentials,
+            encoder=config.encoder,
+            timeout=config.publisher_timeout,
+            blocking=config.publisher_blocking,
+            client_options=config.client_options,
+        )
+
+        mock_publisher_client.assert_called_with(
+            credentials=ANY,
+            client_options={"api_endpoint": "custom-api.interconnect.example.com"},
+        )
+
     def test_returns_future_when_published_called(self, published_at, publisher):
         message = {"foo": "bar"}
         result = publisher.publish(
diff --git a/tests/test_publishing.py b/tests/test_publishing.py
index d5dabed..7e3aba6 100644
--- a/tests/test_publishing.py
+++ b/tests/test_publishing.py
@@ -1,8 +1,9 @@
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, Mock, patch
 
 import pytest
 
 from rele import Publisher, publishing
+from rele.config import Config
 from tests import settings
 
 
@@ -49,3 +50,21 @@ class TestInitGlobalPublisher:
         mock_publisher.return_value = MagicMock(spec=Publisher)
         publishing.publish(topic="order-cancelled", data=message, myattr="hello")
         assert id(publishing._publisher) == publisher_id
+
+    @patch("rele.publishing.Publisher", autospec=True)
+    def test_creates_publisher_with_api_endpoint_option(self, mock_publisher, config):
+        publishing._publisher = None
+        mock_publisher.return_value = MagicMock(spec=Publisher)
+        publishing.init_global_publisher(config)
+
+        message = {"foo": "bar"}
+        publishing.publish(topic="order-cancelled", data=message, myattr="hello")
+
+        mock_publisher.assert_called_with(
+            gc_project_id="rele-test",
+            credentials=ANY,
+            encoder=ANY,
+            timeout=3.0,
+            blocking=False,
+            client_options={"api_endpoint": "custom-api.interconnect.example.com"},
+        )
diff --git a/tests/test_subscriber.py b/tests/test_subscriber.py
index 91fd72b..a65ab37 100644
--- a/tests/test_subscriber.py
+++ b/tests/test_subscriber.py
@@ -1,4 +1,4 @@
-from unittest.mock import patch
+from unittest.mock import ANY, patch
 
 import pytest
 from google.api_core import exceptions
@@ -21,6 +21,23 @@ class TestSubscriber:
         ) as mock:
             yield mock
 
+    @patch("rele.client.pubsub_v1.SubscriberClient", autospec=True)
+    def test_creates_subscriber_client_with_client_options(
+        self, mock_subscriber_client, config
+    ):
+        Subscriber(
+            gc_project_id=config.gc_project_id,
+            credentials=config.credentials,
+            message_storage_policy=config.gc_storage_region,
+            client_options={"api_endpoint": "custom-api.interconnect.example.com"},
+            default_ack_deadline=60,
+        )
+
+        mock_subscriber_client.assert_called_with(
+            credentials=ANY,
+            client_options={"api_endpoint": "custom-api.interconnect.example.com"},
+        )
+
     @patch.object(SubscriberClient, "create_subscription")
     @patch.object(SubscriberClient, "update_subscription")
     def test_creates_subscription_with_default_ack_deadline_when_none_provided(
@@ -197,6 +214,7 @@ class TestSubscriber:
             config_with_retry_policy.gc_project_id,
             config_with_retry_policy.credentials,
             config_with_retry_policy.gc_storage_region,
+            config_with_retry_policy.client_options,
             60,
             config_with_retry_policy.retry_policy,
         )
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 1cc95cf..fa23d98 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -26,6 +26,7 @@ def worker(config):
     subscriptions = (sub_stub,)
     return Worker(
         subscriptions,
+        config.client_options,
         config.gc_project_id,
         config.credentials,
         config.gc_storage_region,
@@ -115,6 +116,7 @@ class TestWorker:
         custom_ack_deadline = 234
         worker = Worker(
             subscriptions,
+            config.client_options,
             config.gc_project_id,
             config.gc_storage_region,
             config.credentials,
@@ -228,6 +230,11 @@ class TestCreateAndRun:
         with patch("rele.worker.Worker", autospec=True) as p:
             yield p
 
+    @pytest.fixture
+    def mock_subscriber(self):
+        with patch("rele.worker.Subscriber", autospec=True) as p:
+            yield p
+
     def test_waits_forever_when_called_with_config_and_subs(
         self, config_with_retry_policy, mock_worker
     ):
@@ -236,6 +243,7 @@ class TestCreateAndRun:
 
         mock_worker.assert_called_with(
             subscriptions,
+            None,
             "rele-test",
             ANY,
             "some-region",
@@ -244,3 +252,16 @@ class TestCreateAndRun:
             RetryPolicy(5, 30),
         )
         mock_worker.return_value.run_forever.assert_called_once_with()
+
+    def test_creates_subscriber_with_correct_arguments(self, mock_subscriber, config):
+        subscriptions = (sub_stub,)
+        create_and_run(subscriptions, config)
+
+        mock_subscriber.assert_called_with(
+            "rele-test",
+            ANY,
+            "some-region",
+            {"api_endpoint": "custom-api.interconnect.example.com"},
+            60,
+            None,
+        )