diff --git a/lib/galaxy/authnz/psa_authnz.py b/lib/galaxy/authnz/psa_authnz.py index 5d1189b3837..75247adc4ce 100644 --- a/lib/galaxy/authnz/psa_authnz.py +++ b/lib/galaxy/authnz/psa_authnz.py @@ -798,8 +798,6 @@ def _send_oidc_profile_update_notification(trans, user, updates: list[str]) -> N NotificationVariant, PersonalNotificationCategory, ) - from galaxy.webapps.galaxy.services.notifications import NotificationService - labels: dict[str, str] = { "email": "email address", "username": "public name", @@ -819,7 +817,7 @@ def _send_oidc_profile_update_notification(trans, user, updates: list[str]) -> N ), galaxy_url=None, ) - NotificationService(trans.app.notification_manager).send_notification_internal(request, force_sync=True) + trans.app.notification_manager.send_notification_internal(request, force_sync=True) except Exception as exc: log.warning("OIDC profile update notification failed for user %s: %s", user.id, exc) diff --git a/lib/galaxy/celery/helpers.py b/lib/galaxy/celery/helpers.py new file mode 100644 index 00000000000..e771e340cb1 --- /dev/null +++ b/lib/galaxy/celery/helpers.py @@ -0,0 +1,25 @@ +from celery.result import AsyncResult + +from galaxy.schema.schema import AsyncTaskResultSummary + + +def async_task_summary(async_result: AsyncResult) -> AsyncTaskResultSummary: + name = None + try: + name = async_result.name + except AttributeError: + # if backend is disabled, we won't have this + pass + queue = None + try: + queue = async_result.queue + except AttributeError: + # if backend is disabled, we won't have this + pass + + return AsyncTaskResultSummary( + id=str(async_result.id), + ignored=async_result.ignored, + name=name, + queue=queue, + ) diff --git a/lib/galaxy/managers/notification.py b/lib/galaxy/managers/notification.py index 081fad7b7aa..67666034253 100644 --- a/lib/galaxy/managers/notification.py +++ b/lib/galaxy/managers/notification.py @@ -5,6 +5,7 @@ from typing import ( cast, NamedTuple, Optional, + Union, ) from urllib.parse import urlparse @@ -28,6 +29,7 @@ from sqlalchemy.sql import Select from typing_extensions import Protocol from galaxy import util +from galaxy.celery.helpers import async_task_summary from galaxy.config import ( GalaxyAppConfiguration, templates, @@ -57,6 +59,7 @@ from galaxy.schema.notifications import ( NotificationBroadcastUpdateRequest, NotificationCategorySettings, NotificationChannelSettings, + NotificationCreatedResponse, NotificationCreateData, NotificationCreateRequest, NotificationRecipients, @@ -67,6 +70,7 @@ from galaxy.schema.notifications import ( UserNotificationPreferences, UserNotificationUpdateRequest, ) +from galaxy.schema.schema import AsyncTaskResultSummary log = logging.getLogger(__name__) @@ -179,6 +183,30 @@ class NotificationManager: return notification, notifications_sent + def send_notification_internal( + self, request: NotificationCreateRequest, force_sync: bool = False + ) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]: + """Sends a notification to a list of recipients (users, groups or roles). + + If `force_sync` is set to `True`, the notification recipients will be processed synchronously instead of + in a background task. + + Note: This function is meant for internal use from other callers that don't need to check sender permissions. + """ + if self.can_send_notifications_async and not force_sync: + # Local import: galaxy.celery.tasks imports NotificationManager at module load, + # so importing it at module level here would be a circular dependency. + from galaxy.celery.tasks import send_notification_to_recipients_async + + result = send_notification_to_recipients_async.delay(request) + return async_task_summary(result) + + notification, recipient_user_count = self.send_notification_to_recipients(request) + return NotificationCreatedResponse( + total_notifications_sent=recipient_user_count, + notification=NotificationResponse.model_validate(notification), + ) + def _create_associations(self, notification: Notification, users: list[User]) -> int: success_count = 0 for user in users: diff --git a/lib/galaxy/webapps/galaxy/services/base.py b/lib/galaxy/webapps/galaxy/services/base.py index 5e096c7a619..d485d277d5b 100644 --- a/lib/galaxy/webapps/galaxy/services/base.py +++ b/lib/galaxy/webapps/galaxy/services/base.py @@ -7,8 +7,7 @@ from typing import ( Optional, ) -from celery.result import AsyncResult - +from galaxy.celery.helpers import async_task_summary as async_task_summary # re-export for existing callers from galaxy.exceptions import ( AuthenticationRequired, ConfigDoesNotAllowException, @@ -32,7 +31,6 @@ from galaxy.model.store import ( ) from galaxy.schema.fields import EncodedDatabaseIdField from galaxy.schema.schema import ( - AsyncTaskResultSummary, ToolRequestDetailedModel, ToolRequestModel, ) @@ -185,28 +183,6 @@ class ConsumesModelStores: ) -def async_task_summary(async_result: AsyncResult) -> AsyncTaskResultSummary: - name = None - try: - name = async_result.name - except AttributeError: - # if backend is disabled, we won't have this - pass - queue = None - try: - queue = async_result.queue - except AttributeError: - # if backend is disabled, we won't have this - pass - - return AsyncTaskResultSummary( - id=str(async_result.id), - ignored=async_result.ignored, - name=name, - queue=queue, - ) - - def _encode_tool_request(tool_request: ToolRequest, security: IdEncodingHelper) -> dict[str, Any]: """Encode request IDs using strongly-typed parameter walking.""" tool_source_model = tool_request.tool_source diff --git a/lib/galaxy/webapps/galaxy/services/notifications.py b/lib/galaxy/webapps/galaxy/services/notifications.py index b040e2b908e..046f5d1718a 100644 --- a/lib/galaxy/webapps/galaxy/services/notifications.py +++ b/lib/galaxy/webapps/galaxy/services/notifications.py @@ -6,7 +6,6 @@ from typing import ( Union, ) -from galaxy.celery.tasks import send_notification_to_recipients_async from galaxy.exceptions import ( AdminRequiredException, AuthenticationRequired, @@ -43,10 +42,7 @@ from galaxy.schema.notifications import ( UserNotificationUpdateRequest, ) from galaxy.schema.schema import AsyncTaskResultSummary -from galaxy.webapps.galaxy.services.base import ( - async_task_summary, - ServiceBase, -) +from galaxy.webapps.galaxy.services.base import ServiceBase class NotificationService(ServiceBase): @@ -87,28 +83,7 @@ class NotificationService(ServiceBase): recipients=payload.recipients, galaxy_url=galaxy_url, ) - return self.send_notification_internal(request) - - def send_notification_internal( - self, request: NotificationCreateRequest, force_sync: bool = False - ) -> Union[NotificationCreatedResponse, AsyncTaskResultSummary]: - """Sends a notification to a list of recipients (users, groups or roles). - - If `force_sync` is set to `True`, the notification recipients will be processed synchronously instead of - in a background task. - - Note: This function is meant for internal use from other services that don't need to check sender permissions. - """ - if self.notification_manager.can_send_notifications_async and not force_sync: - result = send_notification_to_recipients_async.delay(request) - summary = async_task_summary(result) - return summary - - notification, recipient_user_count = self.notification_manager.send_notification_to_recipients(request) - return NotificationCreatedResponse( - total_notifications_sent=recipient_user_count, - notification=NotificationResponse.model_validate(notification), - ) + return self.notification_manager.send_notification_internal(request) def broadcast( self, sender_context: ProvidesUserContext, payload: BroadcastNotificationCreateRequest diff --git a/lib/galaxy/webapps/galaxy/services/sharable.py b/lib/galaxy/webapps/galaxy/services/sharable.py index 33697adf51f..2fc7174ae06 100644 --- a/lib/galaxy/webapps/galaxy/services/sharable.py +++ b/lib/galaxy/webapps/galaxy/services/sharable.py @@ -187,7 +187,7 @@ class ShareableService: ) # We can set force_sync=True here because we already have the set of users to notify # and there is no need to resolve them asynchronously as no groups or roles are involved. - self.notification_service.send_notification_internal(request, force_sync=True) + self.notification_service.notification_manager.send_notification_internal(request, force_sync=True) class SharedItemNotificationFactory: diff --git a/test/integration/test_history_sse.py b/test/integration/test_history_sse.py index 4c6d3cc8e12..18b17226e5a 100644 --- a/test/integration/test_history_sse.py +++ b/test/integration/test_history_sse.py @@ -38,7 +38,9 @@ class TestHistorySSEIntegration(IntegrationTestCase): """The history_update event should contain the history's encoded ID.""" history_id = self._create_history() - listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key) + api_key = self.galaxy_interactor.api_key + assert api_key is not None + listener = SSELineListener(self._events_stream_url(), api_key) listener.start() try: self.dataset_populator.new_dataset(history_id, wait=False) @@ -61,7 +63,9 @@ class TestHistorySSEIntegration(IntegrationTestCase): user_a_history_id = self._create_history() - listener = SSELineListener(self._events_stream_url(), self.galaxy_interactor.api_key) + api_key = self.galaxy_interactor.api_key + assert api_key is not None + listener = SSELineListener(self._events_stream_url(), api_key) listener.start() try: # User B creates a history and uploads to it. User A must NOT see this. diff --git a/test/integration/test_notification_sse.py b/test/integration/test_notification_sse.py index 1f734061898..ecfc5c34f2a 100644 --- a/test/integration/test_notification_sse.py +++ b/test/integration/test_notification_sse.py @@ -94,7 +94,9 @@ class TestNotificationSSEIntegration(IntegrationTestCase): def test_sse_receives_broadcast_events(self): """When a broadcast is created, the SSE stream should receive it.""" - listener = SSELineListener(self._stream_url(), self.galaxy_interactor.api_key) + api_key = self.galaxy_interactor.api_key + assert api_key is not None + listener = SSELineListener(self._stream_url(), api_key) listener.start() try: subject = f"broadcast_sse_test_{uuid4()}" diff --git a/test/integration_selenium/test_notification_sse.py b/test/integration_selenium/test_notification_sse.py index 70ad646c0ca..715898bbd2e 100644 --- a/test/integration_selenium/test_notification_sse.py +++ b/test/integration_selenium/test_notification_sse.py @@ -94,7 +94,7 @@ class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): # the UI still shows the notification, polling picked it up — a silent # regression this assertion catches. self._wait_for_sse_event_after(baseline_ts) - self.driver.wait_for_selector_visible(f"text={subject}", timeout=SSE_EVENT_TIMEOUT_SECONDS * 1000) + self.wait_for_selector_visible(f"text={subject}", timeout=SSE_EVENT_TIMEOUT_SECONDS * 1000) self.screenshot("notification_sse_after") @selenium_test @@ -129,7 +129,7 @@ class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): self._wait_for_sse_event_after(baseline_ts) # The indicator dot should appear on the bell (within the #activity-notifications element) - self.driver.wait_for_selector_visible( + self.wait_for_selector_visible( "#activity-notifications .indicator", timeout=SSE_EVENT_TIMEOUT_SECONDS * 1000 ) self.screenshot("notification_bell_indicator") diff --git a/test/unit/authnz/test_psa_authnz.py b/test/unit/authnz/test_psa_authnz.py index 6bd2429d0eb..994d4aea12e 100644 --- a/test/unit/authnz/test_psa_authnz.py +++ b/test/unit/authnz/test_psa_authnz.py @@ -9,10 +9,7 @@ from datetime import ( ) from types import SimpleNamespace from typing import Optional -from unittest.mock import ( - MagicMock, - patch, -) +from unittest.mock import MagicMock import jwt import pytest @@ -400,15 +397,16 @@ def test_oidc_config_custom_auth_pipeline_and_extra(mock_oidc_config_file, mock_ def test_sync_user_profile_skips_when_account_interface_enabled(): manager = MagicMock() session = MagicMock() + notify = MagicMock() app_config = SimpleNamespace(enable_account_interface=True, enable_notification_system=True) - app = SimpleNamespace(config=app_config, user_manager=manager, notification_manager=SimpleNamespace()) + notification_manager = SimpleNamespace(send_notification_internal=notify) + app = SimpleNamespace(config=app_config, user_manager=manager, notification_manager=notification_manager) trans = SimpleNamespace(app=app, sa_session=session) strategy = SimpleNamespace(config={"GALAXY_TRANS": trans, "FIXED_DELEGATED_AUTH": True}) user = SimpleNamespace(id=1, preferences={}) details = {"email": "new@example.com", "username": "newname"} - with patch("galaxy.webapps.galaxy.services.notifications.NotificationService.send_notification_internal") as notify: - sync_user_profile(strategy=strategy, details=details, user=user) + sync_user_profile(strategy=strategy, details=details, user=user) manager.update_email.assert_not_called() manager.update_username.assert_not_called() @@ -419,15 +417,16 @@ def test_sync_user_profile_skips_when_account_interface_enabled(): def test_sync_user_profile_skips_when_fixed_delegated_auth_disabled(): manager = MagicMock() session = MagicMock() + notify = MagicMock() app_config = SimpleNamespace(enable_account_interface=False, enable_notification_system=True) - app = SimpleNamespace(config=app_config, user_manager=manager, notification_manager=SimpleNamespace()) + notification_manager = SimpleNamespace(send_notification_internal=notify) + app = SimpleNamespace(config=app_config, user_manager=manager, notification_manager=notification_manager) trans = SimpleNamespace(app=app, sa_session=session) strategy = SimpleNamespace(config={"GALAXY_TRANS": trans, "FIXED_DELEGATED_AUTH": False}) user = SimpleNamespace(id=2, email="old@example.com", username="oldname", preferences={}) details = {"email": "new@example.com", "username": "newname"} - with patch("galaxy.webapps.galaxy.services.notifications.NotificationService.send_notification_internal") as notify: - sync_user_profile(strategy=strategy, details=details, user=user) + sync_user_profile(strategy=strategy, details=details, user=user) manager.update_email.assert_not_called() manager.update_username.assert_not_called() @@ -438,16 +437,16 @@ def test_sync_user_profile_skips_when_fixed_delegated_auth_disabled(): def test_sync_user_profile_updates_when_account_interface_disabled(): manager = MagicMock() session = MagicMock() + notify = MagicMock() app_config = SimpleNamespace(enable_account_interface=False, enable_notification_system=True) - notification_manager = SimpleNamespace(notifications_enabled=True) + notification_manager = SimpleNamespace(notifications_enabled=True, send_notification_internal=notify) app = SimpleNamespace(config=app_config, user_manager=manager, notification_manager=notification_manager) trans = SimpleNamespace(app=app, sa_session=session) strategy = SimpleNamespace(config={"GALAXY_TRANS": trans, "FIXED_DELEGATED_AUTH": True}) user = SimpleNamespace(id=2, email="old@example.com", username="oldname", preferences={}) details = {"email": "new@example.com", "username": "newname"} - with patch("galaxy.webapps.galaxy.services.notifications.NotificationService.send_notification_internal") as notify: - sync_user_profile(strategy=strategy, details=details, user=user) + sync_user_profile(strategy=strategy, details=details, user=user) manager.update_email.assert_called_once_with( trans, user, "new@example.com", commit=False, send_activation_email=False