Refactor method names and lint fixes

lib/galaxy/authnz/__init__.py

@@ -87,3 +87,16 @@ def logout(self, trans, post_user_logout_href=None):
         :param post_user_logout_href: Optional URL to redirect to after logging out of IDP.
         """
         raise NotImplementedError()
+
+    def find_user_by_access_token(self, sa_session, access_token):
+        """
+        Locates a user by access_token. The access token must be verified prior
+        to returning the relevant user.
+
+        :type  sa_session:      sqlalchemy.orm.scoping.scoped_session
+        :param sa_session:      SQLAlchemy database handle.
+
+        :type  access_token: string
+        :param access_token: An OIDC access token
+        """
+        raise NotImplementedError()

lib/galaxy/authnz/custos_authnz.py

@@ -501,7 +501,7 @@ def _username_from_userinfo(trans, userinfo):
         else:
             return username
 
-    def match_access_token_to_user(self, sa_session, access_token):
+    def find_user_by_access_token(self, sa_session, access_token):
         signing_key = self.jwks_client.get_signing_key_from_jwt(access_token)
         decoded_jwt = jwt.decode(
             access_token,

lib/galaxy/authnz/managers.py

@@ -408,26 +408,28 @@ def create_user(self, provider, token, trans, login_redirect_url):
             log.exception(msg)
             return False, msg, (None, None)
 
-    def find_user_by_access_token_in_provider(self, sa_session, provider, access_token):
+    def _find_user_by_access_token_in_provider(self, sa_session, provider, access_token):
         try:
             success, message, backend = self._get_authnz_backend(provider)
             if success is False:
                 msg = f"An error occurred when obtaining user by token with provider `{provider}`: {message}"
                 log.error(msg)
                 return None
-            user = backend.match_access_token_to_user(sa_session, access_token)
+            user = backend.find_user_by_access_token(sa_session, access_token)
             if user:
                 log.debug(f"Found user: {user} via `{provider}` identity provider")
                 return user
             return None
+        except NotImplementedError:
+            return None
         except Exception as e:
             msg = f"An error occurred with provider: {provider} when finding user by token: {e}"
             log.error(msg)
             return None
 
-    def find_user_by_access_token(self, sa_session, access_token):
+    def match_access_token_to_user(self, sa_session, access_token):
         for provider in self.oidc_backends_config:
-            user = self.find_user_by_access_token_in_provider(sa_session, provider, access_token)
+            user = self._find_user_by_access_token_in_provider(sa_session, provider, access_token)
             if user:
                 return user
         return None

lib/galaxy/managers/users.py

@@ -297,7 +297,7 @@ def by_api_key(self, api_key: str, sa_session=None):
 
     def by_oidc_access_token(self, access_token: str):
         if hasattr(self.app, "authnz_manager") and self.app.authnz_manager:
-            user = self.app.authnz_manager.find_user_by_access_token(self.app.model.session, access_token)  # type: ignore[attr-defined]
+            user = self.app.authnz_manager.match_access_token_to_user(self.app.model.session, access_token)  # type: ignore[attr-defined]
             return user
         else:
             return None

test/integration/oidc/test_auth_oidc.py

@@ -119,12 +119,11 @@ def generate_oidc_config_file(cls, server_wrapper):
 
         @classmethod
         def configure_oidc_and_restart(cls):
-            with tempfile.NamedTemporaryFile("w+t", delete=False) as tmp_file:
-                server_wrapper = cls._test_driver.server_wrappers[0]
-                cls.backend_config_file = cls.generate_oidc_config_file(server_wrapper)
-                # Explicitly assign the previously used port, as it's random otherwise
-                del os.environ["GALAXY_TEST_PORT_RANDOM"]
-                os.environ["GALAXY_TEST_PORT"] = os.environ["GALAXY_WEB_PORT"]
+            server_wrapper = cls._test_driver.server_wrappers[0]
+            cls.backend_config_file = cls.generate_oidc_config_file(server_wrapper)
+            # Explicitly assign the previously used port, as it's random otherwise
+            del os.environ["GALAXY_TEST_PORT_RANDOM"]
+            os.environ["GALAXY_TEST_PORT"] = os.environ["GALAXY_WEB_PORT"]
             cls._test_driver.restart(config_object=cls, handle_config=cls.handle_galaxy_oidc_config_kwds)
 
         @classmethod
@@ -166,32 +165,33 @@ def _login_via_keycloak(
         self,
         username,
         password,
-        expected_codes=[200, 404],
+        expected_codes=None,
         save_cookies=False,
     ):
+        if expected_codes is None:
+            expected_codes = [200, 404]
         session = 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)
         auth_url = html.unescape(matches.groups(1)[0])
         response = session.post(auth_url, data={"username": username, "password": password}, verify=False)
-        if expected_codes:
-            assert response.status_code in expected_codes, response
+        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=[]
+        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,
+            "scope": scopes or [],
         }
         response = requests.post(f"{KEYCLOAK_URL}/protocol/openid-connect/token", data=data, verify=False)
         return response.json()["access_token"]