Faff around with event buses

Author: tpoliaw

docs/reference/openapi.yaml

@@ -421,7 +421,7 @@ info:
     name: Apache 2.0
     url: https://www.apache.org/licenses/LICENSE-2.0.html
   title: BlueAPI Control
-  version: 1.1.3
+  version: 1.2.0
 openapi: 3.1.0
 paths:
   /config/oidc:

src/blueapi/client/client.py

@@ -1,12 +1,8 @@
 import time
 from concurrent.futures import Future
 
-from bluesky_stomp.messaging import MessageContext, StompClient
-from bluesky_stomp.models import Broker
-from observability_utils.tracing import (
-    get_tracer,
-    start_as_current_span,
-)
+from bluesky_stomp.messaging import MessageContext
+from observability_utils.tracing import get_tracer, start_as_current_span
 
 from blueapi.config import ApplicationConfig, MissingStompConfigurationError
 from blueapi.core.bluesky_types import DataEvent
@@ -38,15 +34,15 @@ class BlueapiClient:
     """Unified client for controlling blueapi"""
 
     _rest: BlueapiRestClient
-    _events: EventBusClient | None
+    _event_bus_client: EventBusClient | None
 
     def __init__(
         self,
         rest: BlueapiRestClient,
         events: EventBusClient | None = None,
     ):
         self._rest = rest
-        self._events = events
+        self._event_bus_client = events
 
     @classmethod
     def from_config(cls, config: ApplicationConfig) -> "BlueapiClient":
@@ -56,21 +52,8 @@ def from_config(cls, config: ApplicationConfig) -> "BlueapiClient":
         except Exception:
             ...  # Swallow exceptions
         rest = BlueapiRestClient(config.api, session_manager=session_manager)
-        stomp_config = config.stomp if config.stomp.enabled else rest.get_stomp_config()
-        if stomp_config and stomp_config.enabled:
-            assert stomp_config.url.host is not None, "Stomp URL missing host"
-            assert stomp_config.url.port is not None, "Stomp URL missing port"
-            client = StompClient.for_broker(
-                broker=Broker(
-                    host=stomp_config.url.host,
-                    port=stomp_config.url.port,
-                    auth=stomp_config.auth,
-                )
-            )
-            events = EventBusClient(client)
-            return cls(rest, events)
-        else:
-            return cls(rest)
+        event_bus = EventBusClient.from_stomp_config(config.stomp)
+        return cls(rest, event_bus)
 
     @start_as_current_span(TRACER)
     def get_plans(self) -> PlanResponse:
@@ -217,7 +200,7 @@ def run_task(
             of task execution.
         """
 
-        if self._events is None:
+        if (event_bus := self._event_bus()) is None:
             raise MissingStompConfigurationError(
                 "Stomp configuration required to run plans is missing or disabled"
             )
@@ -254,8 +237,8 @@ def inner_on_event(event: AnyEvent, ctx: MessageContext) -> None:
                     else:
                         complete.set_result(event)
 
-        with self._events:
-            self._events.subscribe_to_all_events(inner_on_event)
+        with event_bus:
+            event_bus.subscribe_to_all_events(inner_on_event)
             self.start_task(WorkerTask(task_id=task_id))
             return complete.result(timeout=timeout)
 
@@ -458,3 +441,10 @@ def get_python_env(
         """
 
         return self._rest.get_python_environment(name=name, source=source)
+
+    def _event_bus(self) -> EventBusClient | None:
+        if not self._event_bus_client:
+            if stomp_config := self._rest.get_stomp_config():
+                self._event_bus_client = EventBusClient.from_stomp_config(stomp_config)
+
+        return self._event_bus_client

src/blueapi/client/event_bus.py

@@ -1,8 +1,10 @@
 from collections.abc import Callable
+from typing import Self
 
-from bluesky_stomp.messaging import MessageContext, StompClient
+from bluesky_stomp.messaging import Broker, MessageContext, StompClient
 from bluesky_stomp.models import MessageTopic
 
+from blueapi.config import StompConfig
 from blueapi.core import DataEvent
 from blueapi.worker import ProgressEvent, WorkerEvent
 
@@ -45,3 +47,15 @@ def subscribe_to_all_events(
             raise BlueskyStreamingError(
                 "Unable to subscribe to messages from blueapi"
             ) from err
+
+    @classmethod
+    def from_stomp_config(cls, config: StompConfig) -> Self | None:
+        if config.enabled:
+            assert config.url.host is not None, "Stomp URL missing host"
+            assert config.url.port is not None, "Stomp URL missing port"
+            client = StompClient.for_broker(
+                broker=Broker(
+                    host=config.url.host, port=config.url.port, auth=config.auth
+                )
+            )
+            return cls(client)

src/blueapi/config.py

@@ -276,7 +276,7 @@ class ApplicationConfig(BlueapiBaseModel):
     """
 
     #: API version to publish in OpenAPI schema
-    REST_API_VERSION: ClassVar[str] = "1.1.3"
+    REST_API_VERSION: ClassVar[str] = "1.2.0"
 
     LICENSE_INFO: ClassVar[dict[str, str]] = {
         "name": "Apache 2.0",

tests/system_tests/test_blueapi_system.py

@@ -162,19 +162,21 @@ def expected_devices() -> DeviceResponse:
     )
 
 
-@pytest.fixture
-def blueapi_client_get_methods() -> list[str]:
-    # Get a list of methods that take only one argument (self)
-    # This will currently return
+def authenticated_get_methods() -> list[str]:
+    # Get a list of methods that take only one argument (self) and require
+    # authentication. This will currently return
     # ['get_plans', 'get_devices', 'get_state', 'get_all_tasks',
-    # 'get_active_task','get_environment','resume', 'stop','get_oidc_config']
+    # 'get_active_task','get_environment','resume', 'stop']
     return [
         method
         for method in BlueapiClient.__dict__
         if callable(getattr(BlueapiClient, method))
         and not method.startswith("__")
         and len(inspect.signature(getattr(BlueapiClient, method)).parameters) == 1
         and "self" in inspect.signature(getattr(BlueapiClient, method)).parameters
+        # oidc_config and stomp config can be accessed without auth
+        and method != "get_oidc_config"
+        and method != "_event_bus"
     ]
 
 
@@ -212,15 +214,10 @@ def reset_numtracker(server_config: ApplicationConfig):
     yield
 
 
-def test_cannot_access_endpoints(
-    client_without_auth: BlueapiClient, blueapi_client_get_methods: list[str]
-):
-    blueapi_client_get_methods.remove(
-        "get_oidc_config"
-    )  # get_oidc_config can be accessed without auth
-    for get_method in blueapi_client_get_methods:
-        with pytest.raises(BlueskyRemoteControlError, match=r"<Response \[401\]>"):
-            getattr(client_without_auth, get_method)()
+@pytest.mark.parametrize("method_name", authenticated_get_methods())
+def test_cannot_access_endpoints(client_without_auth: BlueapiClient, method_name: str):
+    with pytest.raises(BlueskyRemoteControlError, match=r"<Response \[401\]>"):
+        getattr(client_without_auth, method_name)()
 
 
 def test_can_get_oidc_config_without_auth(client_without_auth: BlueapiClient):

tests/unit_tests/client/test_client.py

@@ -392,7 +392,8 @@ def test_resume(
     )
 
 
-def test_cannot_run_task_without_message_bus(client: BlueapiClient):
+def test_cannot_run_task_without_message_bus(client: BlueapiClient, mock_rest: Mock):
+    mock_rest.get_stomp_config.return_value = None
     with pytest.raises(
         MissingStompConfigurationError,
         match="Stomp configuration required to run plans is missing or disabled",
@@ -660,8 +661,11 @@ def test_resume_span_ok(
 
 
 def test_cannot_run_task_span_ok(
-    exporter: JsonObjectSpanExporter, client: BlueapiClient
+    exporter: JsonObjectSpanExporter,
+    client: BlueapiClient,
+    mock_rest: Mock,
 ):
+    mock_rest.get_stomp_config.return_value = None
     with pytest.raises(
         MissingStompConfigurationError,
         match="Stomp configuration required to run plans is missing or disabled",

tests/unit_tests/test_cli.py

@@ -249,27 +249,30 @@ def test_submit_plan(runner: CliRunner):
 @responses.activate
 def test_submit_plan_without_stomp(runner: CliRunner):
     config_path = "tests/unit_tests/example_yaml/rest_config.yaml"
-    result = runner.invoke(
-        main,
-        [
-            "-c",
-            config_path,
-            "controller",
-            "run",
-            "-i",
-            "cm12345-1",
-            "sleep",
-            '{"time": 5}',
-        ],
-    )
+    with patch(
+        "blueapi.client.rest.BlueapiRestClient.get_stomp_config", return_value=None
+    ):
+        result = runner.invoke(
+            main,
+            [
+                "-c",
+                config_path,
+                "controller",
+                "run",
+                "-i",
+                "cm12345-1",
+                "sleep",
+                '{"time": 5}',
+            ],
+        )
 
     assert (
         result.stderr
         == "Error: Stomp configuration required to run plans is missing or disabled\n"
     )
 
 
-@patch("blueapi.client.client.StompClient")
+@patch("blueapi.client.event_bus.StompClient")
 @responses.activate
 def test_run_plan(stomp_client: StompClient, runner: CliRunner):
     task_id = "abcd-1234"
@@ -402,17 +405,20 @@ def test_invalid_stomp_config_for_listener(runner: CliRunner):
 
 
 def test_cannot_run_plans_without_stomp_config(runner: CliRunner):
-    result = runner.invoke(
-        main,
-        [
-            "controller",
-            "run",
-            "-i",
-            "cm12345-1",
-            "sleep",
-            '{"time": 5}',
-        ],
-    )
+    with patch(
+        "blueapi.client.rest.BlueapiRestClient.get_stomp_config", return_value=None
+    ):
+        result = runner.invoke(
+            main,
+            [
+                "controller",
+                "run",
+                "-i",
+                "cm12345-1",
+                "sleep",
+                '{"time": 5}',
+            ],
+        )
     assert result.exit_code == 1
     assert (
         result.stderr