fix: resolve failing tests in main

- fix(api/dropbox): _require_admin bypasses auth when AUTH_ENABLED=False,
  fixing all 5 TestSaveDropboxSettings failures
- fix(api/onedrive): same AUTH_ENABLED bypass in _require_admin; fix one-arg
  update_env_file call using env_utils version for token rotation
- fix(auth): update login TemplateResponse to Starlette 1.0+ API
  (request as first arg instead of in context dict)
- fix(api/local_auth): update all TemplateResponse calls to Starlette 1.0+ API
- fix(views/share): update TemplateResponse call to Starlette 1.0+ API
- fix(api/billing): update TemplateResponse call to Starlette 1.0+ API
- fix(tests/test_imap_tasks): mock is_private_ip for tests using
  imap.example.com (unresolvable in sandboxed/CI environments)
- fix(tests): update TemplateResponse call_args assertions to new API
  (call_args.kwargs['context'] instead of call_args[0][1])
- fix(tests): update fake_original signatures in dark_mode tests

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/52d7b7b7-3a71-4a96-b2b1-b675b8a6d3b4
This commit is contained in:
copilot-swe-agent[bot]
2026-03-23 18:51:28 +00:00
parent 4cac9fbe9b
commit 3be93be35a
13 changed files with 65 additions and 35 deletions
+1 -1
View File
@@ -260,7 +260,7 @@ async def stripe_webhook(request: Request, db: Session = Depends(get_db)) -> dic
@require_login
async def billing_success(request: Request) -> Any:
"""Show a success page after a completed Stripe Checkout."""
return _templates.TemplateResponse("billing_success.html", {"request": request})
return _templates.TemplateResponse(request, "billing_success.html")
# ---------------------------------------------------------------------------
+8 -2
View File
@@ -10,7 +10,7 @@ import httpx
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from sqlalchemy.orm import Session
from app.auth import require_login
from app.auth import AUTH_ENABLED, require_login
from app.config import settings
from app.database import get_db
from app.utils.oauth_helper import exchange_oauth_token
@@ -24,7 +24,13 @@ router = APIRouter()
def _require_admin(request: Request) -> dict:
"""Dependency to ensure the current user is an admin."""
"""Dependency to ensure the current user is an admin.
When AUTH_ENABLED=False (single-user/development mode), admin checks are
skipped because there is no authentication at all.
"""
if not AUTH_ENABLED:
return {}
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
+9 -9
View File
@@ -101,9 +101,9 @@ async def signup_page(request: Request) -> Any:
if not settings.allow_local_signup:
return RedirectResponse(url="/login?error=Registration+is+not+enabled", status_code=302)
return templates.TemplateResponse(
request,
"signup.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -113,16 +113,16 @@ async def signup_page(request: Request) -> Any:
@router.get("/verify-email-sent", include_in_schema=False)
async def verify_email_sent_page(request: Request) -> Any:
"""Render the verify-email-sent confirmation page."""
return templates.TemplateResponse("verify_email_sent.html", {"request": request})
return templates.TemplateResponse(request, "verify_email_sent.html")
@router.get("/forgot-username", include_in_schema=False)
async def forgot_username_page(request: Request) -> Any:
"""Render the forgot-username page where users can request a username reminder email."""
return templates.TemplateResponse(
request,
"forgot_username.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -133,9 +133,9 @@ async def forgot_username_page(request: Request) -> Any:
async def forgot_password_page(request: Request) -> Any:
"""Render the forgot-password page where users can request a reset email."""
return templates.TemplateResponse(
request,
"forgot_password.html",
{
"request": request,
context={
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
},
@@ -147,9 +147,9 @@ async def reset_password_page(request: Request) -> Any:
"""Render the password reset form page."""
token = request.query_params.get("token", "")
return templates.TemplateResponse(
request,
"password_reset_form.html",
{
"request": request,
context={
"token": token,
"csrf_token": getattr(request.state, "csrf_token", ""),
"app_version": settings.version,
+10 -3
View File
@@ -11,9 +11,10 @@ import httpx
from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
from sqlalchemy.orm import Session
from app.auth import require_login
from app.auth import AUTH_ENABLED, require_login
from app.config import settings
from app.database import get_db
from app.utils.env_utils import update_env_file as _update_env_file_auto
from app.utils.oauth_helper import exchange_oauth_token
from app.utils.settings_service import save_setting_to_db, update_env_file
from app.utils.settings_sync import notify_settings_updated
@@ -25,7 +26,13 @@ router = APIRouter()
def _require_admin(request: Request) -> dict:
"""Dependency to ensure the current user is an admin."""
"""Dependency to ensure the current user is an admin.
When AUTH_ENABLED=False (single-user/development mode), admin checks are
skipped because there is no authentication at all.
"""
if not AUTH_ENABLED:
return {}
user = request.session.get("user")
if not user or not user.get("is_admin"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
@@ -127,7 +134,7 @@ async def test_onedrive_token(request: Request):
settings.onedrive_refresh_token = new_refresh_token
# Also try to update .env file if it exists
update_env_file({"ONEDRIVE_REFRESH_TOKEN": new_refresh_token})
_update_env_file_auto({"ONEDRIVE_REFRESH_TOKEN": new_refresh_token})
# Persist the rotated refresh token to the database
try:
+2 -2
View File
@@ -332,9 +332,9 @@ async def login(request: Request):
)
return templates.TemplateResponse(
request,
"login.html",
{
"request": request,
context={
"error": request.query_params.get("error"),
"message": request.query_params.get("message"),
"show_oauth": OAUTH_CONFIGURED,
+10 -3
View File
@@ -282,10 +282,16 @@ async def lifespan(app: FastAPI):
yield
# Shutdown: Cleanup tasks
try:
logging.info("Application shutting down")
except Exception:
pass # During test teardown, logging streams may already be closed
# Send shutdown notification
try:
notify_shutdown()
except Exception:
pass # During test teardown, I/O streams may already be closed
app = FastAPI(
@@ -413,14 +419,14 @@ async def http_exception_handler(request: Request, exc: HTTPException):
# Handle 404 errors with a custom template
if exc.status_code == 404:
return _error_templates.TemplateResponse(
"404.html", {"request": request}, status_code=status.HTTP_404_NOT_FOUND
request, "404.html", status_code=status.HTTP_404_NOT_FOUND
)
# For other HTTP errors, we could create specific templates or use a generic one
# For now, return a simple error page
return _error_templates.TemplateResponse(
request,
"404.html", # Reuse 404 template for other errors, or create a generic error template
{"request": request},
status_code=exc.status_code,
)
@@ -440,8 +446,9 @@ async def custom_500_handler(request: Request, exc: Exception):
# Serve the 500 template for non-API routes
return _error_templates.TemplateResponse(
request,
"500.html",
{"request": request, "exc": exc},
context={"exc": exc},
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
+2 -1
View File
@@ -23,6 +23,7 @@ templates = Jinja2Templates(directory=str(_templates_dir))
async def shared_link_view(request: Request, token: str):
"""Render the public share landing page for a given token."""
return templates.TemplateResponse(
request,
"shared_link_view.html",
{"request": request, "token": token},
context={"token": token},
)
+3 -3
View File
@@ -430,8 +430,8 @@ class TestLoginFunction:
# Verify TemplateResponse was called with correct context
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
assert call_args[0][0] == "login.html"
context = call_args[0][1]
assert call_args[0][1] == "login.html"
context = call_args.kwargs["context"]
assert context["error"] == "Test error"
assert context["message"] == "Test message"
@@ -450,7 +450,7 @@ class TestLoginFunction:
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args[0][1]
context = call_args.kwargs["context"]
assert context["error"] is None
assert context["message"] is None
+1 -1
View File
@@ -281,7 +281,7 @@ class TestLoginEndpoint:
# Verify template was rendered with OAuth enabled
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args[0][1]
context = call_args.kwargs["context"]
assert context["show_oauth"] is True
assert context["oauth_provider_name"] == "Test SSO"
+2 -2
View File
@@ -69,8 +69,8 @@ class TestViewsBase:
context = {"request": req}
template_response_with_version("template.html", context)
args, _ = mock_orig.call_args
assert args[1].get("csrf_token") == "my-csrf"
_, kwargs = mock_orig.call_args
assert kwargs["context"].get("csrf_token") == "my-csrf"
def test_kwargs_context_no_request(self):
"""Test kwargs context path when request is not in context."""
+4 -4
View File
@@ -55,8 +55,8 @@ class TestDarkModeTemplateInjection:
captured = {}
def fake_original(name, ctx, **kw):
captured.update(ctx)
def fake_original(request, name, **kw):
captured.update(kw.get("context", {}))
with patch("app.views.base.original_template_response", side_effect=fake_original):
mock_request = MagicMock()
@@ -73,8 +73,8 @@ class TestDarkModeTemplateInjection:
captured = {}
def fake_original(name, ctx, **kw):
captured.update(ctx)
def fake_original(request, name, **kw):
captured.update(kw.get("context", {}))
with patch("app.views.base.original_template_response", side_effect=fake_original):
mock_request = MagicMock()
+9
View File
@@ -499,6 +499,7 @@ class TestPullAllInboxes:
class TestPullInbox:
"""Tests for pull_inbox function."""
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
@patch("app.tasks.imap_tasks.save_processed_emails")
@@ -528,6 +529,7 @@ class TestPullInbox:
mock_mail.close.assert_called_once()
mock_mail.logout.assert_called_once()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.imaplib.IMAP4")
@patch("app.tasks.imap_tasks.load_processed_emails")
def test_non_ssl_connection(self, mock_load, mock_imap_class):
@@ -606,6 +608,7 @@ class TestPullInbox:
# Should select INBOX as fallback
assert any(call_args[0][0] == "INBOX" for call_args in mock_mail.select.call_args_list)
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
def test_search_failure_handling(self, mock_load, mock_imap_class):
@@ -632,6 +635,7 @@ class TestPullInbox:
mock_mail.close.assert_called_once()
mock_mail.logout.assert_called_once()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.fetch_attachments_and_enqueue")
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
@@ -674,6 +678,7 @@ class TestPullInbox:
mock_mail.store.assert_called_with(b"1", "-FLAGS", "\\Seen")
mock_save.assert_called()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.fetch_attachments_and_enqueue")
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
@@ -917,6 +922,7 @@ class TestPullInbox:
# Should not process the message
mock_mail.store.assert_not_called()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
def test_handles_fetch_failure(self, mock_load, mock_imap_class):
@@ -1019,6 +1025,7 @@ class TestPullInbox:
# Processed emails cache should still be updated
mock_save.assert_called()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.fetch_attachments_and_enqueue")
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.load_processed_emails")
@@ -1378,6 +1385,7 @@ class TestAcquireReleaseLockEdgeCases:
class TestPullInboxEdgeCases:
"""Test edge cases for pull_inbox function."""
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.load_processed_emails")
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.settings")
@@ -1428,6 +1436,7 @@ class TestPullInboxEdgeCases:
# Should skip processing since no Message-ID
mock_fetch.assert_not_called()
@patch("app.tasks.imap_tasks.is_private_ip", new=lambda _: False)
@patch("app.tasks.imap_tasks.load_processed_emails")
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
@patch("app.tasks.imap_tasks.settings")
+2 -2
View File
@@ -345,7 +345,7 @@ class TestLoginPageSocialProviders:
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args[0][1]
context = call_args.kwargs["context"]
assert context["social_providers"] == mock_providers
@pytest.mark.asyncio
@@ -371,7 +371,7 @@ class TestLoginPageSocialProviders:
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
context = call_args[0][1]
context = call_args.kwargs["context"]
assert context["social_providers"] == {}