From 7384f192b6f5fbfd897dae9f10fd71f146b29553 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 21 Apr 2026 10:26:38 +0200 Subject: [PATCH] Fix SSE selenium tests for Playwright backend and user identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three related fixes to make the SSE selenium tests pass end-to-end: - Swap ``self.driver.execute_script`` for ``self.execute_script`` so the tests work with both the Selenium and Playwright backends; the former raises ``NotImplementedError`` under Playwright. - Switch user-info lookup from ``self._get("users/current")`` (API interactor's default key) to ``self.api_get("users/current")`` (browser cookie). ``ensure_registered`` creates a distinct Selenium user, so the API-key user and the browser-SSE user were different and ``push_to_user`` landed in no queue. - Add ``interactivetools_enable`` and ``enable_sse_entry_point_updates`` to the entry-point test config — App.vue won't start watching entry points, and the store won't open an SSE connection, without both. - Use ``self.get(path)`` (goes through ``build_url``) instead of raw URL concatenation, and pass wait timeouts in seconds (not milliseconds); the timeout helper multiplies by 1000 internally. - Target ``.nav-indicator`` on the activity-bar item rather than a ``.indicator`` class that the bell component no longer renders. --- .../test_entry_point_sse.py | 15 ++++++++-- .../test_notification_sse.py | 29 ++++++++++++------- 2 files changed, 31 insertions(+), 13 deletions(-) diff --git a/test/integration_selenium/test_entry_point_sse.py b/test/integration_selenium/test_entry_point_sse.py index e661cb0a79f..e579c70a3ab 100644 --- a/test/integration_selenium/test_entry_point_sse.py +++ b/test/integration_selenium/test_entry_point_sse.py @@ -30,17 +30,23 @@ class TestEntryPointSSESeleniumIntegration(SeleniumIntegrationTestCase): def handle_galaxy_config_kwds(cls, config): super().handle_galaxy_config_kwds(config) config["enable_celery_tasks"] = False + # App.vue only calls entryPointStore.startWatchingEntryPoints() when + # interactivetools_enable is True, and the store only opens an SSE + # connection when enable_sse_entry_point_updates is True. Without both, + # __galaxy_sse_connected never becomes true and the gate below times out. + config["interactivetools_enable"] = True + config["enable_sse_entry_point_updates"] = True def _wait_for_sse_connected(self) -> None: """Block until the frontend confirms the SSE pipeline is live.""" wait_on( - lambda: True if self.driver.execute_script("return window.__galaxy_sse_connected === true") else None, + lambda: True if self.execute_script("return window.__galaxy_sse_connected === true") else None, "window.__galaxy_sse_connected === true", timeout=SSE_CONNECT_TIMEOUT_SECONDS, ) def _last_sse_event_ts(self) -> int: - return self.driver.execute_script("return window.__galaxy_sse_last_event_ts || 0") or 0 + return self.execute_script("return window.__galaxy_sse_last_event_ts || 0") or 0 def _wait_for_sse_event_after(self, baseline_ts: int) -> None: wait_on( @@ -55,7 +61,10 @@ class TestEntryPointSSESeleniumIntegration(SeleniumIntegrationTestCase): Job, ) - user_info = self._get("users/current").json() + # Use the browser's cookie-authenticated user, not the API interactor's + # default: SSE connects under the Selenium-registered user, and the + # dispatch's user_id must match or push_to_user finds no queues. + user_info = self.api_get("users/current") user_id = self._app.security.decode_id(user_info["id"]) sa_session = self._app.model.context job = Job() diff --git a/test/integration_selenium/test_notification_sse.py b/test/integration_selenium/test_notification_sse.py index cae3b6e7b6b..d1902d2b1be 100644 --- a/test/integration_selenium/test_notification_sse.py +++ b/test/integration_selenium/test_notification_sse.py @@ -35,14 +35,14 @@ class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): test would falsely pass. """ wait_on( - lambda: True if self.driver.execute_script("return window.__galaxy_sse_connected === true") else None, + lambda: True if self.execute_script("return window.__galaxy_sse_connected === true") else None, "window.__galaxy_sse_connected === true", timeout=SSE_CONNECT_TIMEOUT_SECONDS, ) def _last_sse_event_ts(self) -> int: """Return the last SSE event timestamp recorded by the composable, or 0.""" - return self.driver.execute_script("return window.__galaxy_sse_last_event_ts || 0") or 0 + return self.execute_script("return window.__galaxy_sse_last_event_ts || 0") or 0 def _wait_for_sse_event_after(self, baseline_ts: int) -> None: """Block until an SSE event arrives after ``baseline_ts``. @@ -61,12 +61,18 @@ class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): @managed_history def test_notification_appears_via_sse(self): """Send a notification via the API and verify it appears in the UI without refresh.""" - # Get the logged-in user's info so we can send a notification to them - user_info = self._get("users/current").json() + # Get the browser-logged-in user's info via the browser's cookie. ``self._get`` + # uses the API interactor's default-user key, which does not match the Selenium + # user created by ``ensure_registered``, so the SSE push would target a + # different user than the one watching the stream. + user_info = self.api_get("users/current") user_id = user_info["id"] - # Navigate to notifications page so the store is watching - self.driver.get(f"{self.target_url_from_selenium}/user/notifications") + # Navigate to notifications page so the store is watching. + # ``get()`` uses ``build_url()`` which handles trailing slashes on the + # base correctly; concatenating against ``target_url_from_selenium`` + # can produce a double-slash that Galaxy routes differently. + self.get("user/notifications") self._wait_for_sse_connected() baseline_ts = self._last_sse_event_ts() self.screenshot("notification_sse_before") @@ -94,14 +100,16 @@ 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.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) self.screenshot("notification_sse_after") @selenium_test @managed_history def test_notification_bell_updates_via_sse(self): """The notification bell indicator should update when a new notification arrives via SSE.""" - user_info = self._get("users/current").json() + # See ``test_notification_appears_via_sse`` — must use the browser's user, + # not the API interactor's default user. + user_info = self.api_get("users/current") user_id = user_info["id"] # Go to home page (bell is in masthead) @@ -128,6 +136,7 @@ class TestNotificationSSESeleniumIntegration(SeleniumIntegrationTestCase): self._assert_status_code_is_ok(response) self._wait_for_sse_event_after(baseline_ts) - # The indicator dot should appear on the bell (within the #activity-notifications element) - self.wait_for_selector_visible("#activity-notifications .indicator", timeout=SSE_EVENT_TIMEOUT_SECONDS * 1000) + # The activity-bar notifications item renders its unread-count badge as + # ``.nav-indicator`` (see ``ActivityItem.vue``) once ``totalUnreadCount > 0``. + self.wait_for_selector_visible("#activity-notifications .nav-indicator", timeout=SSE_EVENT_TIMEOUT_SECONDS) self.screenshot("notification_bell_indicator")