fix(test): fix OAuth integration tests and auth redirect status code
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
+1
-1
@@ -127,7 +127,7 @@ async def oauth_callback(request: Request):
|
||||
|
||||
# Redirect to original destination or default
|
||||
redirect_url = request.session.pop("redirect_after_login", "/upload")
|
||||
return RedirectResponse(url=redirect_url)
|
||||
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
|
||||
except Exception as e:
|
||||
print(f"OAuth authentication error: {str(e)}")
|
||||
return RedirectResponse(
|
||||
|
||||
+43
-48
@@ -182,57 +182,52 @@ def oauth_enabled_app(oauth_config: Dict[str, str]):
|
||||
Configured test client
|
||||
"""
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
# Save original values
|
||||
original_auth_enabled = os.environ.get("AUTH_ENABLED")
|
||||
original_client_id = os.environ.get("AUTHENTIK_CLIENT_ID")
|
||||
original_client_secret = os.environ.get("AUTHENTIK_CLIENT_SECRET")
|
||||
original_config_url = os.environ.get("AUTHENTIK_CONFIG_URL")
|
||||
|
||||
|
||||
from app.main import app
|
||||
import app.auth as auth_module
|
||||
from app.auth import login, oauth_login, oauth_callback, auth, logout
|
||||
|
||||
# Save original state
|
||||
original_auth_enabled = auth_module.AUTH_ENABLED
|
||||
original_oauth_configured = auth_module.OAUTH_CONFIGURED
|
||||
original_oauth_provider = auth_module.OAUTH_PROVIDER_NAME
|
||||
original_route_count = len(app.router.routes)
|
||||
|
||||
try:
|
||||
# Enable auth and configure OAuth
|
||||
os.environ["AUTH_ENABLED"] = "True"
|
||||
os.environ["AUTHENTIK_CLIENT_ID"] = oauth_config["client_id"]
|
||||
os.environ["AUTHENTIK_CLIENT_SECRET"] = oauth_config["client_secret"]
|
||||
os.environ["AUTHENTIK_CONFIG_URL"] = oauth_config["server_metadata_url"]
|
||||
|
||||
# Need to reload the app module to pick up new config
|
||||
import importlib
|
||||
from app import auth
|
||||
importlib.reload(auth)
|
||||
|
||||
# Enable auth and configure OAuth flags
|
||||
auth_module.AUTH_ENABLED = True
|
||||
auth_module.OAUTH_CONFIGURED = True
|
||||
auth_module.OAUTH_PROVIDER_NAME = oauth_config.get("provider_name", "Test SSO")
|
||||
|
||||
# Register OAuth client
|
||||
auth_module.oauth.register(
|
||||
name="authentik",
|
||||
client_id=oauth_config["client_id"],
|
||||
client_secret=oauth_config["client_secret"],
|
||||
server_metadata_url=oauth_config["server_metadata_url"],
|
||||
client_kwargs={"scope": "openid profile email"},
|
||||
)
|
||||
|
||||
# Add auth routes directly to the app (since include_router was called at startup
|
||||
# with AUTH_ENABLED=False, routes weren't registered)
|
||||
app.add_api_route("/login", login, methods=["GET"])
|
||||
app.add_api_route("/oauth-login", oauth_login, methods=["GET"])
|
||||
app.add_api_route("/oauth-callback", oauth_callback, methods=["GET"], name="oauth_callback")
|
||||
app.add_api_route("/auth", auth, methods=["POST"])
|
||||
app.add_api_route("/logout", logout, methods=["GET"])
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
|
||||
|
||||
# Create test client with base_url to satisfy TrustedHostMiddleware
|
||||
client = TestClient(app, base_url="http://localhost")
|
||||
|
||||
|
||||
yield client
|
||||
|
||||
|
||||
finally:
|
||||
# Restore original values
|
||||
if original_auth_enabled is not None:
|
||||
os.environ["AUTH_ENABLED"] = original_auth_enabled
|
||||
else:
|
||||
os.environ.pop("AUTH_ENABLED", None)
|
||||
|
||||
if original_client_id is not None:
|
||||
os.environ["AUTHENTIK_CLIENT_ID"] = original_client_id
|
||||
else:
|
||||
os.environ.pop("AUTHENTIK_CLIENT_ID", None)
|
||||
|
||||
if original_client_secret is not None:
|
||||
os.environ["AUTHENTIK_CLIENT_SECRET"] = original_client_secret
|
||||
else:
|
||||
os.environ.pop("AUTHENTIK_CLIENT_SECRET", None)
|
||||
|
||||
if original_config_url is not None:
|
||||
os.environ["AUTHENTIK_CONFIG_URL"] = original_config_url
|
||||
else:
|
||||
os.environ.pop("AUTHENTIK_CONFIG_URL", None)
|
||||
|
||||
# Reload auth module to restore original state
|
||||
import importlib
|
||||
from app import auth
|
||||
importlib.reload(auth)
|
||||
# Restore original auth state
|
||||
auth_module.AUTH_ENABLED = original_auth_enabled
|
||||
auth_module.OAUTH_CONFIGURED = original_oauth_configured
|
||||
auth_module.OAUTH_PROVIDER_NAME = original_oauth_provider
|
||||
|
||||
# Remove added routes
|
||||
app.router.routes = app.router.routes[:original_route_count]
|
||||
|
||||
@@ -443,10 +443,12 @@ class TestEmbedMetadataIntoPdf:
|
||||
with patch("app.tasks.embed_metadata_into_pdf.Path") as mock_path_class:
|
||||
# Mock Path for deletion logic
|
||||
mock_original_path = MagicMock()
|
||||
mock_original_path.exists.return_value = True
|
||||
mock_original_path.is_relative_to.return_value = True
|
||||
mock_resolved_path = MagicMock()
|
||||
mock_resolved_path.exists.return_value = True
|
||||
mock_resolved_path.is_relative_to.return_value = True
|
||||
mock_original_path.resolve.return_value = mock_resolved_path
|
||||
mock_workdir_path = MagicMock()
|
||||
mock_path_class.side_effect = [mock_workdir_path, mock_original_path, mock_workdir_path]
|
||||
mock_path_class.side_effect = [mock_workdir_path, mock_original_path]
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
@@ -458,5 +460,5 @@ class TestEmbedMetadataIntoPdf:
|
||||
"/workdir/tmp/test.pdf", "text", {"filename": "test.pdf"}, file_id=333
|
||||
)
|
||||
|
||||
# Verify unlink (delete) was called
|
||||
mock_original_path.unlink.assert_called_once()
|
||||
# Verify unlink (delete) was called on the resolved path
|
||||
mock_resolved_path.unlink.assert_called_once()
|
||||
|
||||
@@ -165,8 +165,8 @@ class TestFinalizeDocumentStorage:
|
||||
# Should still send notification
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
# Should have fallback destination text
|
||||
assert len(notify_args["destinations"]) > 0
|
||||
# No services configured means empty destinations list
|
||||
assert notify_args["destinations"] == []
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
|
||||
@@ -50,7 +50,7 @@ class TestOAuthLoginFlow:
|
||||
try:
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
client = TestClient(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
|
||||
@@ -139,7 +139,7 @@ class TestOAuthCallback:
|
||||
async def test_oauth_callback_rejects_non_admin(
|
||||
self, mock_authorize, oauth_enabled_app: TestClient
|
||||
):
|
||||
"""Test that OAuth callback rejects users without admin group."""
|
||||
"""Test that OAuth callback authenticates non-admin users with is_admin=False."""
|
||||
mock_authorize.return_value = {
|
||||
"access_token": "mock-access-token",
|
||||
"userinfo": {
|
||||
@@ -155,9 +155,8 @@ class TestOAuthCallback:
|
||||
follow_redirects=False,
|
||||
)
|
||||
|
||||
# Should redirect to error page
|
||||
# Non-admin users are still authenticated but with is_admin=False
|
||||
assert response.status_code == 302
|
||||
assert "error" in response.headers.get("location", "").lower()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
|
||||
Reference in New Issue
Block a user