Fix failing unit test, handle queue can not be found

Author: AlanCoding

dispatcher/brokers/__init__.py

@@ -1,4 +1,5 @@
 import importlib
+from typing import Optional
 from types import ModuleType
 
 from dispatcher.brokers.base import BaseBroker
@@ -19,8 +20,16 @@ def get_sync_broker(broker_name, broker_config) -> BaseBroker:
     return broker_module.SyncBroker(**broker_config)
 
 
-def get_sync_publisher_from_settings() -> BaseBroker:
-    publish_broker = settings.publish['default_broker']
+def get_sync_publisher_from_settings(publish_broker: Optional[str] = None) -> BaseBroker:
+    if publish_broker:
+        pass
+    elif len(settings.brokers) == 1:
+        publish_broker = list(settings.brokers.keys())[0]
+    elif 'default_broker' in settings.publish:
+        publish_broker = settings.publish['default_broker']
+    else:
+        raise RuntimeError(f'Could not determine which broker to publish with between options {list(settings.brokers.keys())}')
+
     return get_sync_broker(publish_broker, settings.brokers[publish_broker])
 
 

dispatcher/brokers/pg_notify.py

@@ -22,7 +22,7 @@ class PGNotifyBase(BaseBroker):
     def __init__(
         self,
         channels: Iterable[str] = ('dispatcher_default',),
-        default_publish_channel: str = 'dispatcher_default',
+        default_publish_channel: Optional[str] = None,
     ) -> None:
         self.channels = channels
         self.default_publish_channel = default_publish_channel
@@ -167,7 +167,9 @@ def create_connection(config) -> psycopg.Connection:
 
     def publish_message(self, channel: Optional[str] = None, message: str = '') -> None:
         connection = self.get_connection()
-        if not channel:
+        if channel is None:
+            if self.default_publish_channel is None:
+                raise ValueError('Could not determine a channel to use publish to from settings or PGNotify config')
             channel = self.default_publish_channel
 
         with connection.cursor() as cur:

tests/conftest.py

@@ -12,6 +12,7 @@
 from dispatcher.control import Control
 
 from dispatcher.brokers.pg_notify import SyncBroker, AsyncBroker
+from dispatcher.registry import DispatcherMethodRegistry
 
 
 # List of channels to listen on
@@ -104,3 +105,9 @@ async def pg_control() -> AsyncIterator[Control]:
 async def psycopg_conn():
     async with aconnection_for_test() as conn:
         yield conn
+
+
+@pytest.fixture
+def registry() -> DispatcherMethodRegistry:
+    "Return a fresh registry, separate from the global one, for testing"
+    return DispatcherMethodRegistry()

tests/integration/publish/test_registry.py

@@ -0,0 +1,28 @@
+from unittest import mock
+
+import pytest
+
+from dispatcher.publish import task
+from dispatcher.config import temporary_settings
+
+
+def test_apply_async_with_no_queue(registry, conn_config):
+    @task(registry=registry)
+    def test_method():
+        return
+
+    dmethod = registry.get_from_callable(test_method)
+
+    # These settings do not specify a default channel, that is the main point
+    with temporary_settings({'brokers': {'pg_notify': {'config': conn_config}}}):
+
+        # Can not run a method if we do not have a queue
+        with pytest.raises(ValueError):
+            dmethod.apply_async()
+
+        # But providing a queue at time of submission works
+        with mock.patch('dispatcher.brokers.pg_notify.SyncBroker.publish_message') as mock_publish_method:
+            dmethod.apply_async(queue='fooqueue')
+        mock_publish_method.assert_called_once_with(channel='fooqueue', message=mock.ANY)
+
+    mock_publish_method.assert_called_once()

tests/unit/conftest.py

@@ -1,9 +0,0 @@
-import pytest
-
-from dispatcher.registry import DispatcherMethodRegistry
-
-
-@pytest.fixture
-def registry() -> DispatcherMethodRegistry:
-    "Return a fresh registry, separate from the global one, for testing"
-    return DispatcherMethodRegistry()

tests/unit/test_publish.py

@@ -1,7 +1,6 @@
 from unittest import mock
 
 from dispatcher.publish import task
-from dispatcher.utils import serialize_task
 
 import pytest
 
@@ -76,21 +75,3 @@ def run(self):
     TestMethod.delay()
 
     mock_apply_async.assert_called_once_with((), {})
-
-
-def test_apply_async_with_no_queue(registry):
-    @task(registry=registry)
-    def test_method():
-        return
-
-    dmethod = registry.get_from_callable(test_method)
-
-    # Can not run a method if we do not have a queue
-    with pytest.raises(ValueError):
-        dmethod.apply_async()
-
-    # But providing a queue at time of submission works
-    with mock.patch('dispatcher.brokers.pg_notify.publish_message') as mock_publish_method:
-        dmethod.apply_async(queue='fooqueue')
-
-    mock_publish_method.assert_called_once()