Fix mypy errors on sse-notifications branch

Move send_notification_internal from NotificationService to
NotificationManager so non-web callers (psa_authnz, sharable service)
can dispatch notifications without constructing a web-layer service —
which is why the sse_manager constructor drift was caught by mypy in
the first place.

Relocate async_task_summary to galaxy.celery.helpers so the manager
can call it without reaching into galaxy.webapps.galaxy.services.base;
existing service callers keep working via an explicit re-export.

Also narrow Optional api_key before SSELineListener in the SSE
integration tests, and call wait_for_selector_visible on self rather
than self.driver in the selenium SSE test.
This commit is contained in:
mvdbeek
2026-04-28 17:17:17 +02:00
parent 079d4ce52d
commit 5ae948c1df
10 changed files with 81 additions and 74 deletions
+1 -3
View File
@@ -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)
+25
View File
@@ -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,
)
+28
View File
@@ -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:
+1 -25
View File
@@ -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
@@ -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
@@ -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:
+6 -2
View File
@@ -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.
+3 -1
View File
@@ -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()}"
@@ -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")
+12 -13
View File
@@ -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