diff --git a/lib/galaxy_test/driver/integration_util.py b/lib/galaxy_test/driver/integration_util.py
index c7bc18c971c..c15d3580ae8 100644
--- a/lib/galaxy_test/driver/integration_util.py
+++ b/lib/galaxy_test/driver/integration_util.py
@@ -22,8 +22,10 @@ from unittest import (
     skip,
     SkipTest,
 )
+from urllib.parse import urljoin
 
 import pytest
+import requests
 
 from galaxy.app import UniverseApplication
 from galaxy.tool_util.verify.test_data import TestDataResolver
@@ -238,6 +240,16 @@ class IntegrationInstance(UsesApiTestCaseMixin, UsesCeleryTasks):
         if not self._app.config.database_connection.startswith("post"):
             raise SkipTest("Test only valid for postgres")
 
+    def _decode_id(self, encoded_id: str) -> int:
+        """Decode an encoded API id to its raw int via the live app's security helper."""
+        return self._app.security.decode_id(encoded_id)
+
+    def _user_id_for_api_key(self, api_key: str) -> int:
+        """Return the raw integer ``User.id`` for the user owning ``api_key``."""
+        response = requests.get(urljoin(self.url, "api/users/current"), params={"key": api_key})
+        response.raise_for_status()
+        return self._decode_id(response.json()["id"])
+
     def _run_tool_test(self, *args, **kwargs):
         return self._test_driver.run_tool_test(*args, **kwargs)
 
diff --git a/test/integration/test_entry_point_sse.py b/test/integration/test_entry_point_sse.py
index ead6e24b10e..1929310a737 100644
--- a/test/integration/test_entry_point_sse.py
+++ b/test/integration/test_entry_point_sse.py
@@ -50,18 +50,6 @@ class TestEntryPointSSEIntegration(IntegrationTestCase):
     def _events_stream_url(self) -> str:
         return urljoin(self.url, "api/events/stream")
 
-    def _user_id_for_api_key(self, api_key: str) -> int:
-        """Return the integer ``User.id`` for the user owning ``api_key``."""
-        # The ``/api/users/current`` endpoint returns the encoded id; decode
-        # via the app's security helper so we get the raw int the job row
-        # needs.
-        import requests
-
-        response = requests.get(urljoin(self.url, "api/users/current"), params={"key": api_key})
-        response.raise_for_status()
-        encoded_id = response.json()["id"]
-        return self._app.security.decode_id(encoded_id)
-
     def _create_it_job_with_entry_point(self, user_id: int, tool_port: int = 8888) -> tuple[int, int]:
         """Create a minimal Job + unconfigured InteractiveToolEntryPoint row pair.
 
diff --git a/test/integration_selenium/test_entry_point_sse.py b/test/integration_selenium/test_entry_point_sse.py
index 11701ec59a4..102a61ad837 100644
--- a/test/integration_selenium/test_entry_point_sse.py
+++ b/test/integration_selenium/test_entry_point_sse.py
@@ -64,7 +64,7 @@ class TestEntryPointSSESeleniumIntegration(SeleniumIntegrationTestCase):
         # 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"])
+        user_id = self._decode_id(user_info["id"])
         sa_session = self._app.model.context
         job = Job()
         job.user_id = user_id