From b0cf4f7058dd6b8f00a3edd83471cf83aaccea28 Mon Sep 17 00:00:00 2001
From: mvdbeek <m.vandenbeek@gmail.com>
Date: Fri, 24 Apr 2026 19:28:20 +0200
Subject: [PATCH] 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.
---
 lib/galaxy_test/base/sse.py               | 21 +++++++++++++++++++++
 test/integration/test_history_sse.py      | 10 ++++++++--
 test/integration/test_notification_sse.py |  2 +-
 3 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/lib/galaxy_test/base/sse.py b/lib/galaxy_test/base/sse.py
index be7b3052bf1..d752d59c2eb 100644
--- a/lib/galaxy_test/base/sse.py
+++ b/lib/galaxy_test/base/sse.py
@@ -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))
diff --git a/test/integration/test_history_sse.py b/test/integration/test_history_sse.py
index 18b17226e5a..b3bb3341c7d 100644
--- a/test/integration/test_history_sse.py
+++ b/test/integration/test_history_sse.py
@@ -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()
 
diff --git a/test/integration/test_notification_sse.py b/test/integration/test_notification_sse.py
index ecfc5c34f2a..7ec3bf7f2e5 100644
--- a/test/integration/test_notification_sse.py
+++ b/test/integration/test_notification_sse.py
@@ -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."""