mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Fix failing SSE integration tests
- test_notification_sse: connect to the unified /api/events/stream rather
than the deleted /api/notifications/stream. The old URL was matching
FastAPI's /api/notifications/{notification_id} route and failing the
encoded-id length check with HTTP 400.
- test_history_sse: add SSELineListener.wait_for_event_where(predicate)
and use it in both history-update tests. wait_for_event returned after
the first history_update of any kind, so a background update for an
unrelated history could land before the test's own upload dispatched,
causing the assertion to fail even though the event was in flight.
This commit is contained in:
@@ -10,6 +10,7 @@ thread instead of silently swallowing them.
|
||||
import queue
|
||||
import threading
|
||||
from typing import (
|
||||
Callable,
|
||||
Optional,
|
||||
)
|
||||
|
||||
@@ -97,6 +98,26 @@ class SSELineListener:
|
||||
|
||||
return wait_on(_check, f"SSE {event_type} event", timeout=timeout)
|
||||
|
||||
def wait_for_event_where(
|
||||
self,
|
||||
event_type: str,
|
||||
predicate: Callable[[dict], bool],
|
||||
timeout: int = DEFAULT_WAIT_TIMEOUT,
|
||||
) -> list[dict]:
|
||||
"""Block until at least one ``event_type`` event matches ``predicate``.
|
||||
|
||||
Returns every ``event_type`` event observed so far, not just matches, so
|
||||
callers can still inspect the surrounding stream (e.g. assert what else
|
||||
did or didn't appear) after the wait resolves.
|
||||
"""
|
||||
|
||||
def _check():
|
||||
self._raise_if_errored()
|
||||
events = self.get_events(event_type)
|
||||
return events if any(predicate(e) for e in events) else None
|
||||
|
||||
return wait_on(_check, f"SSE {event_type} matching predicate", timeout=timeout)
|
||||
|
||||
def get_events(self, event_type: Optional[str] = None) -> list[dict]:
|
||||
"""Return all collected events so far, optionally filtered by type."""
|
||||
all_events = parse_sse_events("".join(self._collected))
|
||||
|
||||
@@ -44,7 +44,10 @@ class TestHistorySSEIntegration(IntegrationTestCase):
|
||||
listener.start()
|
||||
try:
|
||||
self.dataset_populator.new_dataset(history_id, wait=False)
|
||||
history_events = listener.wait_for_event("history_update")
|
||||
history_events = listener.wait_for_event_where(
|
||||
"history_update",
|
||||
lambda e: history_id in json.loads(e["data"]).get("history_ids", []),
|
||||
)
|
||||
found = any(history_id in json.loads(e["data"]).get("history_ids", []) for e in history_events)
|
||||
assert found, f"Expected history_id '{history_id}' in history_update events, got: {history_events}"
|
||||
finally:
|
||||
@@ -85,7 +88,10 @@ class TestHistorySSEIntegration(IntegrationTestCase):
|
||||
|
||||
# User A uploads to their own history — this is what A's stream must observe.
|
||||
self.dataset_populator.new_dataset(user_a_history_id, wait=False)
|
||||
history_events = listener.wait_for_event("history_update")
|
||||
history_events = listener.wait_for_event_where(
|
||||
"history_update",
|
||||
lambda e: user_a_history_id in json.loads(e["data"]).get("history_ids", []),
|
||||
)
|
||||
finally:
|
||||
listener.stop()
|
||||
|
||||
|
||||
@@ -66,7 +66,7 @@ class TestNotificationSSEIntegration(IntegrationTestCase):
|
||||
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
|
||||
|
||||
def _stream_url(self) -> str:
|
||||
return urljoin(self.url, "api/notifications/stream")
|
||||
return urljoin(self.url, "api/events/stream")
|
||||
|
||||
def test_sse_receives_notification_events(self):
|
||||
"""When a notification is created, the SSE stream should receive it."""
|
||||
|
||||
Reference in New Issue
Block a user