Integration tests

Signed-off-by: Saad Zaher <szaher@redhat.com>

Author: szaher

src/openshift_ai_auth/config.py

@@ -8,7 +8,7 @@
 import os
 import warnings
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Optional
 
 from .exceptions import ConfigurationError
 
@@ -60,7 +60,7 @@ class AuthConfig:
     client_id: Optional[str] = None
     client_secret: Optional[str] = None
     openshift_token: Optional[str] = None
-    scopes: List[str] = field(default_factory=lambda: ["openid"])
+    scopes: list[str] = field(default_factory=lambda: ["openid"])
     use_device_flow: bool = False
     use_keyring: bool = False
     oidc_callback_port: int = 8080

src/openshift_ai_auth/strategies/base.py

@@ -7,7 +7,6 @@
 """
 
 from abc import ABC, abstractmethod
-from typing import Optional
 
 from kubernetes.client import ApiClient
 

src/openshift_ai_auth/strategies/incluster.py

@@ -10,7 +10,8 @@
 import os
 from pathlib import Path
 
-from kubernetes import client, config as k8s_config
+from kubernetes import client
+from kubernetes import config as k8s_config
 from kubernetes.client import ApiClient
 from kubernetes.config import ConfigException
 

src/openshift_ai_auth/strategies/kubeconfig.py

@@ -11,7 +11,8 @@
 from pathlib import Path
 from typing import Optional
 
-from kubernetes import client, config as k8s_config
+from kubernetes import client
+from kubernetes import config as k8s_config
 from kubernetes.client import ApiClient
 from kubernetes.config import ConfigException
 
@@ -114,7 +115,7 @@ def authenticate(self) -> ApiClient:
             ) from e
         except Exception as e:
             raise AuthenticationError(
-                f"Unexpected error loading kubeconfig",
+                "Unexpected error loading kubeconfig",
                 f"Error: {type(e).__name__}: {str(e)}"
             ) from e
 

src/openshift_ai_auth/strategies/oidc.py

@@ -12,19 +12,16 @@
 
 import base64
 import hashlib
-import json
 import logging
-import os
 import secrets
 import threading
 import time
 import webbrowser
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Optional, Dict, Any, Tuple
-from urllib.parse import urlencode, urlparse, parse_qs
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlencode, urlparse
 
 import requests
-from kubernetes import client
 from kubernetes.client import ApiClient, Configuration
 
 from ..config import AuthConfig
@@ -76,7 +73,7 @@ def __init__(self, config: AuthConfig) -> None:
             config: AuthConfig instance with OIDC parameters
         """
         super().__init__(config)
-        self._oidc_config: Optional[Dict[str, Any]] = None
+        self._oidc_config: Optional[dict[str, Any]] = None
         self._access_token: Optional[str] = None
         self._refresh_token: Optional[str] = None
         self._token_expiry: Optional[float] = None
@@ -155,7 +152,7 @@ def authenticate(self) -> ApiClient:
         # Create and configure ApiClient
         return self._create_api_client()
 
-    def _discover_oidc_config(self) -> Dict[str, Any]:
+    def _discover_oidc_config(self) -> dict[str, Any]:
         """Discover OIDC configuration from issuer.
 
         Fetches the .well-known/openid-configuration document.
@@ -210,7 +207,7 @@ def _authenticate_device_flow(self) -> None:
         if not device_authorization_endpoint:
             raise AuthenticationError(
                 "Device Code Flow not supported by this OIDC provider",
-                f"The OIDC discovery document does not include 'device_authorization_endpoint'"
+                "The OIDC discovery document does not include 'device_authorization_endpoint'"
             )
 
         # Request device code
@@ -246,10 +243,10 @@ def _authenticate_device_flow(self) -> None:
         interval = device_response.get("interval", 5)
 
         print(f"\n{'='*60}")
-        print(f"OIDC Device Code Authentication")
+        print("OIDC Device Code Authentication")
         print(f"{'='*60}")
         if verification_uri_complete:
-            print(f"\nPlease visit this URL to authenticate:")
+            print("\nPlease visit this URL to authenticate:")
             print(f"\n  {verification_uri_complete}\n")
         else:
             print(f"\nPlease visit this URL: {verification_uri}")
@@ -413,7 +410,7 @@ def log_message(self, format, *args):
         auth_url = f"{authorization_endpoint}?{urlencode(auth_params)}"
 
         # Open browser
-        print(f"\nOpening browser for authentication...")
+        print("\nOpening browser for authentication...")
         print(f"If the browser doesn't open, visit this URL:\n{auth_url}\n")
 
         webbrowser.open(auth_url)

src/openshift_ai_auth/strategies/openshift.py

@@ -14,26 +14,22 @@
 
 import base64
 import hashlib
-import json
 import logging
 import os
 import secrets
 import threading
-import time
 import webbrowser
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Optional, Dict, Any, Tuple
-from urllib.parse import urlencode, urlparse, parse_qs
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlencode, urlparse
 
 import requests
-from kubernetes import client
 from kubernetes.client import ApiClient, Configuration
 
 from ..config import AuthConfig
 from ..exceptions import (
     AuthenticationError,
     ConfigurationError,
-    OpenShiftOAuthError,
     StrategyNotAvailableError,
 )
 from .base import AuthStrategy
@@ -81,7 +77,7 @@ def __init__(self, config: AuthConfig) -> None:
             config: AuthConfig instance with OpenShift parameters
         """
         super().__init__(config)
-        self._oauth_metadata: Optional[Dict[str, Any]] = None
+        self._oauth_metadata: Optional[dict[str, Any]] = None
         self._access_token: Optional[str] = None
 
     def is_available(self) -> bool:
@@ -164,7 +160,7 @@ def authenticate(self) -> ApiClient:
         # Create and configure ApiClient
         return self._create_api_client()
 
-    def _discover_oauth_metadata(self) -> Dict[str, Any]:
+    def _discover_oauth_metadata(self) -> dict[str, Any]:
         """Discover OpenShift OAuth server metadata.
 
         Fetches the /.well-known/oauth-authorization-server document.
@@ -297,7 +293,7 @@ def log_message(self, format, *args):
         auth_url = f"{authorization_endpoint}?{urlencode(auth_params)}"
 
         # Open browser
-        print(f"\nOpening browser for OpenShift authentication...")
+        print("\nOpening browser for OpenShift authentication...")
         print(f"If the browser doesn't open, visit this URL:\n{auth_url}\n")
 
         webbrowser.open(auth_url)

tests/conftest.py

@@ -5,9 +5,8 @@
 environments, kubeconfig files, and service account tokens.
 """
 
-import os
+from collections.abc import Generator
 from pathlib import Path
-from typing import Generator
 
 import pytest
 

tests/integration/test_oidc_integration.py

@@ -10,14 +10,15 @@
     pytest tests/integration/ -v  # Run all integration tests
 """
 
-import pytest
 import warnings
-from unittest.mock import patch, MagicMock
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from openshift_ai_auth import AuthConfig
 from openshift_ai_auth.config import SecurityWarning
-from openshift_ai_auth.strategies.oidc import OIDCStrategy
 from openshift_ai_auth.exceptions import AuthenticationError
+from openshift_ai_auth.strategies.oidc import OIDCStrategy
 
 
 @pytest.mark.integration
@@ -45,11 +46,11 @@ def test_full_auth_code_flow_with_mock_server(self, mock_oauth_server, mock_env_
         assert strategy.is_available()
 
         # Mock the browser opening and callback handling
-        with patch('webbrowser.open') as mock_browser:
+        with patch('webbrowser.open'):
             # Mock the callback server to simulate successful auth
             with patch('openshift_ai_auth.strategies.oidc.HTTPServer') as mock_server:
                 # Simulate auth code callback
-                mock_handler_instance = MagicMock()
+                MagicMock()
                 mock_server_instance = MagicMock()
                 mock_server.return_value = mock_server_instance
 
@@ -59,17 +60,7 @@ def test_full_auth_code_flow_with_mock_server(self, mock_oauth_server, mock_env_
                 def simulate_callback(*args, **kwargs):
                     """Simulate receiving auth code."""
                     # Request auth code from mock server
-                    import requests
-                    from urllib.parse import urlencode
-
-                    params = {
-                        "client_id": mock_oauth_server.client_id,
-                        "response_type": "code",
-                        "redirect_uri": "http://localhost:8080/callback",
-                        "code_challenge": "test_challenge",
-                        "code_challenge_method": "S256",
-                        "state": "test_state",
-                    }
+
 
                     # The mock server will auto-approve and return a code
                     # We simulate this by directly calling token endpoint
@@ -79,7 +70,6 @@ def simulate_callback(*args, **kwargs):
 
                 # For this test, we'll patch the entire auth flow
                 # to use our mock server's tokens
-                original_auth = strategy._authenticate_auth_code_flow
 
                 def mock_auth():
                     """Mock authentication that uses real mock server."""
@@ -241,11 +231,12 @@ class TestOIDCIntegrationPKCE:
 
     def test_pkce_code_challenge_generation(self, mock_oauth_server, mock_env_vars):
         """Test that PKCE code challenge is properly generated and verified."""
-        import requests
         import base64
         import hashlib
         import secrets
 
+        import requests
+
         # Generate PKCE parameters
         code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
         code_challenge = base64.urlsafe_b64encode(
@@ -272,7 +263,7 @@ def test_pkce_code_challenge_generation(self, mock_oauth_server, mock_env_vars):
         assert "code=" in location
 
         # Extract auth code
-        from urllib.parse import urlparse, parse_qs
+        from urllib.parse import parse_qs, urlparse
         parsed = urlparse(location)
         params = parse_qs(parsed.query)
         auth_code = params["code"][0]
@@ -297,11 +288,12 @@ def test_pkce_code_challenge_generation(self, mock_oauth_server, mock_env_vars):
 
     def test_pkce_verification_failure(self, mock_oauth_server, mock_env_vars):
         """Test that PKCE verification fails with wrong verifier."""
-        import requests
         import base64
         import hashlib
         import secrets
 
+        import requests
+
         # Generate PKCE parameters
         code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
         code_challenge = base64.urlsafe_b64encode(
@@ -322,7 +314,7 @@ def test_pkce_verification_failure(self, mock_oauth_server, mock_env_vars):
         )
 
         # Extract auth code
-        from urllib.parse import urlparse, parse_qs
+        from urllib.parse import parse_qs, urlparse
         location = auth_response.headers.get("Location", "")
         parsed = urlparse(location)
         params = parse_qs(parsed.query)

tests/mock_oauth_server.py

@@ -17,8 +17,8 @@
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Dict, Any, Optional
-from urllib.parse import parse_qs, urlparse, urlencode
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlencode, urlparse
 
 
 class MockOAuthServer:
@@ -38,9 +38,9 @@ def __init__(self, host: str = "localhost", port: int = 9999):
         self.base_url = f"http://{host}:{port}"
 
         # Storage for active flows
-        self.auth_codes: Dict[str, Dict[str, Any]] = {}
-        self.device_codes: Dict[str, Dict[str, Any]] = {}
-        self.tokens: Dict[str, Dict[str, Any]] = {}
+        self.auth_codes: dict[str, dict[str, Any]] = {}
+        self.device_codes: dict[str, dict[str, Any]] = {}
+        self.tokens: dict[str, dict[str, Any]] = {}
 
         # Configuration
         self.issuer = self.base_url
@@ -135,7 +135,7 @@ def _handle_openshift_discovery(self):
                 }
                 self._send_json(200, discovery)
 
-            def _handle_authorize(self, params: Dict[str, list]):
+            def _handle_authorize(self, params: dict[str, list]):
                 """Handle authorization request."""
                 client_id = params.get("client_id", [""])[0]
                 redirect_uri = params.get("redirect_uri", [""])[0]
@@ -171,7 +171,7 @@ def _handle_authorize(self, params: Dict[str, list]):
                     # Return approval page (for manual testing)
                     self._send_html(200, "<h1>Approval Page</h1><p>Auto-approve is disabled</p>")
 
-            def _handle_token(self, params: Dict[str, list]):
+            def _handle_token(self, params: dict[str, list]):
                 """Handle token endpoint requests."""
                 grant_type = params.get("grant_type", [""])[0]
                 client_id = params.get("client_id", [""])[0]
@@ -188,10 +188,10 @@ def _handle_token(self, params: Dict[str, list]):
                         "error_description": f"Grant type '{grant_type}' is not supported"
                     })
 
-            def _handle_token_auth_code(self, params: Dict[str, list], client_id: str):
+            def _handle_token_auth_code(self, params: dict[str, list], client_id: str):
                 """Handle authorization code grant."""
                 code = params.get("code", [""])[0]
-                redirect_uri = params.get("redirect_uri", [""])[0]
+                params.get("redirect_uri", [""])[0]
                 code_verifier = params.get("code_verifier", [""])[0]
 
                 # Validate auth code
@@ -259,7 +259,7 @@ def _handle_token_auth_code(self, params: Dict[str, list], client_id: str):
                     "scope": "openid profile email"
                 })
 
-            def _handle_token_refresh(self, params: Dict[str, list], client_id: str):
+            def _handle_token_refresh(self, params: dict[str, list], client_id: str):
                 """Handle refresh token grant."""
                 refresh_token = params.get("refresh_token", [""])[0]
 
@@ -285,7 +285,7 @@ def _handle_token_refresh(self, params: Dict[str, list], client_id: str):
                     "scope": "openid profile email"
                 })
 
-            def _handle_device_code(self, params: Dict[str, list]):
+            def _handle_device_code(self, params: dict[str, list]):
                 """Handle device code request."""
                 client_id = params.get("client_id", [""])[0]
 
@@ -308,7 +308,7 @@ def _handle_device_code(self, params: Dict[str, list]):
                     "interval": 1  # Fast polling for tests
                 })
 
-            def _handle_token_device_code(self, params: Dict[str, list], client_id: str):
+            def _handle_token_device_code(self, params: dict[str, list], client_id: str):
                 """Handle device code token request (polling)."""
                 device_code = params.get("device_code", [""])[0]