Fix mypy type errors

Author: nuwang

lib/galaxy/authnz/oidc.py

@@ -57,7 +57,7 @@ def auth_params(self, state=None):
         # Add PKCE parameters if enabled
         if self.PKCE_ENABLED:
             # Generate PKCE challenge
-            code_verifier, code_challenge = generate_pkce_pair(code_length=96)
+            code_verifier, code_challenge = generate_pkce_pair(96)
             params["code_challenge"] = code_challenge
             params["code_challenge_method"] = "S256"
             # Store verifier in session for later use

lib/galaxy/authnz/psa_authnz.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import time
+from typing import cast
 from urllib.parse import quote
 
 import jwt
@@ -20,6 +21,7 @@
 )
 from sqlalchemy import func
 from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
 
 from galaxy.exceptions import MalformedContents
 from galaxy.model import (
@@ -117,8 +119,6 @@
     # Update the user record with any changed info from the auth service.
     "social_core.pipeline.user.user_details",
     "galaxy.authnz.psa_authnz.decode_access_token",
-    # Set appropriate redirect URL based on context
-    "galaxy.authnz.psa_authnz.set_redirect_url",
 )
 
 DISCONNECT_PIPELINE = ("galaxy.authnz.psa_authnz.allowed_to_disconnect", "galaxy.authnz.psa_authnz.disconnect")
@@ -810,7 +810,10 @@ def associate_by_email_if_logged_in(
         return
 
     # Check if an account with this email exists
+    # sa_session is guaranteed to be set by on_the_fly_config() before pipeline runs
     sa_session = UserAuthnzToken.sa_session
+    if sa_session is None:
+        raise RuntimeError("sa_session must be set by on_the_fly_config before pipeline execution")
     existing_user = sa_session.query(User).where(func.lower(User.email) == email.lower()).first()
 
     if user:
@@ -872,7 +875,10 @@ def check_user_creation_confirmation(
         # Check if there's an existing user with this email
         email = details.get("email")
         if email:
+            # sa_session is guaranteed to be set by on_the_fly_config() before pipeline runs
             sa_session = UserAuthnzToken.sa_session
+            if sa_session is None:
+                raise RuntimeError("sa_session must be set by on_the_fly_config before pipeline execution")
             existing_user = sa_session.query(User).where(func.lower(User.email) == email.lower()).first()
 
             # If no existing user, redirect to confirmation page
@@ -900,18 +906,3 @@ def check_user_creation_confirmation(
 
     # Continue with user creation if confirmation is not required or user already exists
     return
-
-
-def set_redirect_url(strategy=None, backend=None, details=None, user=None, is_new=False, social=None, **kwargs):
-    """
-    Placeholder pipeline step for redirect URL handling.
-
-    The actual redirect URL logic is handled in the callback() method
-    based on the fixed_delegated_auth setting, where LOGIN_REDIRECT_URL
-    is set before calling do_complete().
-
-    This step exists to maintain compatibility with the pipeline but
-    doesn't need to do anything.
-    """
-    # No action needed - redirect URL is set in callback() method
-    return

lib/galaxy/model/__init__.py

@@ -10882,7 +10882,7 @@ class UserAuthnzToken(Base, UserMixin, RepresentById):
     )
 
     # This static property is set at: galaxy.authnz.psa_authnz.PSAAuthnz
-    sa_session = None
+    sa_session: ClassVar[Optional[Session]] = None
 
     def __init__(self, provider, uid, extra_data=None, lifetime=None, assoc_type=None, user=None):
         self.provider = provider

test/unit/authnz/test_oidc_backends.py

@@ -14,9 +14,14 @@
 )
 
 import pytest
+from social_core.utils import setting_name
 
 from galaxy.authnz.cilogon import CILogonOpenIdConnect
 from galaxy.authnz.keycloak import KeycloakOpenIdConnect
+from galaxy.authnz.psa_authnz import (
+    associate_by_email_if_logged_in,
+    check_user_creation_confirmation,
+)
 
 
 class MockStrategy:
@@ -506,11 +511,6 @@ def test_callback_user_not_created_when_does_not_exist(self):
         they should be redirected to a confirmation page instead of having their
         account created immediately.
         """
-        from galaxy.authnz.psa_authnz import (
-            check_user_creation_confirmation,
-            setting_name,
-        )
-
         # Setup strategy with require_create_confirmation enabled
         strategy = MockStrategy(
             {
@@ -680,11 +680,6 @@ class TestFixedDelegatedAuth:
 
     def test_fixed_delegated_auth_auto_associates_existing_user(self):
         """Test that fixed_delegated_auth automatically associates with existing user."""
-        from galaxy.authnz.psa_authnz import (
-            associate_by_email_if_logged_in,
-            setting_name,
-        )
-
         strategy = MockStrategy(
             {
                 "FIXED_DELEGATED_AUTH": True,
@@ -724,11 +719,6 @@ def test_fixed_delegated_auth_auto_associates_existing_user(self):
 
     def test_fixed_delegated_auth_continues_when_no_user(self):
         """Test that fixed_delegated_auth continues to user creation when no user exists."""
-        from galaxy.authnz.psa_authnz import (
-            associate_by_email_if_logged_in,
-            setting_name,
-        )
-
         strategy = MockStrategy(
             {
                 "FIXED_DELEGATED_AUTH": True,
@@ -764,11 +754,6 @@ def test_fixed_delegated_auth_continues_when_no_user(self):
 
     def test_without_fixed_delegated_auth_prompts_for_login(self):
         """Test that without fixed_delegated_auth, users are prompted to log in."""
-        from galaxy.authnz.psa_authnz import (
-            associate_by_email_if_logged_in,
-            setting_name,
-        )
-
         strategy = MockStrategy(
             {
                 "FIXED_DELEGATED_AUTH": False,
@@ -809,61 +794,5 @@ def test_without_fixed_delegated_auth_prompts_for_login(self):
             assert "connect_external_email=" in result
 
 
-class TestRedirectURL:
-    """Test the set_redirect_url pipeline step."""
-
-    def test_fixed_delegated_auth_redirects_to_root(self):
-        """Test that fixed_delegated_auth redirects to root URL."""
-        from galaxy.authnz.psa_authnz import (
-            set_redirect_url,
-            setting_name,
-        )
-
-        strategy = MockStrategy(
-            {
-                "FIXED_DELEGATED_AUTH": True,
-                setting_name("LOGIN_REDIRECT_URL"): "http://localhost:8080/",
-            }
-        )
-
-        backend = Mock()
-
-        set_redirect_url(
-            strategy=strategy,
-            backend=backend,
-            details={},
-            user=Mock(),
-        )
-
-        # Should set redirect to root URL
-        assert strategy.session.get("next") == "http://localhost:8080/"
-
-    def test_without_fixed_delegated_auth_redirects_to_external_ids(self):
-        """Test that without fixed_delegated_auth, redirect goes to user/external_ids."""
-        from galaxy.authnz.psa_authnz import (
-            set_redirect_url,
-            setting_name,
-        )
-
-        strategy = MockStrategy(
-            {
-                "FIXED_DELEGATED_AUTH": False,
-                setting_name("LOGIN_REDIRECT_URL"): "http://localhost:8080/",
-            }
-        )
-
-        backend = Mock()
-
-        set_redirect_url(
-            strategy=strategy,
-            backend=backend,
-            details={},
-            user=Mock(),
-        )
-
-        # Should set redirect to user/external_ids
-        assert strategy.session.get("next") == "http://localhost:8080/user/external_ids"
-
-
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])