Remove unused imports and make format

Author: nuwang

lib/galaxy/authnz/oidc_utils.py

@@ -6,10 +6,8 @@
 """
 
 import logging
-from typing import Optional
 
 import jwt
-from jwt import InvalidTokenError
 from social_core.backends.open_id_connect import OpenIdConnectAuth
 
 from galaxy.exceptions import MalformedContents

lib/galaxy/authnz/psa_authnz.py

@@ -11,7 +11,6 @@
     do_complete,
     do_disconnect,
 )
-from social_core.backends.open_id_connect import OpenIdConnectAuth
 from social_core.backends.utils import get_backend
 from social_core.strategy import BaseStrategy
 from social_core.utils import (

lib/galaxy/model/migrations/data_fixes/custos_to_psa.py

@@ -1,12 +1,12 @@
 """Reusable helpers for migrating Custos authentication tokens into PSA format."""
 
-import jwt
 from datetime import datetime
 from typing import (
     cast,
     Optional,
 )
 
+import jwt
 from sqlalchemy import (
     Column,
     DateTime,
@@ -26,7 +26,7 @@
 CUSTOS_ASSOC_TYPE = "custos_migrated"
 
 
-def _extract_iat_from_token(token: str) -> Optional[int]:
+def _extract_iat_from_token(token: Optional[str]) -> Optional[int]:
     """
     Extract the 'iat' (issued at) claim from a JWT token.
     Returns None if the token cannot be decoded or doesn't have an iat claim.
@@ -110,11 +110,7 @@ def migrate_custos_tokens_to_psa(
         # Extract auth_time from token's 'iat' claim (issued at time)
         # Priority: access_token.iat > id_token.iat > current time
         # We prefer access_token.iat since that's the token whose expiration we're tracking
-        auth_time = (
-            _extract_iat_from_token(record.access_token)
-            or _extract_iat_from_token(record.id_token)
-            or now_ts
-        )
+        auth_time = _extract_iat_from_token(record.access_token) or _extract_iat_from_token(record.id_token) or now_ts
         extra_data["auth_time"] = auth_time
 
         # Calculate expires from expiration_time

test/unit/authnz/test_psa_authnz.py

@@ -197,7 +197,7 @@ def test_decode_access_token():
     mock_backend.strategy.config = {"accepted_audiences": dummy_access_token.access_token_data["aud"]}
     mock_backend.id_token_issuer.return_value = dummy_access_token.access_token_data["iss"]
     # Make isinstance() checks pass by setting __class__ after configuring the mock
-    mock_backend.__class__ = OpenIdConnectAuth
+    mock_backend.__class__ = OpenIdConnectAuth  # type: ignore[assignment]
     # Check that access token is decoded successfully to return the original data
     data = decode_access_token(social=mock_social, backend=mock_backend)
     assert data["access_token"] == dummy_access_token.access_token_data
@@ -219,7 +219,7 @@ def test_decode_access_token_invalid_key():
     mock_backend.strategy.config = {"accepted_audiences": dummy_access_token.access_token_data["aud"]}
     mock_backend.id_token_issuer.return_value = dummy_access_token.access_token_data["iss"]
     # Make isinstance() checks pass by setting __class__ after configuring the mock
-    mock_backend.__class__ = OpenIdConnectAuth
+    mock_backend.__class__ = OpenIdConnectAuth  # type: ignore[assignment]
     # Test that the decode function returns None for the access token
     result = decode_access_token(social=mock_social, backend=mock_backend)
     assert result["access_token"] is None
@@ -244,7 +244,7 @@ def test_decode_access_token_invalid_issuer():
     mock_backend.strategy.config = {"accepted_audiences": dummy_access_token.access_token_data["aud"]}
     mock_backend.id_token_issuer.return_value = "https://validissuer.com"
     # Make isinstance() checks pass by setting __class__ after configuring the mock
-    mock_backend.__class__ = OpenIdConnectAuth
+    mock_backend.__class__ = OpenIdConnectAuth  # type: ignore[assignment]
     # Test that the decode function returns None for the access token
     result = decode_access_token(social=mock_social, backend=mock_backend)
     assert result["access_token"] is None
@@ -269,7 +269,7 @@ def test_decode_access_token_invalid_audience():
     mock_backend.strategy.config = {"accepted_audiences": ["https://validaudience.url"]}
     mock_backend.id_token_issuer.return_value = dummy_access_token.access_token_data["iss"]
     # Make isinstance() checks pass by setting __class__ after configuring the mock
-    mock_backend.__class__ = OpenIdConnectAuth
+    mock_backend.__class__ = OpenIdConnectAuth  # type: ignore[assignment]
     # Test that the decode function returns None for the access token
     result = decode_access_token(social=mock_social, backend=mock_backend)
     assert result["access_token"] is None

test/unit/data/model/test_migrate_custos_to_psa.py

@@ -5,13 +5,13 @@
 from custos_authnz_token to oidc_user_authnz_tokens.
 """
 
-import jwt
 from datetime import (
     datetime,
     timedelta,
 )
 from typing import Any
 
+import jwt
 import pytest
 from sqlalchemy import (
     Column,
@@ -29,8 +29,8 @@
 
 from galaxy.model.custom_types import MutableJSONType
 from galaxy.model.migrations.data_fixes.custos_to_psa import (
-    CUSTOS_ASSOC_TYPE,
     _extract_iat_from_token,
+    CUSTOS_ASSOC_TYPE,
     migrate_custos_tokens_to_psa,
     remove_migrated_psa_tokens,
     restore_custos_tokens_from_psa,