feat(config): add LOG_LEVEL setting and configure root logging at startup

- Add `log_level` setting to config.py (default: INFO, env: LOG_LEVEL)
- Configure Python root logger in main.py with standard precedence:
  LOG_LEVEL explicit > DEBUG=true implies DEBUG > default INFO
- Add timestamp to log format for production readability
- Suppress noisy third-party loggers at DEBUG level
- Add comprehensive debug logging to all auth functions
- Add LOG_LEVEL/DEBUG to .env.demo and ConfigurationGuide.md
- Add tests for logging config and auth debug output

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-03-16 09:33:50 +00:00
parent 6a044753eb
commit 18c49c6b2d
7 changed files with 382 additions and 2 deletions
+7
View File
@@ -7,6 +7,13 @@ GOTENBERG_URL=http://gotenberg:3000
ALLOW_FILE_DELETE=true # Allow deletion of file records ALLOW_FILE_DELETE=true # Allow deletion of file records
COMPLIANCE_ENABLED=true # Enable compliance templates dashboard (GDPR, HIPAA, SOC 2) COMPLIANCE_ENABLED=true # Enable compliance templates dashboard (GDPR, HIPAA, SOC 2)
# **Logging**
# LOG_LEVEL controls the Python root-logger level.
# Accepted values: DEBUG, INFO, WARNING, ERROR, CRITICAL (default: INFO).
# When DEBUG=true and LOG_LEVEL is not set, the level is automatically lowered to DEBUG.
# LOG_LEVEL=INFO
# DEBUG=false
# **UI / Appearance** # **UI / Appearance**
# Default colour scheme: system (follow OS), light, or dark # Default colour scheme: system (follow OS), light, or dark
# Individual users can always override with the navbar dark-mode toggle. # Individual users can always override with the navbar dark-mode toggle.
+81 -2
View File
@@ -125,8 +125,17 @@ def get_current_user(request: Request):
# Check for Bearer token auth first (API tokens) # Check for Bearer token auth first (API tokens)
api_user = getattr(request.state, "api_token_user", None) api_user = getattr(request.state, "api_token_user", None)
if isinstance(api_user, dict): if isinstance(api_user, dict):
logger.debug("[AUTH] get_current_user: resolved from API token (user_id=%s)", api_user.get("id"))
return api_user return api_user
return request.session.get("user") session_user = request.session.get("user")
if session_user:
logger.debug(
"[AUTH] get_current_user: resolved from session (user=%s)",
session_user.get("preferred_username") or session_user.get("email") or session_user.get("id"),
)
else:
logger.debug("[AUTH] get_current_user: no user in session or API token")
return session_user
def _resolve_bearer_user(request: Request, db: Session) -> dict | None: def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
@@ -141,10 +150,12 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
""" """
auth_header = request.headers.get("authorization", "") auth_header = request.headers.get("authorization", "")
if not isinstance(auth_header, str) or not auth_header.startswith("Bearer "): if not isinstance(auth_header, str) or not auth_header.startswith("Bearer "):
logger.debug("[AUTH] _resolve_bearer_user: no Bearer token in Authorization header")
return None return None
raw_token = auth_header[7:] raw_token = auth_header[7:]
if not raw_token or not isinstance(raw_token, str): if not raw_token or not isinstance(raw_token, str):
logger.debug("[AUTH] _resolve_bearer_user: empty or invalid token after 'Bearer ' prefix")
return None return None
from app.api.api_tokens import hash_token from app.api.api_tokens import hash_token
@@ -153,8 +164,15 @@ def _resolve_bearer_user(request: Request, db: Session) -> dict | None:
token_hash = hash_token(raw_token) token_hash = hash_token(raw_token)
db_token = db.query(ApiToken).filter(ApiToken.token_hash == token_hash, ApiToken.is_active.is_(True)).first() db_token = db.query(ApiToken).filter(ApiToken.token_hash == token_hash, ApiToken.is_active.is_(True)).first()
if db_token is None: if db_token is None:
logger.debug("[AUTH] _resolve_bearer_user: no active API token matched the provided hash")
return None return None
logger.debug(
"[AUTH] _resolve_bearer_user: matched API token id=%s owner=%s",
db_token.id,
db_token.owner_id,
)
# Update usage tracking # Update usage tracking
try: try:
db_token.last_used_at = datetime.now(timezone.utc) db_token.last_used_at = datetime.now(timezone.utc)
@@ -205,16 +223,18 @@ def require_login(func):
@wraps(func) @wraps(func)
async def wrapper(request: Request, *args, **kwargs): async def wrapper(request: Request, *args, **kwargs):
url_path = urlparse(str(request.url)).path
# Check session auth first # Check session auth first
if request.session.get("user"): if request.session.get("user"):
logger.debug("[AUTH] require_login: session auth OK for %s", url_path)
if inspect.iscoroutinefunction(func): if inspect.iscoroutinefunction(func):
return await func(*args, request=request, **kwargs) return await func(*args, request=request, **kwargs)
else: else:
return func(*args, request=request, **kwargs) return func(*args, request=request, **kwargs)
# Fall back to Bearer token auth for API endpoints # Fall back to Bearer token auth for API endpoints
url_path = urlparse(str(request.url)).path
if url_path.startswith("/api/"): if url_path.startswith("/api/"):
logger.debug("[AUTH] require_login: no session, trying Bearer token for %s", url_path)
try: try:
from app.database import SessionLocal from app.database import SessionLocal
@@ -228,17 +248,22 @@ def require_login(func):
if api_user: if api_user:
request.state.api_token_user = api_user request.state.api_token_user = api_user
logger.debug(
"[AUTH] require_login: Bearer token auth OK for %s (user=%s)", url_path, api_user.get("id")
)
if inspect.iscoroutinefunction(func): if inspect.iscoroutinefunction(func):
return await func(*args, request=request, **kwargs) return await func(*args, request=request, **kwargs)
else: else:
return func(*args, request=request, **kwargs) return func(*args, request=request, **kwargs)
logger.debug("[AUTH] require_login: no valid auth for API endpoint %s — returning 401", url_path)
return JSONResponse( return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Not authenticated"}, content={"error": "Not authenticated"},
) )
# Non-API endpoint with no session — redirect to login # Non-API endpoint with no session — redirect to login
logger.debug("[AUTH] require_login: no session for %s — redirecting to /login", url_path)
request.session["redirect_after_login"] = str(request.url) request.session["redirect_after_login"] = str(request.url)
return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login", status_code=status.HTTP_302_FOUND)
@@ -307,9 +332,15 @@ async def login(request: Request):
async def oauth_login(request: Request): async def oauth_login(request: Request):
"""Handle OAuth login flow""" """Handle OAuth login flow"""
if not OAUTH_CONFIGURED: if not OAUTH_CONFIGURED:
logger.debug("[AUTH] oauth_login: OAuth not configured — redirecting to /login")
return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login?error=OAuth+not+configured", status_code=status.HTTP_302_FOUND)
redirect_uri = request.url_for("oauth_callback") redirect_uri = request.url_for("oauth_callback")
logger.debug(
"[AUTH] oauth_login: initiating Authentik OAuth redirect_uri=%s session_keys=%s",
redirect_uri,
list(request.session.keys()),
)
return await oauth.authentik.authorize_redirect(request, redirect_uri) return await oauth.authentik.authorize_redirect(request, redirect_uri)
@@ -324,13 +355,23 @@ async def social_login(request: Request, provider: str):
A redirect to the provider's authorization page, or back to /login on error. A redirect to the provider's authorization page, or back to /login on error.
""" """
if provider not in SOCIAL_PROVIDERS: if provider not in SOCIAL_PROVIDERS:
logger.debug(
"[AUTH] social_login: unknown provider=%r (registered=%s)", provider, list(SOCIAL_PROVIDERS.keys())
)
return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
redirect_uri = request.url_for("social_callback", provider=provider) redirect_uri = request.url_for("social_callback", provider=provider)
oauth_client = getattr(oauth, provider, None) oauth_client = getattr(oauth, provider, None)
if oauth_client is None: if oauth_client is None:
logger.debug("[AUTH] social_login: provider=%r registered but OAuth client not configured", provider)
return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
logger.debug(
"[AUTH] social_login: initiating %s OAuth, redirect_uri=%s session_keys=%s",
provider,
redirect_uri,
list(request.session.keys()),
)
return await oauth_client.authorize_redirect(request, redirect_uri) return await oauth_client.authorize_redirect(request, redirect_uri)
@@ -389,27 +430,39 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
A redirect to the user's original destination or the upload page. A redirect to the user's original destination or the upload page.
""" """
if provider not in SOCIAL_PROVIDERS: if provider not in SOCIAL_PROVIDERS:
logger.debug("[AUTH] social_callback: unknown provider=%r", provider)
return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login?error=Unknown+social+provider", status_code=status.HTTP_302_FOUND)
oauth_client = getattr(oauth, provider, None) oauth_client = getattr(oauth, provider, None)
if oauth_client is None: if oauth_client is None:
logger.debug("[AUTH] social_callback: provider=%r not configured", provider)
return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/login?error=Provider+not+configured", status_code=status.HTTP_302_FOUND)
try: try:
logger.debug("[AUTH] social_callback: exchanging auth code for provider=%s", provider)
token = await oauth_client.authorize_access_token(request) token = await oauth_client.authorize_access_token(request)
# Try standard OIDC userinfo first, fall back to token-embedded userinfo # Try standard OIDC userinfo first, fall back to token-embedded userinfo
raw_userinfo = token.get("userinfo") raw_userinfo = token.get("userinfo")
if not raw_userinfo: if not raw_userinfo:
logger.debug("[AUTH] social_callback: no userinfo in token, fetching from userinfo endpoint")
try: try:
resp = await oauth_client.userinfo(token=token) resp = await oauth_client.userinfo(token=token)
raw_userinfo = resp if isinstance(resp, dict) else resp.json() if hasattr(resp, "json") else {} raw_userinfo = resp if isinstance(resp, dict) else resp.json() if hasattr(resp, "json") else {}
except Exception: except Exception:
logger.debug("[AUTH] social_callback: userinfo endpoint failed, using empty dict", exc_info=True)
raw_userinfo = {} raw_userinfo = {}
user_data = _normalize_social_userinfo(provider, token, raw_userinfo) user_data = _normalize_social_userinfo(provider, token, raw_userinfo)
logger.debug(
"[AUTH] social_callback: normalized user_data email=%s sub=%s provider=%s",
user_data.get("email"),
user_data.get("sub"),
provider,
)
if not user_data.get("email"): if not user_data.get("email"):
logger.debug("[AUTH] social_callback: no email in user_data — aborting")
return RedirectResponse( return RedirectResponse(
url="/login?error=Could+not+retrieve+email+from+provider", url="/login?error=Could+not+retrieve+email+from+provider",
status_code=status.HTTP_302_FOUND, status_code=status.HTTP_302_FOUND,
@@ -442,21 +495,29 @@ async def social_callback(request: Request, provider: str, db: Session = Depends
) )
# Mobile app flow: issue an inline API token and redirect back to the app. # Mobile app flow: issue an inline API token and redirect back to the app.
logger.debug(
"[MOBILE] social_callback: checking for mobile redirect (session has mobile_redirect_uri=%s)",
"mobile_redirect_uri" in request.session,
)
mobile_resp = _create_mobile_redirect(request, db) mobile_resp = _create_mobile_redirect(request, db)
if mobile_resp: if mobile_resp:
logger.info("[MOBILE] social_callback: returning mobile redirect response for provider=%s", provider)
return mobile_resp return mobile_resp
if user_id: if user_id:
profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first() profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
if profile and not profile.onboarding_completed: if profile and not profile.onboarding_completed:
logger.debug("[AUTH] social_callback: user=%s needs onboarding, redirecting", user_id)
post_onboarding = request.session.pop("redirect_after_login", "/upload") post_onboarding = request.session.pop("redirect_after_login", "/upload")
request.session["post_onboarding_redirect"] = post_onboarding request.session["post_onboarding_redirect"] = post_onboarding
return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
redirect_url = request.session.pop("redirect_after_login", "/upload") redirect_url = request.session.pop("redirect_after_login", "/upload")
logger.debug("[AUTH] social_callback: login complete, redirecting to %s", redirect_url)
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND) return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
except Exception as e: except Exception as e:
logger.warning("[SECURITY] SOCIAL_LOGIN_FAILURE provider=%s error=%s", provider, type(e).__name__) logger.warning("[SECURITY] SOCIAL_LOGIN_FAILURE provider=%s error=%s", provider, type(e).__name__)
logger.debug("[AUTH] social_callback: full exception for provider=%s", provider, exc_info=True)
return RedirectResponse( return RedirectResponse(
url="/login?error=Social+login+failed.+Please+try+again.", status_code=status.HTTP_302_FOUND url="/login?error=Social+login+failed.+Please+try+again.", status_code=status.HTTP_302_FOUND
) )
@@ -561,15 +622,23 @@ def _ensure_user_profile(db: Session, user_data: dict, is_admin: bool = False) -
async def oauth_callback(request: Request, db: Session = Depends(get_db)): async def oauth_callback(request: Request, db: Session = Depends(get_db)):
"""Handle OAuth callback from provider""" """Handle OAuth callback from provider"""
try: try:
logger.debug("[AUTH] oauth_callback: exchanging authorization code for token")
token = await oauth.authentik.authorize_access_token(request) token = await oauth.authentik.authorize_access_token(request)
userinfo = token.get("userinfo") userinfo = token.get("userinfo")
if not userinfo: if not userinfo:
logger.debug("[AUTH] oauth_callback: no userinfo in token response — aborting")
return RedirectResponse( return RedirectResponse(
url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND url="/login?error=Failed+to+retrieve+user+information", status_code=status.HTTP_302_FOUND
) )
# Store user info in session # Store user info in session
user_data = dict(userinfo) user_data = dict(userinfo)
logger.debug(
"[AUTH] oauth_callback: received userinfo email=%s sub=%s groups=%s",
user_data.get("email"),
user_data.get("sub"),
user_data.get("groups", []),
)
# Add Gravatar picture if no picture is provided # Add Gravatar picture if no picture is provided
if not user_data.get("picture") and user_data.get("email"): if not user_data.get("picture") and user_data.get("email"):
@@ -584,6 +653,12 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
groups = user_data.get("groups", []) groups = user_data.get("groups", [])
admin_group = (settings.admin_group_name or "admin").strip().lower() admin_group = (settings.admin_group_name or "admin").strip().lower()
is_admin = admin_group in [group.lower() for group in groups] is_admin = admin_group in [group.lower() for group in groups]
logger.debug(
"[AUTH] oauth_callback: admin group check — looking for %r in %s → is_admin=%s",
admin_group,
[g.lower() for g in groups],
is_admin,
)
# Set is_admin flag (defaults to False for OAuth users unless they're in admin group) # Set is_admin flag (defaults to False for OAuth users unless they're in admin group)
user_data["is_admin"] = is_admin user_data["is_admin"] = is_admin
@@ -623,15 +698,18 @@ async def oauth_callback(request: Request, db: Session = Depends(get_db)):
if user_id: if user_id:
profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first() profile = db.query(_UserProfile).filter(_UserProfile.user_id == user_id).first()
if profile and not profile.onboarding_completed: if profile and not profile.onboarding_completed:
logger.debug("[AUTH] oauth_callback: user=%s needs onboarding, redirecting", user_id)
post_onboarding = request.session.pop("redirect_after_login", "/upload") post_onboarding = request.session.pop("redirect_after_login", "/upload")
request.session["post_onboarding_redirect"] = post_onboarding request.session["post_onboarding_redirect"] = post_onboarding
return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND) return RedirectResponse(url="/onboarding", status_code=status.HTTP_302_FOUND)
# Redirect to original destination or default # Redirect to original destination or default
redirect_url = request.session.pop("redirect_after_login", "/upload") redirect_url = request.session.pop("redirect_after_login", "/upload")
logger.debug("[AUTH] oauth_callback: login complete, redirecting to %s", redirect_url)
return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND) return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
except Exception as e: except Exception as e:
logger.warning(f"[SECURITY] OAUTH_LOGIN_FAILURE error={type(e).__name__}") logger.warning(f"[SECURITY] OAUTH_LOGIN_FAILURE error={type(e).__name__}")
logger.debug("[AUTH] oauth_callback: full exception details", exc_info=True)
return RedirectResponse(url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND) return RedirectResponse(url=f"/login?error=Authentication+failed:+{str(e)}", status_code=status.HTTP_302_FOUND)
@@ -931,6 +1009,7 @@ async def logout(request: Request, db: Session = Depends(get_db)):
username = "unknown" username = "unknown"
if isinstance(user, dict): if isinstance(user, dict):
username = user.get("preferred_username") or user.get("email") or "unknown" username = user.get("preferred_username") or user.get("email") or "unknown"
logger.debug("[AUTH] logout: clearing session for user=%s client_ip=%s", username, get_client_ip(request))
logger.info(f"[SECURITY] LOGOUT user={username}") logger.info(f"[SECURITY] LOGOUT user={username}")
try: try:
from app.utils.audit_service import record_event from app.utils.audit_service import record_event
+14
View File
@@ -48,6 +48,20 @@ class Settings(BaseSettings):
workdir: str workdir: str
debug: bool = False # Default to False debug: bool = False # Default to False
# Logging level for the application. Accepts standard Python level names:
# DEBUG, INFO, WARNING, ERROR, CRITICAL. When *debug* is True and
# *log_level* has not been explicitly set, the effective level is forced to
# DEBUG so that all ``logger.debug()`` calls produce output.
log_level: str = Field(
default="INFO",
description=(
"Python logging level for the application root logger. "
"Accepts: DEBUG, INFO, WARNING, ERROR, CRITICAL. "
"When DEBUG=True and LOG_LEVEL is not explicitly set, "
"the effective level is automatically lowered to DEBUG."
),
)
# Making Dropbox optional # Making Dropbox optional
dropbox_enabled: bool = Field( dropbox_enabled: bool = Field(
default=True, default=True,
+44
View File
@@ -36,6 +36,50 @@ from app.views import router as frontend_router
# Explicitly include the files router # Explicitly include the files router
from app.views.files import router as files_router from app.views.files import router as files_router
# ---------------------------------------------------------------------------
# Configure Python root logging level early so that *all* loggers (including
# those already created via ``logging.getLogger(__name__)`` in other modules)
# respect the configured level.
#
# Standard behaviour (matches Django, Flask, 12-factor conventions):
# • ``LOG_LEVEL`` env var takes precedence when explicitly set.
# • When ``DEBUG=True`` and ``LOG_LEVEL`` is **not** set, the effective
# level is automatically lowered to ``DEBUG``.
# • Default (neither flag set): ``INFO``.
#
# Noisy third-party loggers (httpx, httpcore, authlib, etc.) are pinned to
# WARNING when the app-level is DEBUG to keep output useful.
# ---------------------------------------------------------------------------
_explicit_log_level = os.environ.get("LOG_LEVEL")
if settings.debug and _explicit_log_level is None:
_effective_level = "DEBUG"
else:
_effective_level = settings.log_level.upper()
_effective_level_int = getattr(logging, _effective_level, logging.INFO)
logging.basicConfig(
level=_effective_level_int,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
force=True,
)
# Keep noisy third-party loggers quiet at DEBUG level
if _effective_level_int <= logging.DEBUG:
for _noisy in (
"httpx",
"httpcore",
"authlib",
"urllib3",
"hpack",
"multipart",
"watchfiles",
):
logging.getLogger(_noisy).setLevel(logging.WARNING)
logging.getLogger(__name__).info("Root logging level set to %s (debug=%s)", _effective_level, settings.debug)
# Load configuration from .env for the session key # Load configuration from .env for the session key
config = Config(".env") config = Config(".env")
# Use settings.session_secret which has proper validation # Use settings.session_secret which has proper validation
+28
View File
@@ -457,6 +457,34 @@ default overage buffer applied across all plans.
DocuElevate supports HTTP security headers to improve browser-side security. **These headers are disabled by default** since most deployments use a reverse proxy (Traefik, Nginx, etc.) that already adds them. Enable only if deploying directly without a reverse proxy. See [Deployment Guide - Security Headers](DeploymentGuide.md#security-headers) for detailed configuration examples. DocuElevate supports HTTP security headers to improve browser-side security. **These headers are disabled by default** since most deployments use a reverse proxy (Traefik, Nginx, etc.) that already adds them. Enable only if deploying directly without a reverse proxy. See [Deployment Guide - Security Headers](DeploymentGuide.md#security-headers) for detailed configuration examples.
### Application Logging
DocuElevate uses Python's standard `logging` module. Two environment variables control log verbosity:
| **Variable** | **Description** | **Default** |
|-------------|----------------|-------------|
| `LOG_LEVEL` | Root logger level. Accepts standard Python level names: `DEBUG`, `INFO`, `WARNING`, `ERROR`, `CRITICAL`. | `INFO` |
| `DEBUG` | Enable debug mode. When `true` **and** `LOG_LEVEL` is **not** explicitly set, the effective log level is automatically lowered to `DEBUG`. | `false` |
**Precedence rules (standard behaviour):**
1. If `LOG_LEVEL` is explicitly set, it always wins — regardless of `DEBUG`.
2. If only `DEBUG=true` is set (no `LOG_LEVEL`), the effective level becomes `DEBUG`.
3. If neither is set, the default level is `INFO`.
```bash
# Typical production (default)
# LOG_LEVEL=INFO
# Quick debug mode — sets level to DEBUG automatically
DEBUG=true
# Explicit level override (DEBUG flag is ignored for level selection)
LOG_LEVEL=WARNING
```
> **Tip:** At `DEBUG` level, noisy third-party libraries (httpx, authlib, urllib3, etc.) are automatically pinned to `WARNING` so that application debug output remains readable.
### Audit Logging ### Audit Logging
DocuElevate provides comprehensive audit logging that records significant actions (logins, document CRUD, settings changes) to an append-only database table. Every entry captures the timestamp, user, action, resource, client IP, and optional JSON details. DocuElevate provides comprehensive audit logging that records significant actions (logins, document CRUD, settings changes) to an append-only database table. Every entry captures the timestamp, user, action, resource, client IP, and optional JSON details.
+35
View File
@@ -2,6 +2,7 @@
import asyncio import asyncio
import hashlib import hashlib
import logging
from unittest.mock import AsyncMock, MagicMock, patch from unittest.mock import AsyncMock, MagicMock, patch
import pytest import pytest
@@ -38,6 +39,40 @@ class TestGetCurrentUser:
result = get_current_user(mock_request) result = get_current_user(mock_request)
assert result is None assert result is None
def test_logs_debug_when_session_user_found(self, caplog):
"""Test that get_current_user emits a DEBUG log when session user is found."""
mock_request = MagicMock(spec=Request)
mock_request.session = {"user": {"id": "u1", "preferred_username": "alice"}}
mock_request.state = MagicMock(spec=[]) # no api_token_user attribute
with caplog.at_level(logging.DEBUG, logger="app.auth"):
get_current_user(mock_request)
assert any("[AUTH] get_current_user: resolved from session" in m for m in caplog.messages)
def test_logs_debug_when_no_user(self, caplog):
"""Test that get_current_user emits a DEBUG log when no user is present."""
mock_request = MagicMock(spec=Request)
mock_request.session = {}
mock_request.state = MagicMock(spec=[])
with caplog.at_level(logging.DEBUG, logger="app.auth"):
get_current_user(mock_request)
assert any("[AUTH] get_current_user: no user in session or API token" in m for m in caplog.messages)
def test_logs_debug_when_api_token_user(self, caplog):
"""Test that get_current_user emits a DEBUG log when resolved from API token."""
mock_request = MagicMock(spec=Request)
mock_request.state.api_token_user = {"id": "tok_user"}
mock_request.session = {}
with caplog.at_level(logging.DEBUG, logger="app.auth"):
result = get_current_user(mock_request)
assert result == {"id": "tok_user"}
assert any("[AUTH] get_current_user: resolved from API token" in m for m in caplog.messages)
@pytest.mark.unit @pytest.mark.unit
class TestGetGravatarUrl: class TestGetGravatarUrl:
+173
View File
@@ -0,0 +1,173 @@
"""Tests for application logging configuration.
Validates that the LOG_LEVEL and DEBUG settings correctly control the
Python root-logger level and that the standard precedence rules are respected:
1. Explicit LOG_LEVEL always wins.
2. DEBUG=True without LOG_LEVEL → effective DEBUG.
3. Neither set → default INFO.
"""
import logging
import os
from unittest.mock import patch
import pytest
from app.config import Settings
@pytest.mark.unit
class TestLogLevelSetting:
"""Tests for the log_level config field."""
_BASE_KWARGS = {
"database_url": "sqlite:///test.db",
"redis_url": "redis://localhost:6379",
"openai_api_key": "test",
"azure_ai_key": "test",
"azure_region": "test",
"azure_endpoint": "https://test.example.com",
"gotenberg_url": "http://localhost:3000",
"workdir": "/tmp",
"auth_enabled": False,
"session_secret": None,
}
def test_log_level_default_is_info(self):
"""Test that log_level defaults to INFO."""
config = Settings(**self._BASE_KWARGS)
assert config.log_level.upper() == "INFO"
def test_log_level_accepts_debug(self):
"""Test that log_level accepts DEBUG."""
config = Settings(**self._BASE_KWARGS, log_level="DEBUG")
assert config.log_level.upper() == "DEBUG"
def test_log_level_accepts_warning(self):
"""Test that log_level accepts WARNING."""
config = Settings(**self._BASE_KWARGS, log_level="WARNING")
assert config.log_level.upper() == "WARNING"
def test_log_level_accepts_error(self):
"""Test that log_level accepts ERROR."""
config = Settings(**self._BASE_KWARGS, log_level="ERROR")
assert config.log_level.upper() == "ERROR"
def test_log_level_case_insensitive(self):
"""Test that log_level is case-insensitive in usage."""
config = Settings(**self._BASE_KWARGS, log_level="debug")
assert config.log_level.upper() == "DEBUG"
def test_debug_flag_defaults_to_false(self):
"""Test that debug defaults to False."""
config = Settings(**self._BASE_KWARGS)
assert config.debug is False
@pytest.mark.unit
class TestEffectiveLogLevel:
"""Tests for the effective log-level resolution logic in main.py."""
def test_debug_true_without_log_level_gives_debug(self):
"""When DEBUG=True and LOG_LEVEL is not set, effective level is DEBUG."""
with patch.dict(os.environ, {"DEBUG": "true"}, clear=False):
# Remove LOG_LEVEL from env if present
env = os.environ.copy()
env.pop("LOG_LEVEL", None)
with patch.dict(os.environ, env, clear=True):
from app.config import Settings as S
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
debug=True,
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "DEBUG"
def test_explicit_log_level_overrides_debug(self):
"""When LOG_LEVEL is explicitly set, it takes precedence over DEBUG=True."""
with patch.dict(os.environ, {"LOG_LEVEL": "WARNING", "DEBUG": "true"}, clear=False):
from app.config import Settings as S
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
debug=True,
log_level="WARNING",
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "WARNING"
def test_default_no_flags_gives_info(self):
"""When neither DEBUG nor LOG_LEVEL is set, effective level is INFO."""
env = os.environ.copy()
env.pop("LOG_LEVEL", None)
env.pop("DEBUG", None)
with patch.dict(os.environ, env, clear=True):
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "INFO"
def test_effective_level_maps_to_logging_constant(self):
"""The effective level string maps to a valid logging constant."""
for level_name in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
assert getattr(logging, level_name) is not None
@pytest.mark.unit
class TestLoggingConfiguredAtStartup:
"""Tests that the main module configures the root logger on import."""
def test_root_logger_has_handler(self):
"""Root logger should have at least one handler after app import."""
root = logging.getLogger()
assert len(root.handlers) > 0, "Root logger has no handlers after app startup"
def test_root_logger_level_is_not_warning_default(self):
"""Root logger should not be at the unconfigured WARNING default.
Our basicConfig(force=True) should have set it to at least INFO.
"""
root = logging.getLogger()
# The test env doesn't set DEBUG=True, so the level should be INFO (20)
assert root.level <= logging.INFO