code review changes

pyproject.toml

@@ -91,7 +91,7 @@ filterwarnings = ["error", "ignore::DeprecationWarning"]
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests"
 markers = [
-    "stomp: marks tests that require the stomp" 
+    "stomp: marks tests that require a message bus configured for stomp" 
 ]
 asyncio_mode = "auto"
 

src/blueapi/client/event_bus.py

@@ -1,7 +1,7 @@
 from collections.abc import Callable
 
 from bluesky_stomp.messaging import MessageContext, MessagingTemplate
-from bluesky_stomp.models import MessageQueue
+from bluesky_stomp.models import MessageTopic
 
 from blueapi.core import DataEvent
 from blueapi.worker import ProgressEvent, WorkerEvent
@@ -34,7 +34,7 @@ def subscribe_to_all_events(
     ) -> None:
         try:
             self.app.subscribe(
-                MessageQueue(name="public.worker.event"),
+                MessageTopic(name="public.worker.event"),
                 on_event,
             )
         except Exception as err:

src/blueapi/service/interface.py

@@ -4,7 +4,7 @@
 from typing import Any
 
 from bluesky_stomp.messaging import MessagingTemplate
-from bluesky_stomp.models import Broker, DestinationBase, MessageQueue
+from bluesky_stomp.models import Broker, DestinationBase, MessageTopic
 
 from blueapi.config import ApplicationConfig
 from blueapi.core.context import BlueskyContext
@@ -59,7 +59,7 @@ def messaging_template() -> MessagingTemplate | None:
         )
 
         task_worker = worker()
-        event_topic = MessageQueue(name="public.worker.event")
+        event_topic = MessageTopic(name="public.worker.event")
 
         _publish_event_streams(
             {

src/blueapi/service/main.py

@@ -312,10 +312,9 @@ def set_state(
         and new_state in _ALLOWED_TRANSITIONS[current_state]
     ):
         if new_state == WorkerState.PAUSED:
-            if state_change_request.defer is not None:
-                runner.run(interface.pause_worker, state_change_request.defer)
-            else:
-                runner.run(interface.pause_worker, False)
+            if state_change_request.defer is None:
+                state_change_request.defer = False
+            runner.run(interface.pause_worker, state_change_request.defer)
         elif new_state == WorkerState.RUNNING:
             runner.run(interface.resume_worker)
         elif new_state in {WorkerState.ABORTING, WorkerState.STOPPING}:

tests/service/test_interface.py

@@ -1,3 +1,4 @@
+import logging
 import uuid
 from dataclasses import dataclass
 from unittest.mock import MagicMock, patch
@@ -273,3 +274,11 @@ def test_get_task_by_id(context_mock: MagicMock):
 def test_stomp_config():
     interface.set_config(ApplicationConfig(stomp=StompConfig()))
     assert interface.messaging_template() is not None
+
+
+@pytest.mark.stomp
+def test_stomp_connection_failure_with_invalid_config(caplog):
+    caplog.at_level(logging.exception)
+    interface.set_config(ApplicationConfig(stomp=StompConfig(host="Not Exists")))
+    assert interface.messaging_template() is None
+    assert "Failed to connect to message bus" in caplog.text

tests/test_cli.py

@@ -149,8 +149,8 @@ def test_valid_stomp_config_for_listener(runner: CliRunner):
         input="\n",
     )
     assert (
-        result.output
-        == 'Subscribing to all bluesky events from localhost:61613\nPress enter to exit{\n  "state": "IDLE",\n  "task_status": null,\n  "errors": [],\n  "warnings": []\n}\n'
+        "Subscribing to all bluesky events from localhost:61613\nPress enter to exit"
+        in result.output
     )
     assert result.exit_code == 0