{text}
" + return text + + class MessagingContainer: """ A container for providing easy access to multiple messaging services. @@ -74,29 +140,7 @@ def add_attachment( MessageObject: The updated message object. """ - if not os.path.isfile(file_path): - raise FileNotFoundError(f"Attachment file not found: {file_path}") - - file_size = os.path.getsize(file_path) - max_size = 5 * 1024 * 1024 # 5 MB - if file_size > max_size: - raise ValueError( - f"Attachment file size exceeds the maximum limit of 5 MB: {file_size} bytes" - ) - - with open(file_path, "rb") as f: - file_data = f.read() - - filename = os.path.basename(file_path) - mime_type, _ = mimetypes.guess_type(filename) - if mime_type is None: - mime_type = "application/octet-stream" - - self._content.append( - messages.MessagingServiceFileContent( - filename=filename, mime_type=mime_type, data=file_data, width=width, height=height - ) - ) + self._content.append(_build_attachment_content(file_path, width=width, height=height)) return self @@ -126,6 +170,7 @@ class MessagingService(ABC, Generic[MessageObjectT]): def __init__(self, redis_connector: RedisConnector) -> None: self._redis_connector = redis_connector self._scopes: set[str] = set() + self._auto_notifications: dict[str, list[str]] = {} self._enabled = False self._default_scope: str | list[str] | None = None self._service_config: messages.AvailableMessagingServicesMessage | None = None @@ -134,6 +179,12 @@ def __init__(self, redis_connector: RedisConnector) -> None: cb=self._on_new_scope_change_msg, from_start=True, ) + self._redis_connector.register( + MessageEndpoints.notification_config(), cb=self._on_notification_config_change_msg + ) + config_msg = self._redis_connector.get(MessageEndpoints.notification_config()) + if config_msg is not None: + self._update_auto_notifications(config_msg) def set_default_scope(self, scope: str | list[str] | None) -> None: """ @@ -146,6 +197,143 @@ def set_default_scope(self, scope: str | list[str] | None) -> None: raise ValueError(f"Scope '{scope}' is not available for this messaging service.") self._default_scope = scope + def set_auto_notifications( + self, + event_type: ( + Literal[ + "new_scan", + "scan_completed", + "alarm_warning", + "alarm_minor", + "alarm_major", + "scan_interlock", + ] + | str + ), + enabled: bool, + scopes: list[str] | str | None = None, + ) -> None: + """ + Set automatic notifications for a specific event type. + + Args: + event_type (Literal["new_scan", "scan_completed", "alarm_warning", "alarm_minor", "alarm_major", "scan_interlock"] | str): The type of event to set notifications for. + enabled (bool): Whether to enable or disable notifications for the event. + scopes (list[str] | str | None): The scopes to apply the notifications to. + """ + event_name = event_type.value if isinstance(event_type, enum.Enum) else event_type + scopes_list: list[str] = [] + if scopes is not None: + if isinstance(scopes, str): + scopes_list = [scopes] + else: + scopes_list = scopes + + for scope in scopes_list: + if scope not in self._scopes: + raise ValueError( + f"Scope '{scope}' is not available for this messaging service." + ) + else: + if self._default_scope is not None: + scopes_list = ( + [self._default_scope] + if isinstance(self._default_scope, str) + else self._default_scope + ) + elif not self._SUPPORTS_EMPTY_SCOPES: + raise ValueError( + "Scopes must be provided when there is no default scope and empty scopes are not supported." + ) + + if enabled: + # merge with existing scopes if already enabled for this event type + existing_scopes = set(self._auto_notifications.get(event_name, [])) + existing_scopes.update(scopes_list) + self._auto_notifications[event_name] = list(existing_scopes) + else: + # if disabling, remove the scopes for this event type, or the entire event type if no scopes are provided + if scopes_list: + existing_scopes = set(self._auto_notifications.get(event_name, [])) + existing_scopes.difference_update(scopes_list) + if existing_scopes: + self._auto_notifications[event_name] = list(existing_scopes) + else: + self._auto_notifications.pop(event_name, None) + else: + self._auto_notifications.pop(event_name, None) + + self._sync_auto_notifications_config(event_name, enabled=enabled, scopes=scopes_list) + + def _sync_auto_notifications_config( + self, event_name: str, enabled: bool, scopes: list[str] + ) -> None: + config_msg = self._redis_connector.get(MessageEndpoints.notification_config()) + if config_msg is None: + config_msg = messages.NotificationConfigMessage() + + routes = {name: list(targets) for name, targets in config_msg.routes.items()} + event_routes = list(routes.get(event_name, [])) + + if enabled: + scopes_to_add = scopes or [None] + for scope in scopes_to_add: + target = messages.NotificationServiceTarget( + service_name=self._SERVICE_NAME, scope=scope + ) + if not any(existing == target for existing in event_routes): + event_routes.append(target) + else: + if scopes: + event_routes = [ + target + for target in event_routes + if not (target.service_name == self._SERVICE_NAME and target.scope in scopes) + ] + else: + event_routes = [ + target for target in event_routes if target.service_name != self._SERVICE_NAME + ] + + if event_routes: + routes[event_name] = event_routes + else: + routes.pop(event_name, None) + + self._redis_connector.set_and_publish( + MessageEndpoints.notification_config(), + messages.NotificationConfigMessage(routes=routes, metadata=config_msg.metadata), + ) + + def _on_notification_config_change_msg( + self, + message: ( + MessageObject[messages.NotificationConfigMessage] + | dict[str, messages.NotificationConfigMessage] + ), + ) -> None: + config_msg = message.value if hasattr(message, "value") else message["data"] + self._update_auto_notifications(config_msg) + + def _update_auto_notifications(self, config_msg: messages.NotificationConfigMessage) -> None: + auto_notifications: dict[str, list[str]] = {} + for event_name, targets in config_msg.routes.items(): + scopes: list[str] = [] + has_matching_service = False + for target in targets: + if target.service_name != self._SERVICE_NAME: + continue + has_matching_service = True + if isinstance(target.scope, str): + scopes.append(target.scope) + elif isinstance(target.scope, list): + scopes.extend(target.scope) + + if has_matching_service: + auto_notifications[event_name] = list(dict.fromkeys(scopes)) + + self._auto_notifications = auto_notifications + def _on_new_scope_change_msg( self, message: dict[str, messages.AvailableMessagingServicesMessage] ) -> None: @@ -300,20 +488,8 @@ def add_text( Examples: >>> msg.add_text("Checks failed", bold=True, color="red") """ - if bold or italic or color: - if bold: - text = f"{text}" - if italic: - text = f"{text}" - if color: - if color == "black": - pass # black is the default - elif color in ["red", "green"]: - text = f'{text}' - elif color in ["yellow", "pink", "blue"]: - text = f'{text}' - text = f"{text}
" - return super().add_text(text) + super().add_text(_format_rich_text(text, bold=bold, italic=italic, color=color)) + return self def add_tags(self, tags: str | list[str]) -> Self: """ @@ -325,11 +501,13 @@ def add_tags(self, tags: str | list[str]) -> Self: Returns: SciLogMessageServiceObject: The updated message object. """ - if isinstance(tags, str): - tags = [tags] - # Ensure that there are no duplicates... - tags = list(set(self._service.get_default_tags() + tags)) # type: ignore + default_tags = ( + self._service.get_default_tags() # type: ignore + if self._service and hasattr(self._service, "get_default_tags") + else [] + ) + tags = _normalize_tags(default_tags + _normalize_tags(tags)) self._content.append(messages.MessagingServiceTagsContent(tags=tags)) return self @@ -345,7 +523,8 @@ def send(self, scope: str | list[str] | None = None) -> None: if not any( isinstance(content, messages.MessagingServiceTagsContent) for content in self._content ): - self.add_tags(self._service.get_default_tags()) # type: ignore + if self._service and hasattr(self._service, "get_default_tags"): + self.add_tags(self._service.get_default_tags()) # type: ignore super().send(scope=scope) @@ -457,3 +636,13 @@ def _normalize_scope(self, scope: str | list[str]) -> str | list[str]: if isinstance(scope, str): return self._normalize_phone_number(scope) return [self._normalize_phone_number(entry) for entry in scope] + + +class NotificationMessageObject(SciLogMessageServiceObject): + """ + Generic notification payload that can be adapted to concrete messaging + services during routing. + """ + + def __init__(self): + super().__init__(service=None) # type: ignore diff --git a/bec_lib/bec_lib/redis_connector.py b/bec_lib/bec_lib/redis_connector.py index 6683a147c..9e12d7aca 100644 --- a/bec_lib/bec_lib/redis_connector.py +++ b/bec_lib/bec_lib/redis_connector.py @@ -57,7 +57,10 @@ DynamicMetricDict, DynamicMetricMessage, ErrorInfo, + NotificationMessage, ) +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject from bec_lib.serialization import MsgpackSerialization logger = bec_logger.logger @@ -655,6 +658,34 @@ def raise_alarm(self, severity: Alarms, info: ErrorInfo, metadata: dict | None = """ alarm_msg = AlarmMessage(severity=severity, info=info, metadata=metadata or {}) self.set_and_publish(MessageEndpoints.alarm(), alarm_msg) + compact_message = info.compact_error_message or info.error_message or info.exception_type + event_by_severity = { + 0: MessagingEvent.ALARM_WARNING, + 1: MessagingEvent.ALARM_MINOR, + 2: MessagingEvent.ALARM_MAJOR, + } + self.notify(event_by_severity[int(severity)], compact_message) + + def notify( + self, + event: MessagingEvent | str, + message: str | NotificationMessageObject, + pipe: Pipeline | None = None, + ) -> None: + """ + Publish a notification event for downstream routing by SciHub. + + Args: + event(MessagingEvent | str): The type of the event that triggered the notification. + message(str | NotificationMessageObject): The notification content to be sent. + pipe(Pipeline, optional): Optional pipeline to enqueue the publish operation into. + """ + if isinstance(event, MessagingEvent): + event = event.value + if isinstance(message, str): + message = NotificationMessageObject().add_text(message) + outgoing = NotificationMessage(event=event, message=message._content) + self.send(MessageEndpoints.notification(event), outgoing, pipe=pipe) def pipeline(self) -> redis.client.Pipeline: """Create a new pipeline""" diff --git a/bec_lib/bec_lib/tests/utils.py b/bec_lib/bec_lib/tests/utils.py index 3e0a879ec..d95b28457 100644 --- a/bec_lib/bec_lib/tests/utils.py +++ b/bec_lib/bec_lib/tests/utils.py @@ -1,9 +1,7 @@ from __future__ import annotations import builtins -import copy import os -import threading import time from types import SimpleNamespace from typing import TYPE_CHECKING, Literal @@ -16,6 +14,8 @@ from bec_lib.devicemanager import DeviceManagerBase from bec_lib.endpoints import EndpointInfo, MessageEndpoints from bec_lib.logger import bec_logger +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject from bec_lib.redis_connector import RedisConnector from bec_lib.scan_manager import ScanManager from bec_lib.scans import Scans @@ -561,6 +561,17 @@ def send(self, topic, msg, pipe=None): raise TypeError("Message must be a BECMessage") return self.raw_send(topic, msg, pipe) + def notify(self, event, message: str | NotificationMessageObject, pipe=None): + if isinstance(event, MessagingEvent): + event = event.value + if isinstance(message, str): + message = NotificationMessageObject().add_text(message) + return self.send( + MessageEndpoints.notification(event), + messages.NotificationMessage(event=event, message=message._content), + pipe=pipe, + ) + def set_and_publish(self, topic, msg, pipe=None, expire: int = None): if pipe: pipe._pipe_buffer.append(("set_and_publish", (topic.endpoint, msg), {"expire": expire})) diff --git a/bec_lib/bec_lib/utils/scan_utils.py b/bec_lib/bec_lib/utils/scan_utils.py index 6abfbf6a3..9a6ac3f77 100644 --- a/bec_lib/bec_lib/utils/scan_utils.py +++ b/bec_lib/bec_lib/utils/scan_utils.py @@ -7,7 +7,8 @@ import csv import datetime from collections import defaultdict -from typing import TYPE_CHECKING +from collections.abc import Mapping +from typing import TYPE_CHECKING, Any from typeguard import typechecked @@ -62,6 +63,46 @@ def scan_to_csv( _write_csv(output_name=output_name, delimiter=delimiter, dialect=dialect, output=data_output) +def compose_cli_input_from_scan_info(scan_info: Mapping[str, Any] | Any) -> str: + """Compose a user-facing scan call string from a scan info object.""" + scan_name = _get_scan_info_value(scan_info, "scan_name") + if not scan_name: + return "scans.unknown()" + + request_inputs = _get_scan_info_value(scan_info, "request_inputs") or {} + arg_bundle = list(request_inputs.get("arg_bundle") or []) + inputs = request_inputs.get("inputs") or {} + kwargs = request_inputs.get("kwargs") or {} + + call_parts: list[str] = [_render_scan_cli_value(value) for value in arg_bundle] + call_parts.extend(f"{name}={_render_scan_cli_value(value)}" for name, value in inputs.items()) + call_parts.extend(f"{name}={_render_scan_cli_value(value)}" for name, value in kwargs.items()) + return f"scans.{scan_name}({', '.join(call_parts)})" + + +def _get_scan_info_value(scan_info: Mapping[str, Any] | Any, key: str) -> Any: + if isinstance(scan_info, Mapping): + return scan_info.get(key) + return getattr(scan_info, key, None) + + +def _render_scan_cli_value(value: Any) -> str: + if isinstance(value, Mapping): + rendered_items = ", ".join( + f"{_render_scan_cli_value(key)}: {_render_scan_cli_value(val)}" + for key, val in value.items() + ) + return "{" + rendered_items + "}" + if isinstance(value, tuple): + inner = ", ".join(_render_scan_cli_value(item) for item in value) + if len(value) == 1: + inner += "," + return f"({inner})" + if isinstance(value, list): + return "[" + ", ".join(_render_scan_cli_value(item) for item in value) + "]" + return repr(value) + + def _write_csv(output_name: str, delimiter: str, output: list, dialect: str = None) -> None: """Write csv file. diff --git a/bec_lib/pyproject.toml b/bec_lib/pyproject.toml index bc39197ba..0212b507b 100644 --- a/bec_lib/pyproject.toml +++ b/bec_lib/pyproject.toml @@ -40,6 +40,7 @@ dependencies = [ "python-slugify~=8.0", "questionary~=2.0", "phonenumbers~=9.0", + "beautifulsoup4~=4.14", ] @@ -73,42 +74,6 @@ bec_lib_fixtures = "bec_lib.tests.fixtures" "Bug Tracker" = "https://github.com/bec-project/bec/issues" Homepage = "https://github.com/bec-project/bec" - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - [tool.hatch.build.targets.wheel] include = ["*"] diff --git a/bec_lib/tests/test_bec_messages.py b/bec_lib/tests/test_bec_messages.py index 3ef23363a..211b426b4 100644 --- a/bec_lib/tests/test_bec_messages.py +++ b/bec_lib/tests/test_bec_messages.py @@ -5,6 +5,7 @@ import pytest from bec_lib import messages +from bec_lib.messaging_services import NotificationMessageObject from bec_lib.serialization import MsgpackSerialization @@ -134,6 +135,36 @@ def test_ClientInfoMessage_raises(): ) +def test_NotificationMessage(): + notification = ( + NotificationMessageObject() + .add_text("Scan started", bold=True, color="red") + .add_tags(["beamline", "scan"]) + ) + msg = messages.NotificationMessage(event="new_scan", message=notification._content) + res = MsgpackSerialization.dumps(msg) + res_loaded = MsgpackSerialization.loads(res) + assert res_loaded == msg + + +def test_NotificationConfigMessage(): + msg = messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="scilog", scope="logbook") + ], + "alarm": [ + messages.NotificationServiceTarget( + service_name="signal", scope=["+41791234567", "+41797654321"] + ) + ], + } + ) + res = MsgpackSerialization.dumps(msg) + res_loaded = MsgpackSerialization.loads(res) + assert res_loaded == msg + + def test_DeviceRPCMessage(): msg = messages.DeviceRPCMessage( device="samx", return_val=1, out="done", success=True, metadata={"RID": "1234"} diff --git a/bec_lib/tests/test_core_utils.py b/bec_lib/tests/test_core_utils.py index f4e0bcbe5..db80fce05 100644 --- a/bec_lib/tests/test_core_utils.py +++ b/bec_lib/tests/test_core_utils.py @@ -9,7 +9,13 @@ from bec_lib.scan_items import ScanItem from bec_lib.scan_report import ScanReport from bec_lib.utils.rpc_utils import user_access -from bec_lib.utils.scan_utils import _extract_scan_data, _write_csv, scan_to_csv, scan_to_dict +from bec_lib.utils.scan_utils import ( + _extract_scan_data, + _render_scan_cli_value, + _write_csv, + scan_to_csv, + scan_to_dict, +) class ScanReportMock(ScanReport): @@ -129,6 +135,20 @@ def test_scan_to_csv(): header=None, write_metadata=True, ) + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("samx", "'samx'"), + ({"motor": "samx", "position": 1.5}, "{'motor': 'samx', 'position': 1.5}"), + ((1,), "(1,)"), + ((1, "samx"), "(1, 'samx')"), + ([1, {"nested": [2, 3]}], "[1, {'nested': [2, 3]}]"), + ], +) +def test_render_scan_cli_value(value, expected): + assert _render_scan_cli_value(value) == expected with pytest.raises(Exception): scan_to_csv( scan_report=[scanreport_mock, scanreport_mock, scanreport_mock], diff --git a/bec_lib/tests/test_messaging_hooks.py b/bec_lib/tests/test_messaging_hooks.py new file mode 100644 index 000000000..9f0d70f3f --- /dev/null +++ b/bec_lib/tests/test_messaging_hooks.py @@ -0,0 +1,258 @@ +from types import SimpleNamespace +from unittest import mock + +import pytest + +from bec_lib import messages +from bec_lib.connector import MessageObject +from bec_lib.endpoints import MessageEndpoints, MessageOp +from bec_lib.messaging_hooks import MessagingEvent, MessagingManager +from bec_lib.messaging_services import NotificationMessageObject + + +@pytest.fixture +def messaging_manager(connected_connector): + manager = MessagingManager(connected_connector) + yield manager + manager.shutdown() + + +@pytest.fixture +def messaging_manager_with_initial_config(connected_connector): + config_msg = messages.NotificationConfigMessage( + routes={ + "new_scan": [messages.NotificationServiceTarget(service_name="scilog", scope="logbook")] + } + ) + connected_connector.set_and_publish(MessageEndpoints.notification_config(), config_msg) + + manager = MessagingManager(connected_connector) + yield manager + manager.shutdown() + + +def _available_services_message(): + return messages.AvailableMessagingServicesMessage( + config=messages.MessagingConfig( + signal=messages.MessagingServiceScopeConfig(enabled=True, default=None), + teams=messages.MessagingServiceScopeConfig(enabled=False, default=None), + scilog=messages.MessagingServiceScopeConfig(enabled=True, default=None), + ), + deployment_services=[ + messages.SciLogServiceInfo( + id="scilog-default", scope="logbook", enabled=True, logbook_id="lb-1" + ), + messages.SignalServiceInfo( + id="signal-default", scope="ops", enabled=True, group_id="g-1" + ), + ], + session_services=[], + ) + + +def test_notification_endpoints(): + event_endpoint = MessageEndpoints.notification("new_scan") + config_endpoint = MessageEndpoints.notification_config() + + assert event_endpoint.endpoint == "internal/messaging_services/notification/new_scan" + assert event_endpoint.message_type is messages.NotificationMessage + assert event_endpoint.message_op == MessageOp.SEND + + assert config_endpoint.endpoint == "user/messaging_services/notification_config" + assert config_endpoint.message_type is messages.NotificationConfigMessage + assert config_endpoint.message_op == MessageOp.SET_PUBLISH + + +def test_messaging_manager_loads_initial_config(messaging_manager_with_initial_config): + assert messaging_manager_with_initial_config.config == { + MessagingEvent.SCAN: [ + messages.NotificationServiceTarget(service_name="scilog", scope="logbook") + ] + } + + +def test_messaging_manager_routes_notifications_to_message_service_queue( + connected_connector, messaging_manager +): + available_services = _available_services_message() + messaging_manager._service_by_name["scilog"]._on_new_scope_change_msg( + {"data": available_services} + ) + messaging_manager._service_by_name["signal"]._on_new_scope_change_msg( + {"data": available_services} + ) + messaging_manager._service_by_name["teams"]._on_new_scope_change_msg( + {"data": available_services} + ) + + messaging_manager.on_notification_config( + messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="scilog", scope="logbook") + ] + } + ) + ) + msg_obj = NotificationMessageObject().add_text("Scan started").add_tags(["bec", "new_scan"]) + messaging_manager.on_notification( + MessagingEvent.SCAN, + messages.NotificationMessage(event="new_scan", message=msg_obj._content), + ) + + out = connected_connector.xread(MessageEndpoints.message_service_queue(), from_start=True) + assert len(out) == 1 + sent_message = out[0]["data"] + assert sent_message.service_name == "scilog" + assert sent_message.scope == "logbook" + assert isinstance(sent_message.message[0], messages.MessagingServiceTextContent) + assert sent_message.message[0].content == "Scan started" + assert isinstance(sent_message.message[1], messages.MessagingServiceTagsContent) + assert sent_message.message[1].tags == ["bec", "new_scan"] + + +def test_messaging_manager_routes_generic_notification_to_signal_as_plain_text( + connected_connector, messaging_manager +): + available_services = _available_services_message() + messaging_manager._service_by_name["signal"]._on_new_scope_change_msg( + {"data": available_services} + ) + + messaging_manager.on_notification_config( + messages.NotificationConfigMessage( + routes={ + "alarm_major": [ + messages.NotificationServiceTarget(service_name="signal", scope="ops") + ] + } + ) + ) + + messaging_manager.on_notification( + MessagingEvent.ALARM_MAJOR, + messages.NotificationMessage( + event="alarm_major", + message=NotificationMessageObject() + .add_text("Beamline checks failed", bold=True, color="red") + .add_tags(["alarm"]) + ._content, + ), + ) + + out = connected_connector.xread(MessageEndpoints.message_service_queue(), from_start=True) + assert len(out) == 1 + sent_message = out[0]["data"] + assert sent_message.service_name == "signal" + assert sent_message.scope == "ops" + assert len(sent_message.message) == 1 + assert isinstance(sent_message.message[0], messages.MessagingServiceTextContent) + assert sent_message.message[0].content == "Beamline checks failed" + + +def test_handle_notification_routes_message_from_topic_suffix(messaging_manager): + notification_message = messages.NotificationMessage( + event="scan_completed", + message=NotificationMessageObject().add_text("Scan complete")._content, + ) + msg_obj = MessageObject( + MessageEndpoints.notification("scan_completed").endpoint, notification_message + ) + + with mock.patch.object(messaging_manager, "on_notification") as on_notification: + messaging_manager._handle_notification(msg_obj) + + on_notification.assert_called_once_with("scan_completed", notification_message) + + +def test_handle_notification_config_forwards_message_value(messaging_manager): + config_message = messages.NotificationConfigMessage( + routes={ + "alarm_minor": [messages.NotificationServiceTarget(service_name="signal", scope="ops")] + } + ) + msg_obj = MessageObject(MessageEndpoints.notification_config().endpoint, config_message) + + with mock.patch.object(messaging_manager, "on_notification_config") as on_notification_config: + messaging_manager._handle_notification_config(msg_obj) + + on_notification_config.assert_called_once_with(config_message) + + +def test_to_service_message_for_teams_strips_html_and_ignores_non_text(messaging_manager): + available_services = messages.AvailableMessagingServicesMessage( + config=messages.MessagingConfig( + signal=messages.MessagingServiceScopeConfig(enabled=True, default=None), + teams=messages.MessagingServiceScopeConfig(enabled=True, default=None), + scilog=messages.MessagingServiceScopeConfig(enabled=True, default=None), + ), + deployment_services=[ + messages.TeamsServiceInfo( + id="teams-default", + scope="ops-team", + enabled=True, + workflow_webhook_url="https://example.invalid/webhook", + ) + ], + session_services=[], + ) + messaging_manager._service_by_name["teams"]._on_new_scope_change_msg( + {"data": available_services} + ) + message = messages.NotificationMessage( + event="alarm_major", + message=NotificationMessageObject() + .add_text("Beamline checks failed", bold=True, color="red") + .add_tags(["alarm"]) + ._content, + ) + + routed_message = messaging_manager.to_service_message( + messaging_manager._service_by_name["teams"], message + ) + + assert len(routed_message._content) == 1 + assert isinstance(routed_message._content[0], messages.MessagingServiceTextContent) + assert routed_message._content[0].content == "Beamline checks failed" + + +def test_to_service_message_for_signal_preserves_files_and_plain_text(messaging_manager): + available_services = _available_services_message() + messaging_manager._service_by_name["signal"]._on_new_scope_change_msg( + {"data": available_services} + ) + file_content = messages.MessagingServiceFileContent( + filename="alarm.png", mime_type="image/png", data=b"png-bytes", width=128, height=64 + ) + message = messages.NotificationMessage( + event="alarm_warning", + message=NotificationMessageObject() + .add_text("Beamline checks failed", italic=True) + .add_tags(["alarm"]) + ._content + + [file_content], + ) + + routed_message = messaging_manager.to_service_message( + messaging_manager._service_by_name["signal"], message + ) + + assert len(routed_message._content) == 2 + assert isinstance(routed_message._content[0], messages.MessagingServiceTextContent) + assert routed_message._content[0].content == "Beamline checks failed" + assert routed_message._content[1] == file_content + + +def test_to_service_message_raises_for_unsupported_service(messaging_manager): + message = messages.NotificationMessage( + event="alarm_minor", + message=NotificationMessageObject().add_text("Beamline checks failed")._content, + ) + + with mock.patch("bec_lib.messaging_hooks.logger.warning") as warning: + with pytest.raises(ValueError, match="Unsupported messaging service: unsupported"): + messaging_manager.to_service_message( + SimpleNamespace(_SERVICE_NAME="unsupported"), message + ) + + warning.assert_not_called() diff --git a/bec_lib/tests/test_messaging_service.py b/bec_lib/tests/test_messaging_service.py index 5ddca089e..b730e1a0e 100644 --- a/bec_lib/tests/test_messaging_service.py +++ b/bec_lib/tests/test_messaging_service.py @@ -1,9 +1,13 @@ +import time + import pytest from bec_lib import messages from bec_lib.endpoints import MessageEndpoints +from bec_lib.messaging_hooks import MessagingEvent, MessagingManager from bec_lib.messaging_services import ( MessageServiceObject, + NotificationMessageObject, SciLogMessagingService, SignalMessageServiceObject, SignalMessagingService, @@ -338,6 +342,66 @@ def test_signal_messaging_service_send_with_sticker(signal_service, connected_co assert sticker_part.sticker_id == "sticker_123" +def test_notification_message_object_to_scilog_message(scilog_service): + scilog_service.set_default_scope("default") + notification = ( + NotificationMessageObject() + .add_text("Beamline checks failed", bold=True, color="red") + .add_tags(["alarm"]) + ) + manager = MessagingManager(scilog_service._redis_connector) # pylint: disable=protected-access + + try: + scilog_message = manager.to_service_message( + scilog_service, + messages.NotificationMessage(event="alarm_major", message=notification._content), + ) + finally: + manager.shutdown() + + assert scilog_message._service == scilog_service # pylint: disable=protected-access + assert scilog_message._scope == "default" # pylint: disable=protected-access + assert len(scilog_message._content) == 2 # pylint: disable=protected-access + assert isinstance( + scilog_message._content[0], messages.MessagingServiceTextContent + ) # pylint: disable=protected-access + assert ( + scilog_message._content[0].content # pylint: disable=protected-access + == 'Beamline checks failed
' + ) + assert isinstance( + scilog_message._content[1], messages.MessagingServiceTagsContent + ) # pylint: disable=protected-access + + +def test_notification_message_object_to_signal_message(signal_service): + signal_service.set_default_scope("default") + notification = ( + NotificationMessageObject() + .add_text("Beamline checks failed", bold=True, color="red") + .add_tags(["alarm"]) + ) + manager = MessagingManager(signal_service._redis_connector) # pylint: disable=protected-access + + try: + signal_message = manager.to_service_message( + signal_service, + messages.NotificationMessage(event="alarm_major", message=notification._content), + ) + finally: + manager.shutdown() + + assert signal_message._service == signal_service # pylint: disable=protected-access + assert signal_message._scope == "default" # pylint: disable=protected-access + assert len(signal_message._content) == 1 # pylint: disable=protected-access + assert isinstance( + signal_message._content[0], messages.MessagingServiceTextContent + ) # pylint: disable=protected-access + assert ( + signal_message._content[0].content == "Beamline checks failed" + ) # pylint: disable=protected-access + + def test_signal_service_can_create_message_without_configured_scopes(connected_connector): """Test that Signal remains enabled even without predefined scopes.""" service = SignalMessagingService(connected_connector) @@ -552,3 +616,158 @@ def test_scilog_add_text_bold_and_color(scilog_message, connected_connector): text_part.content == 'Beamline checks failed
' ) + + +def test_set_auto_notifications_persists_notification_config(scilog_service, connected_connector): + scilog_service.set_auto_notifications(MessagingEvent.SCAN, enabled=True, scopes="default") + + config_msg = connected_connector.get(MessageEndpoints.notification_config()) + assert config_msg == messages.NotificationConfigMessage( + routes={ + "new_scan": [messages.NotificationServiceTarget(service_name="scilog", scope="default")] + } + ) + assert scilog_service._auto_notifications == { + "new_scan": ["default"] + } # pylint: disable=protected-access + + +def test_set_auto_notifications_merges_with_existing_routes(scilog_service, connected_connector): + connected_connector.set_and_publish( + MessageEndpoints.notification_config(), + messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops") + ] + } + ), + ) + + scilog_service.set_auto_notifications(MessagingEvent.SCAN, enabled=True, scopes="default") + + config_msg = connected_connector.get(MessageEndpoints.notification_config()) + assert config_msg == messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops"), + messages.NotificationServiceTarget(service_name="scilog", scope="default"), + ] + } + ) + assert scilog_service._auto_notifications == { + "new_scan": ["default"] + } # pylint: disable=protected-access + + +def test_set_auto_notifications_disable_removes_only_matching_service_scope( + scilog_service, connected_connector +): + connected_connector.set_and_publish( + MessageEndpoints.notification_config(), + messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops"), + messages.NotificationServiceTarget(service_name="scilog", scope="default"), + ] + } + ), + ) + + scilog_service.set_auto_notifications(MessagingEvent.SCAN, enabled=False, scopes="default") + + config_msg = connected_connector.get(MessageEndpoints.notification_config()) + assert config_msg == messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops") + ] + } + ) + assert scilog_service._auto_notifications == {} # pylint: disable=protected-access + + +def test_set_auto_notifications_uses_default_scope_when_scopes_omitted( + scilog_service, connected_connector +): + scilog_service.set_default_scope("default") + + scilog_service.set_auto_notifications(MessagingEvent.SCAN, enabled=True) + + config_msg = connected_connector.get(MessageEndpoints.notification_config()) + assert config_msg == messages.NotificationConfigMessage( + routes={ + "new_scan": [messages.NotificationServiceTarget(service_name="scilog", scope="default")] + } + ) + assert scilog_service._auto_notifications == { + "new_scan": ["default"] + } # pylint: disable=protected-access + + +def test_messaging_service_tracks_external_notification_config_updates( + scilog_service, connected_connector +): + connected_connector.set_and_publish( + MessageEndpoints.notification_config(), + messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="signal", scope="beamline-ops"), + messages.NotificationServiceTarget(service_name="scilog", scope="default"), + ], + "alarm_major": [ + messages.NotificationServiceTarget( + service_name="scilog", scope=["default", "secondary"] + ) + ], + } + ), + ) + + deadline = time.time() + 1 + while ( + time.time() < deadline + and scilog_service._auto_notifications # pylint: disable=protected-access + != {"new_scan": ["default"], "alarm_major": ["default", "secondary"]} + ): + time.sleep(0.01) + + assert scilog_service._auto_notifications == { # pylint: disable=protected-access + "new_scan": ["default"], + "alarm_major": ["default", "secondary"], + } + + +def test_messaging_service_loads_notification_config_on_init(connected_connector): + connected_connector.set_and_publish( + MessageEndpoints.notification_config(), + messages.NotificationConfigMessage( + routes={ + "new_scan": [ + messages.NotificationServiceTarget(service_name="scilog", scope="default") + ] + } + ), + ) + + service = SciLogMessagingService(connected_connector) + available_services = messages.AvailableMessagingServicesMessage( + config=messages.MessagingConfig( + signal=messages.MessagingServiceScopeConfig(enabled=False), + teams=messages.MessagingServiceScopeConfig(enabled=False), + scilog=messages.MessagingServiceScopeConfig(enabled=True), + ), + deployment_services=[ + messages.SciLogServiceInfo( + id="test_scilog", scope="default", enabled=True, logbook_id="test_logbook" + ) + ], + session_services=[], + ) + service._on_new_scope_change_msg(message={"data": available_services}) + + assert service._auto_notifications == { + "new_scan": ["default"] + } # pylint: disable=protected-access diff --git a/bec_lib/tests/test_redis_connector.py b/bec_lib/tests/test_redis_connector.py index 7cd9101ec..2dff189a2 100644 --- a/bec_lib/tests/test_redis_connector.py +++ b/bec_lib/tests/test_redis_connector.py @@ -15,6 +15,7 @@ ClientInfoMessage, ProcedureExecutionMessage, ) +from bec_lib.messaging_hooks import MessagingEvent from bec_lib.redis_connector import ( IncompatibleMessageForEndpoint, IncompatibleRedisOperation, @@ -63,15 +64,41 @@ def test_redis_connector_send_client_info(connector): @pytest.mark.parametrize( - "severity, alarm_type, msg, compact_msg, metadata", + "severity, expected_event, alarm_type, msg, compact_msg, metadata", [ - [Alarms.MAJOR, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}], - [Alarms.MINOR, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}], - [Alarms.WARNING, "alarm", "content1", "compact_msg", {"metadata": "metadata1"}], + [ + Alarms.MAJOR, + MessagingEvent.ALARM_MAJOR, + "alarm", + "content1", + "compact_msg", + {"metadata": "metadata1"}, + ], + [ + Alarms.MINOR, + MessagingEvent.ALARM_MINOR, + "alarm", + "content1", + "compact_msg", + {"metadata": "metadata1"}, + ], + [ + Alarms.WARNING, + MessagingEvent.ALARM_WARNING, + "alarm", + "content1", + "compact_msg", + {"metadata": "metadata1"}, + ], ], ) -def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compact_msg, metadata): - with mock.patch.object(connector, "set_and_publish", return_value=None): +def test_redis_connector_raise_alarm( + connector, severity, expected_event, alarm_type, msg, compact_msg, metadata +): + with ( + mock.patch.object(connector, "set_and_publish", return_value=None), + mock.patch.object(connector, "notify", return_value=None), + ): info = messages.ErrorInfo( error_message=msg, compact_error_message=compact_msg, exception_type=alarm_type ) @@ -80,6 +107,7 @@ def test_redis_connector_raise_alarm(connector, severity, alarm_type, msg, compa connector.set_and_publish.assert_called_once_with( MessageEndpoints.alarm(), AlarmMessage(severity=severity, info=info, metadata=metadata) ) + connector.notify.assert_called_once_with(expected_event, compact_msg) @pytest.mark.parametrize( diff --git a/bec_server/bec_server/actors/scan_interlock.py b/bec_server/bec_server/actors/scan_interlock.py index 89f3379fb..be1806847 100644 --- a/bec_server/bec_server/actors/scan_interlock.py +++ b/bec_server/bec_server/actors/scan_interlock.py @@ -6,6 +6,8 @@ ScanInterlockModifyStateTableMessage, ScanInterlockStateTableContent, ) +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject from bec_server.actors.actor import BlStateActor logger = bec_logger.logger @@ -81,6 +83,13 @@ def mismatched_states(self): ] def some_mismatch_action(self, client: BECClient): + notification = NotificationMessageObject() + notification.add_text( + f"Scan interlock triggered for beamline states: {self.mismatched_states}", color="red" + ) + notification.add_tags("scan_interlock") + self.client.connector.notify(MessagingEvent.SCAN_INTERLOCK, notification) + if self.client.queue is None: return logger.info( @@ -107,6 +116,10 @@ def _unlock(self): f"{self.name} removing queue lock if present; " f"queue_locks={primary.locks}; cache={self.state_cache}; table={self.state_table}" ) + notification = NotificationMessageObject() + notification.add_text("Scan interlock cleared", color="green") + notification.add_tags("scan_interlock") + self.client.connector.notify(MessagingEvent.SCAN_INTERLOCK, notification) self.client.queue.remove_queue_lock(queue="primary", lock_id=self._LOCK_ID) def run(self): diff --git a/bec_server/bec_server/scan_server/generator_scan_worker.py b/bec_server/bec_server/scan_server/generator_scan_worker.py index 117a3f269..08f0c423d 100644 --- a/bec_server/bec_server/scan_server/generator_scan_worker.py +++ b/bec_server/bec_server/scan_server/generator_scan_worker.py @@ -11,6 +11,9 @@ from bec_lib.endpoints import MessageEndpoints from bec_lib.file_utils import compile_file_components from bec_lib.logger import bec_logger +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject +from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info from .errors import DeviceInstructionError, ScanAbortion, UserScanInterruption from .scan_queue import InstructionQueueItem, InstructionQueueStatus, RequestBlock @@ -341,6 +344,29 @@ def _send_scan_status( self.worker.device_manager.connector.set_and_publish( MessageEndpoints.scan_status(), msg, pipe=pipe ) + cli_input = compose_cli_input_from_scan_info(self.current_scan_info) + scan_number = self.current_scan_info.get("scan_number") + scan_id = self.current_scan_id + if status == "open": + notification = NotificationMessageObject() + notification.add_text( + f"Scan started: scan_number={scan_number} ({cli_input}, scan_id={scan_id})", + color="green", + ) + notification.add_tags("scan_start") + self.worker.device_manager.connector.notify( + MessagingEvent.SCAN, notification, pipe=pipe + ) + elif status in {"closed", "user_completed"}: + notification = NotificationMessageObject() + notification.add_text( + f"Scan completed: scan_number={scan_number} ({cli_input}, scan_id={scan_id})", + color="green", + ) + notification.add_tags("scan_completed") + self.worker.device_manager.connector.notify( + MessagingEvent.SCAN_COMPLETED, notification, pipe=pipe + ) pipe.execute() def update_instr_with_scan_report(self, instr: messages.DeviceInstructionMessage): diff --git a/bec_server/bec_server/scan_server/scans/scan_actions.py b/bec_server/bec_server/scan_server/scans/scan_actions.py index d4bc32655..30f90602c 100644 --- a/bec_server/bec_server/scan_server/scans/scan_actions.py +++ b/bec_server/bec_server/scan_server/scans/scan_actions.py @@ -14,6 +14,9 @@ from bec_lib.endpoints import MessageEndpoints from bec_lib.file_utils import compile_file_components from bec_lib.logger import bec_logger +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject +from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info from bec_server.scan_server.scan_stubs import ScanStubStatus if TYPE_CHECKING: @@ -1062,6 +1065,25 @@ def _send_scan_status( MessageEndpoints.public_scan_info(scan.scan_info.scan_id), msg, pipe=pipe, expire=expire ) self._connector.set_and_publish(MessageEndpoints.scan_status(), msg, pipe=pipe) + cli_input = compose_cli_input_from_scan_info(scan.scan_info) + scan_number = scan.scan_info.scan_number + scan_id = scan.scan_info.scan_id + if status == "open": + msg = NotificationMessageObject() + msg.add_text( + f"Scan started: scan_number={scan_number} ({cli_input}, scan_id={scan_id})", + color="green", + ) + msg.add_tags("scan_start") + self._connector.notify(MessagingEvent.SCAN, msg, pipe=pipe) + elif status in {"closed", "user_completed"}: + msg = NotificationMessageObject() + msg.add_text( + f"Scan completed: scan_number={scan_number} ({cli_input}, scan_id={scan_id})", + color="green", + ) + msg.add_tags("scan_completed") + self._connector.notify(MessagingEvent.SCAN_COMPLETED, msg, pipe=pipe) pipe.execute() def _build_scan_status_message( diff --git a/bec_server/bec_server/scihub/scihub.py b/bec_server/bec_server/scihub/scihub.py index a7af08447..86ff8946e 100644 --- a/bec_server/bec_server/scihub/scihub.py +++ b/bec_server/bec_server/scihub/scihub.py @@ -4,6 +4,7 @@ from bec_lib import messages from bec_lib.bec_service import BECService +from bec_lib.messaging_hooks import MessagingManager from bec_lib.service_config import ServiceConfig from bec_server.scihub.atlas import AtlasConnector from bec_server.scihub.service_handler.service_handler import ServiceHandler @@ -18,8 +19,10 @@ def __init__(self, config: ServiceConfig, connector_cls: type[RedisConnector]) - self.config = config self.atlas_connector = None self.service_handler = None + self.messaging_manager = None self._start_atlas_connector() self._start_service_handler() + self._start_messaging_manager() self.status = messages.BECStatus.RUNNING def _start_atlas_connector(self): @@ -31,7 +34,15 @@ def _start_service_handler(self): self.service_handler = ServiceHandler(self.connector) self.service_handler.start() + def _start_messaging_manager(self): + self.messaging_manager = MessagingManager(self.connector) + def shutdown(self): - super().shutdown() + """ + Shutdown the SciHub service with all its components. + """ + if self.messaging_manager: + self.messaging_manager.shutdown() if self.atlas_connector: self.atlas_connector.shutdown() + super().shutdown() diff --git a/bec_server/tests/tests_scan_server/scans_v4/test_scan_actions.py b/bec_server/tests/tests_scan_server/scans_v4/test_scan_actions.py index 2b1aaf073..1a380e6cf 100644 --- a/bec_server/tests/tests_scan_server/scans_v4/test_scan_actions.py +++ b/bec_server/tests/tests_scan_server/scans_v4/test_scan_actions.py @@ -8,8 +8,11 @@ from bec_lib import messages from bec_lib.device import ReadoutPriority from bec_lib.endpoints import MessageEndpoints +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject from bec_lib.tests.fixtures import dm_with_devices # noqa: F401 from bec_lib.tests.utils import ConnectorMock +from bec_lib.utils.scan_utils import compose_cli_input_from_scan_info from bec_server.scan_server.instruction_handler import InstructionHandler from bec_server.scan_server.scan_stubs import ScanStubStatus from bec_server.scan_server.scans.scan_base import ScanBase, ScanInfo @@ -107,6 +110,40 @@ def test_scan_info_stores_scan_report_device_objects_as_names(dm_with_devices): assert scan_info.scan_report_devices == ["samz"] +def test_compose_cli_input_from_scan_info_uses_named_inputs(): + scan_info = ScanInfo( + scan_name="_v4_test_scan", + scan_id="scan-id-test", + scan_type=None, + request_inputs={ + "arg_bundle": [], + "inputs": {"device": "samx", "target": 1}, + "kwargs": {"relative": True}, + }, + ) + + assert ( + compose_cli_input_from_scan_info(scan_info) + == "scans._v4_test_scan(device='samx', target=1, relative=True)" + ) + + +def test_compose_cli_input_from_scan_info_uses_arg_bundle(): + scan_info = { + "scan_name": "line_scan", + "request_inputs": { + "arg_bundle": ["samx", -5, 5, 3], + "inputs": {}, + "kwargs": {"exp_time": 0.1}, + }, + } + + assert ( + compose_cli_input_from_scan_info(scan_info) + == "scans.line_scan('samx', -5, 5, 3, exp_time=0.1)" + ) + + def _last_device_instruction(ctx, action): return _sent_device_instructions(ctx, action)[-1] @@ -528,8 +565,10 @@ def test_send_scan_status_publishes_message(action_context): ctx.connector.pipeline = mock.MagicMock(return_value=pipe) ctx.connector.set = mock.MagicMock() ctx.connector.set_and_publish = mock.MagicMock() + ctx.connector.notify = mock.MagicMock() status_msg = messages.ScanStatusMessage(scan_id="scan-id-test", status="closed", info={}) ctx.actions._build_scan_status_message = mock.MagicMock(return_value=status_msg) + ctx.scan.scan_info.request_inputs = {"arg_bundle": [], "inputs": {}, "kwargs": {}} ctx.actions._send_scan_status("closed", reason="alarm") @@ -540,9 +579,50 @@ def test_send_scan_status_publishes_message(action_context): ctx.connector.set_and_publish.assert_called_once_with( MessageEndpoints.scan_status(), status_msg, pipe=pipe ) + assert ctx.connector.notify.call_count == 1 + event, notification = ctx.connector.notify.call_args.args + assert event == MessagingEvent.SCAN_COMPLETED + assert isinstance(notification, NotificationMessageObject) + assert notification._content == [ # pylint: disable=protected-access + messages.MessagingServiceTextContent( + content='Scan completed: scan_number=1 (scans._v4_test_scan(), scan_id=scan-id-test)
' + ), + messages.MessagingServiceTagsContent(tags=["scan_completed"]), + ] + assert ctx.connector.notify.call_args.kwargs == {"pipe": pipe} pipe.execute.assert_called_once_with() +def test_send_scan_status_publishes_new_scan_notification(action_context): + ctx = action_context() + pipe = mock.MagicMock() + ctx.connector.pipeline = mock.MagicMock(return_value=pipe) + ctx.connector.set = mock.MagicMock() + ctx.connector.set_and_publish = mock.MagicMock() + ctx.connector.notify = mock.MagicMock() + status_msg = messages.ScanStatusMessage(scan_id="scan-id-test", status="open", info={}) + ctx.actions._build_scan_status_message = mock.MagicMock(return_value=status_msg) + ctx.scan.scan_info.request_inputs = { + "arg_bundle": [], + "inputs": {"device": "samx"}, + "kwargs": {}, + } + + ctx.actions._send_scan_status("open") + + assert ctx.connector.notify.call_count == 1 + event, notification = ctx.connector.notify.call_args.args + assert event == MessagingEvent.SCAN + assert isinstance(notification, NotificationMessageObject) + assert notification._content == [ # pylint: disable=protected-access + messages.MessagingServiceTextContent( + content="Scan started: scan_number=1 (scans._v4_test_scan(device='samx'), scan_id=scan-id-test)
" + ), + messages.MessagingServiceTagsContent(tags=["scan_start"]), + ] + assert ctx.connector.notify.call_args.kwargs == {"pipe": pipe} + + def test_get_file_base_path_uses_account_and_templates(action_context): ctx = action_context() ctx.device_manager.parent = _TestParent("/tmp/data") diff --git a/bec_server/tests/tests_scan_server/test_generator_scan_worker.py b/bec_server/tests/tests_scan_server/test_generator_scan_worker.py index bd183c96d..c58898205 100644 --- a/bec_server/tests/tests_scan_server/test_generator_scan_worker.py +++ b/bec_server/tests/tests_scan_server/test_generator_scan_worker.py @@ -403,7 +403,15 @@ def test_send_scan_status(generator_worker_mock, status, expire): worker = generator_worker_mock worker.worker.device_manager.connector = ConnectorMock() worker.current_scan_id = str(uuid.uuid4()) - worker.current_scan_info = {"scan_number": 5} + worker.current_scan_info = { + "scan_number": 5, + "scan_name": "line_scan", + "request_inputs": { + "arg_bundle": ["samx", -5, 5, 3], + "inputs": {}, + "kwargs": {"exp_time": 0.1}, + }, + } worker._send_scan_status(status) scan_info_msgs = [ msg @@ -413,6 +421,36 @@ def test_send_scan_status(generator_worker_mock, status, expire): ] assert len(scan_info_msgs) == 1 assert scan_info_msgs[0]["expire"] == expire + notification_msgs = [ + msg + for msg in worker.worker.device_manager.connector.message_sent + if msg["queue"] + == MessageEndpoints.notification( + "new_scan" if status == "open" else "scan_completed" + ).endpoint + ] + if status == "open": + assert len(notification_msgs) == 1 + assert notification_msgs[0]["msg"].event == "new_scan" + assert notification_msgs[0]["msg"].message == [ + messages.MessagingServiceTextContent( + content='Scan started: scan_number=5 ' + f"(scans.line_scan('samx', -5, 5, 3, exp_time=0.1), scan_id={worker.current_scan_id})
" + ), + messages.MessagingServiceTagsContent(tags=["scan_start"]), + ] + elif status == "closed": + assert len(notification_msgs) == 1 + assert notification_msgs[0]["msg"].event == "scan_completed" + assert notification_msgs[0]["msg"].message == [ + messages.MessagingServiceTextContent( + content='Scan completed: scan_number=5 ' + f"(scans.line_scan('samx', -5, 5, 3, exp_time=0.1), scan_id={worker.current_scan_id})
" + ), + messages.MessagingServiceTagsContent(tags=["scan_completed"]), + ] + else: + assert notification_msgs == [] @pytest.mark.parametrize("abortion", [False, True]) diff --git a/bec_server/tests/tests_scan_server/test_scan_interlock.py b/bec_server/tests/tests_scan_server/test_scan_interlock.py index db29d609d..a32981a32 100644 --- a/bec_server/tests/tests_scan_server/test_scan_interlock.py +++ b/bec_server/tests/tests_scan_server/test_scan_interlock.py @@ -3,6 +3,8 @@ import pytest +from bec_lib.messaging_hooks import MessagingEvent +from bec_lib.messaging_services import NotificationMessageObject from bec_server.actors.scan_interlock import ScanInterlockActor @@ -104,6 +106,11 @@ def test_mismatched_states_handles_list_of_statuses(self, actor): def test_some_mismatch_action_adds_lock(self, actor, mock_client): actor.some_mismatch_action(mock_client) + mock_client.connector.notify.assert_called_once() + event, notification = mock_client.connector.notify.call_args.args + assert event == MessagingEvent.SCAN_INTERLOCK + assert isinstance(notification, NotificationMessageObject) + assert notification._content[1].tags == ["scan_interlock"] mock_client.queue.add_queue_lock.assert_called_once_with( queue="primary", reason="Interlock for beamline states: []", @@ -114,6 +121,7 @@ def test_some_mismatch_action_skips_if_no_queue(self, actor, mock_client): add_queue_lock = mock_client.queue.add_queue_lock actor.client.queue = None actor.some_mismatch_action(mock_client) + mock_client.connector.notify.assert_called_once() add_queue_lock.assert_not_called() def test_all_match_action_unlocks(self, actor): @@ -124,6 +132,23 @@ def test_all_match_action_unlocks(self, actor): def test_unlock_removes_lock(self, actor, mock_client): actor._unlock() + mock_client.connector.notify.assert_called_once() + event, notification = mock_client.connector.notify.call_args.args + assert event == MessagingEvent.SCAN_INTERLOCK + assert isinstance(notification, NotificationMessageObject) + assert ( + notification._content[0].content + == 'Scan interlock cleared
' + ) + assert notification._content[1].tags == ["scan_interlock"] mock_client.queue.remove_queue_lock.assert_called_with( queue="primary", lock_id="ScanInterlockActor" ) + + def test_unlock_skips_notify_without_lock(self, actor, mock_client): + mock_client.queue.queue_storage.current_scan_queue = {"primary": MagicMock(locks=[])} + + actor._unlock() + + mock_client.connector.notify.assert_not_called() + mock_client.queue.remove_queue_lock.assert_not_called() diff --git a/bec_server/tests/tests_scihub/test_scihub_cli_launch.py b/bec_server/tests/tests_scihub/test_scihub_cli_launch.py index c8ca81ebf..cd0af333b 100644 --- a/bec_server/tests/tests_scihub/test_scihub_cli_launch.py +++ b/bec_server/tests/tests_scihub/test_scihub_cli_launch.py @@ -1,6 +1,9 @@ from unittest import mock +from bec_lib.service_config import ServiceConfig +from bec_lib.tests.utils import ConnectorMock from bec_server.scihub.cli.launch import main +from bec_server.scihub.scihub import SciHub def test_main(): @@ -27,3 +30,28 @@ def test_main_shutdown(): mock_scihub.assert_called_once() mock_event.assert_called_once() mock_scihub.return_value.shutdown.assert_called_once() + + +def test_scihub_starts_and_stops_messaging_manager(): + config = ServiceConfig( + redis={"host": "dummy", "port": 6379}, + service_config={ + "file_writer": {"plugin": "default_NeXus_format", "base_path": "./"}, + "log_writer": {"base_path": "./"}, + }, + ) + + with ( + mock.patch.object(SciHub, "_start_metrics_emitter"), + mock.patch.object(SciHub, "wait_for_service"), + mock.patch("bec_server.scihub.scihub.AtlasConnector"), + mock.patch("bec_server.scihub.scihub.ServiceHandler"), + mock.patch("bec_server.scihub.scihub.MessagingManager") as mock_messaging_manager, + ): + scihub = SciHub(config, ConnectorMock) + try: + mock_messaging_manager.assert_called_once_with(scihub.connector) + finally: + scihub.shutdown() + + mock_messaging_manager.return_value.shutdown.assert_called_once()