Add integration tests for the Pulsar ARC job runner

These tests verify that the ARC endpoint URL and OIDC provider selection logic works correctly when queuing jobs. They
do not test actual job execution, as that is covered by Pulsar's own tests.

Author: kysrpex

test/integration/job_resource_rules/test_pulsar_arc.py

@@ -0,0 +1,50 @@
+"""Dynamic job rule for the Pulsar ARC (Advanced Resource Connector) job runner integration tests.
+
+This file contains a dynamic job rule for the Pulsar ARC job runner integration tests in
+`test/integration/test_pulsar_arc.py`. The rule allows selecting a job destination based on a dynamic reference that can
+be read and updated during the tests using `test_pulsar_arc_get_job_destination()` and
+`test_pulsar_arc_set_job_destination()` respectively.
+"""
+
+from typing import Optional
+
+from galaxy.app import UniverseApplication
+from galaxy.jobs import JobMappingException  # type: ignore[attr-defined]
+from galaxy.jobs import (
+    JobDestination,
+)
+from galaxy.model import (
+    Job,
+    User as GalaxyUser,
+)
+from galaxy.tools import Tool as GalaxyTool
+
+__all__ = ("test_pulsar_arc", "test_pulsar_arc_get_job_destination", "test_pulsar_arc_set_job_destination")
+
+
+test_pulsar_arc_job_destination_ref: list[JobDestination] = []
+
+
+def test_pulsar_arc_get_job_destination() -> Optional[JobDestination]:
+    """
+    Return the last stored job destination (if any).
+    """
+    return test_pulsar_arc_job_destination_ref[0] if test_pulsar_arc_job_destination_ref else None
+
+
+def test_pulsar_arc_set_job_destination(job_destination: JobDestination) -> None:
+    """
+    Store a job destination, overwriting any previous one.
+    """
+    test_pulsar_arc_job_destination_ref[:] = [job_destination]
+
+
+def test_pulsar_arc(app: UniverseApplication, job: Job, tool: GalaxyTool, user: GalaxyUser) -> JobDestination:
+    """
+    Dynamic job rule that maps jobs to the stored job destination.
+    """
+    job_destination = test_pulsar_arc_get_job_destination()
+    if not job_destination:
+        raise JobMappingException("No job destination set for dynamic job rule.")
+
+    return job_destination

test/integration/oidc/test_auth_oidc.py

@@ -93,6 +93,9 @@ class BaseKeycloakIntegrationTestCase(integration_util.IntegrationTestCase):
         backend_config_file: ClassVar[str]
         saved_oauthlib_insecure_transport: ClassVar[bool]
 
+        REGEX_KEYCLOAK_LOGIN_ACTION = re.compile(r"action=\"(.*)\"\s+")
+        REGEX_GALAXY_CSRF_TOKEN = re.compile(r"session_csrf_token\": \"(.*)\"")
+
         @classmethod
         def setUpClass(cls):
             # By default, the oidc callback must be done over a secure transport, so
@@ -158,40 +161,38 @@ def handle_galaxy_oidc_config_kwds(cls, config):
         def _get_interactor(self, api_key=None, allow_anonymous=False) -> "ApiTestInteractor":
             return super()._get_interactor(api_key=None, allow_anonymous=True)
 
+        def _login_via_keycloak(self, username, password, expected_codes=None, save_cookies=False, session=None):
+            if expected_codes is None:
+                expected_codes = [200, 404]
+            session = session or requests.Session()
+            response = session.get(f"{self.url}authnz/keycloak/login")
+            provider_url = response.json()["redirect_uri"]
+            response = session.get(provider_url, verify=False)
+            matches = self.REGEX_KEYCLOAK_LOGIN_ACTION.search(response.text)
+            assert matches
+            auth_url = html.unescape(str(matches.groups(1)[0]))
+            response = session.post(auth_url, data={"username": username, "password": password}, verify=False)
+            assert response.status_code in expected_codes, response
+            if save_cookies:
+                self.galaxy_interactor.cookies = session.cookies
+            return session, response
+
+        def _get_keycloak_access_token(
+            self, client_id="gxyclient", username=KEYCLOAK_TEST_USERNAME, password=KEYCLOAK_TEST_PASSWORD, scopes=None
+        ):
+            data = {
+                "client_id": client_id,
+                "client_secret": "dummyclientsecret",
+                "grant_type": "password",
+                "username": username,
+                "password": password,
+                "scope": scopes or [],
+            }
+            response = requests.post(f"{KEYCLOAK_URL}/protocol/openid-connect/token", data=data, verify=False)
+            return response.json()["access_token"]
+
 
 class TestGalaxyOIDCLoginIntegration(AbstractTestCases.BaseKeycloakIntegrationTestCase):
-    REGEX_KEYCLOAK_LOGIN_ACTION = re.compile(r"action=\"(.*)\"\s+")
-    REGEX_GALAXY_CSRF_TOKEN = re.compile(r"session_csrf_token\": \"(.*)\"")
-
-    def _login_via_keycloak(self, username, password, expected_codes=None, save_cookies=False, session=None):
-        if expected_codes is None:
-            expected_codes = [200, 404]
-        session = session or requests.Session()
-        response = session.get(f"{self.url}authnz/keycloak/login")
-        provider_url = response.json()["redirect_uri"]
-        response = session.get(provider_url, verify=False)
-        matches = self.REGEX_KEYCLOAK_LOGIN_ACTION.search(response.text)
-        assert matches
-        auth_url = html.unescape(str(matches.groups(1)[0]))
-        response = session.post(auth_url, data={"username": username, "password": password}, verify=False)
-        assert response.status_code in expected_codes, response
-        if save_cookies:
-            self.galaxy_interactor.cookies = session.cookies
-        return session, response
-
-    def _get_keycloak_access_token(
-        self, client_id="gxyclient", username=KEYCLOAK_TEST_USERNAME, password=KEYCLOAK_TEST_PASSWORD, scopes=None
-    ):
-        data = {
-            "client_id": client_id,
-            "client_secret": "dummyclientsecret",
-            "grant_type": "password",
-            "username": username,
-            "password": password,
-            "scope": scopes or [],
-        }
-        response = requests.post(f"{KEYCLOAK_URL}/protocol/openid-connect/token", data=data, verify=False)
-        return response.json()["access_token"]
 
     def test_oidc_login_new_user(self):
         _, response = self._login_via_keycloak(KEYCLOAK_TEST_USERNAME, KEYCLOAK_TEST_PASSWORD, save_cookies=True)