refactor: consolidate linting tools into Ruff
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -9,8 +9,9 @@ These tests use a real OIDC flow with a mock OAuth2 server to test:
|
||||
- Session management
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@@ -25,15 +26,13 @@ class TestOAuthLoginFlow:
|
||||
# Check that OAuth option is shown
|
||||
assert b"oauth" in response.content.lower() or b"sign" in response.content.lower()
|
||||
|
||||
def test_oauth_login_redirects_to_provider(
|
||||
self, oauth_enabled_app: TestClient, oauth_config: dict
|
||||
):
|
||||
def test_oauth_login_redirects_to_provider(self, oauth_enabled_app: TestClient, oauth_config: dict):
|
||||
"""Test that /oauth-login redirects to the OAuth provider."""
|
||||
response = oauth_enabled_app.get("/oauth-login", follow_redirects=False)
|
||||
|
||||
|
||||
# Should redirect to authorization endpoint
|
||||
assert response.status_code == 302
|
||||
|
||||
|
||||
# Redirect location should contain the authorization endpoint
|
||||
location = response.headers.get("location", "")
|
||||
if oauth_config["mode"] == "mock":
|
||||
@@ -44,14 +43,17 @@ class TestOAuthLoginFlow:
|
||||
"""Test that OAuth login fails gracefully when not configured."""
|
||||
# Test with OAuth disabled
|
||||
import os
|
||||
|
||||
original = os.environ.get("AUTH_ENABLED")
|
||||
os.environ["AUTH_ENABLED"] = "False"
|
||||
|
||||
|
||||
try:
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.main import app
|
||||
|
||||
client = TestClient(app, base_url="http://localhost")
|
||||
|
||||
|
||||
response = client.get("/oauth-login", follow_redirects=False)
|
||||
# Should either redirect to error page or show login page
|
||||
assert response.status_code in [302, 404]
|
||||
@@ -77,13 +79,13 @@ class TestOAuthCallback:
|
||||
"expires_in": 3600,
|
||||
"userinfo": test_user_info,
|
||||
}
|
||||
|
||||
|
||||
# Simulate OAuth callback with authorization code
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=test-auth-code&state=test-state",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Should redirect after successful login
|
||||
assert response.status_code == 302
|
||||
|
||||
@@ -97,24 +99,22 @@ class TestOAuthCallback:
|
||||
"access_token": "mock-access-token",
|
||||
"userinfo": test_user_info,
|
||||
}
|
||||
|
||||
|
||||
# First, initiate OAuth flow to set up session
|
||||
oauth_enabled_app.get("/oauth-login", follow_redirects=False)
|
||||
|
||||
|
||||
# Then handle callback
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=test-auth-code",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Should set session cookie
|
||||
assert "set-cookie" in response.headers or response.status_code == 302
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("app.auth.oauth.authentik.authorize_access_token")
|
||||
async def test_oauth_callback_with_admin_user(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient
|
||||
):
|
||||
async def test_oauth_callback_with_admin_user(self, mock_authorize, oauth_enabled_app: TestClient):
|
||||
"""Test OAuth callback with admin user group."""
|
||||
mock_authorize.return_value = {
|
||||
"access_token": "mock-access-token",
|
||||
@@ -125,20 +125,18 @@ class TestOAuthCallback:
|
||||
"groups": ["admin"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=test-auth-code",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Should successfully authenticate
|
||||
assert response.status_code == 302
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("app.auth.oauth.authentik.authorize_access_token")
|
||||
async def test_oauth_callback_rejects_non_admin(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient
|
||||
):
|
||||
async def test_oauth_callback_rejects_non_admin(self, mock_authorize, oauth_enabled_app: TestClient):
|
||||
"""Test that OAuth callback authenticates non-admin users with is_admin=False."""
|
||||
mock_authorize.return_value = {
|
||||
"access_token": "mock-access-token",
|
||||
@@ -149,12 +147,12 @@ class TestOAuthCallback:
|
||||
"groups": ["users"], # No admin group
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=test-auth-code",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Non-admin users are still authenticated but with is_admin=False
|
||||
assert response.status_code == 302
|
||||
|
||||
@@ -174,46 +172,42 @@ class TestOAuthSessionManagement:
|
||||
"access_token": "mock-access-token",
|
||||
"userinfo": test_user_info,
|
||||
}
|
||||
|
||||
|
||||
# Authenticate
|
||||
oauth_enabled_app.get("/oauth-callback?code=test-auth-code")
|
||||
|
||||
|
||||
# Try to access a protected route (e.g., files page)
|
||||
response = oauth_enabled_app.get("/files")
|
||||
|
||||
|
||||
# Should be able to access with valid session
|
||||
# Note: May redirect to login if session not properly set
|
||||
assert response.status_code in [200, 302]
|
||||
|
||||
def test_unauthenticated_user_redirected_to_login(
|
||||
self, oauth_enabled_app: TestClient
|
||||
):
|
||||
def test_unauthenticated_user_redirected_to_login(self, oauth_enabled_app: TestClient):
|
||||
"""Test that unauthenticated users are redirected to login."""
|
||||
# Try to access protected route without authentication
|
||||
response = oauth_enabled_app.get("/files", follow_redirects=False)
|
||||
|
||||
|
||||
# Should redirect to login page
|
||||
if response.status_code == 302:
|
||||
assert "/login" in response.headers.get("location", "")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("app.auth.oauth.authentik.authorize_access_token")
|
||||
async def test_logout_clears_session(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict
|
||||
):
|
||||
async def test_logout_clears_session(self, mock_authorize, oauth_enabled_app: TestClient, test_user_info: dict):
|
||||
"""Test that logout clears user session."""
|
||||
# Mock successful authentication
|
||||
mock_authorize.return_value = {
|
||||
"access_token": "mock-access-token",
|
||||
"userinfo": test_user_info,
|
||||
}
|
||||
|
||||
|
||||
# Authenticate
|
||||
oauth_enabled_app.get("/oauth-callback?code=test-auth-code")
|
||||
|
||||
|
||||
# Logout
|
||||
response = oauth_enabled_app.get("/logout", follow_redirects=False)
|
||||
|
||||
|
||||
# Should redirect after logout
|
||||
assert response.status_code == 302
|
||||
|
||||
@@ -222,29 +216,25 @@ class TestOAuthSessionManagement:
|
||||
class TestOAuthErrorHandling:
|
||||
"""Test error handling in OAuth flows."""
|
||||
|
||||
def test_oauth_callback_without_code_shows_error(
|
||||
self, oauth_enabled_app: TestClient
|
||||
):
|
||||
def test_oauth_callback_without_code_shows_error(self, oauth_enabled_app: TestClient):
|
||||
"""Test OAuth callback without authorization code."""
|
||||
response = oauth_enabled_app.get("/oauth-callback", follow_redirects=False)
|
||||
|
||||
|
||||
# Should handle error gracefully
|
||||
assert response.status_code in [302, 400]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("app.auth.oauth.authentik.authorize_access_token")
|
||||
async def test_oauth_callback_with_invalid_token(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient
|
||||
):
|
||||
async def test_oauth_callback_with_invalid_token(self, mock_authorize, oauth_enabled_app: TestClient):
|
||||
"""Test OAuth callback with invalid token."""
|
||||
# Mock token exchange failure
|
||||
mock_authorize.side_effect = Exception("Invalid authorization code")
|
||||
|
||||
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=invalid-code",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Should redirect to error page
|
||||
assert response.status_code == 302
|
||||
location = response.headers.get("location", "")
|
||||
@@ -252,21 +242,19 @@ class TestOAuthErrorHandling:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch("app.auth.oauth.authentik.authorize_access_token")
|
||||
async def test_oauth_callback_without_userinfo(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient
|
||||
):
|
||||
async def test_oauth_callback_without_userinfo(self, mock_authorize, oauth_enabled_app: TestClient):
|
||||
"""Test OAuth callback when userinfo is missing."""
|
||||
# Mock token without userinfo
|
||||
mock_authorize.return_value = {
|
||||
"access_token": "mock-access-token",
|
||||
"userinfo": None,
|
||||
}
|
||||
|
||||
|
||||
response = oauth_enabled_app.get(
|
||||
"/oauth-callback?code=test-auth-code",
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
|
||||
# Should handle missing userinfo
|
||||
assert response.status_code == 302
|
||||
|
||||
@@ -276,7 +264,7 @@ class TestOAuthErrorHandling:
|
||||
class TestRealOAuthIntegration:
|
||||
"""
|
||||
Integration tests using real OAuth credentials from GitHub Actions secrets.
|
||||
|
||||
|
||||
These tests are skipped unless real OAuth credentials are available.
|
||||
"""
|
||||
|
||||
@@ -284,11 +272,12 @@ class TestRealOAuthIntegration:
|
||||
"""Test that real OAuth .well-known endpoint is accessible."""
|
||||
if not use_real_oauth:
|
||||
pytest.skip("Real OAuth credentials not available")
|
||||
|
||||
|
||||
import requests
|
||||
|
||||
response = requests.get(oauth_config["server_metadata_url"], timeout=10)
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
config = response.json()
|
||||
assert "authorization_endpoint" in config
|
||||
assert "token_endpoint" in config
|
||||
@@ -298,16 +287,17 @@ class TestRealOAuthIntegration:
|
||||
"""Test that real OAuth JWKS endpoint is accessible."""
|
||||
if not use_real_oauth:
|
||||
pytest.skip("Real OAuth credentials not available")
|
||||
|
||||
|
||||
import requests
|
||||
|
||||
# Get well-known config first
|
||||
response = requests.get(oauth_config["server_metadata_url"], timeout=10)
|
||||
config = response.json()
|
||||
|
||||
|
||||
# Test JWKS endpoint
|
||||
jwks_response = requests.get(config["jwks_uri"], timeout=10)
|
||||
assert jwks_response.status_code == 200
|
||||
|
||||
|
||||
jwks = jwks_response.json()
|
||||
assert "keys" in jwks
|
||||
assert len(jwks["keys"]) > 0
|
||||
|
||||
Reference in New Issue
Block a user