From a66eb4b2a2365691308fa4cf02a49c612038a6b5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 10 Mar 2026 21:59:58 +0000 Subject: [PATCH] fix: merge main into i18n branch and relink migration to 029 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Resolve conflict in app/api/__init__.py (keep both audit_logs_router and i18n_router) - Incorporate AuditLog model, audit_service, audit_logs API/views from main - Relink migration from 026→027 to 028→029 (chain after 028_add_audit_logs) - Update migrations/env.py with full model import list from main Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .env.demo | 16 + BUILD_DATE | 2 +- CHANGELOG.md | 22 + GIT_SHA | 2 +- RUNTIME_INFO | 12 +- VERSION | 2 +- app/api/__init__.py | 2 + app/api/audit_logs.py | 115 ++++ app/config.py | 43 ++ app/models.py | 21 + app/utils/audit_service.py | 331 ++++++++++ app/utils/db_migrate.py | 2 + app/utils/settings_service.py | 72 +++ app/views/__init__.py | 2 + app/views/audit_logs.py | 46 ++ docs/ConfigurationGuide.md | 55 ++ frontend/templates/audit_logs.html | 222 +++++++ frontend/templates/base.html | 3 + migrations/env.py | 16 + .../versions/027_ensure_shared_links_table.py | 62 ++ migrations/versions/028_add_audit_logs.py | 47 ++ ...py => 029_add_user_language_preference.py} | 8 +- tests/conftest.py | 1 + tests/test_audit_logs.py | 365 +++++++++++ tests/test_database.py | 79 +++ tests/test_views_coverage_boost.py | 608 ++++++++++++++++++ 26 files changed, 2143 insertions(+), 13 deletions(-) create mode 100644 app/api/audit_logs.py create mode 100644 app/utils/audit_service.py create mode 100644 app/views/audit_logs.py create mode 100644 frontend/templates/audit_logs.html create mode 100644 migrations/versions/027_ensure_shared_links_table.py create mode 100644 migrations/versions/028_add_audit_logs.py rename migrations/versions/{027_add_user_language_preference.py => 029_add_user_language_preference.py} (75%) create mode 100644 tests/test_audit_logs.py create mode 100644 tests/test_views_coverage_boost.py diff --git a/.env.demo b/.env.demo index 65f82fdf..f8a9a888 100644 --- a/.env.demo +++ b/.env.demo @@ -96,6 +96,22 @@ MAX_UPLOAD_SIZE=1073741824 # Allowed request headers (use * to allow all) # CORS_ALLOWED_HEADERS=* +# **Audit Logging & SIEM Integration** (see docs/ConfigurationGuide.md#audit-logging) +# Enable HTTP request audit logging middleware +AUDIT_LOGGING_ENABLED=true +# Include client IP in audit log entries (disable for GDPR-sensitive deployments) +AUDIT_LOG_INCLUDE_CLIENT_IP=true + +# Forward audit events to an external SIEM system (Syslog, Splunk, Logstash, Grafana, etc.) +# AUDIT_SIEM_ENABLED=false +# AUDIT_SIEM_TRANSPORT=syslog # syslog | http +# AUDIT_SIEM_SYSLOG_HOST=localhost +# AUDIT_SIEM_SYSLOG_PORT=514 +# AUDIT_SIEM_SYSLOG_PROTOCOL=udp # udp | tcp +# AUDIT_SIEM_HTTP_URL= # e.g. https://splunk:8088/services/collector/event +# AUDIT_SIEM_HTTP_TOKEN= # Bearer / HEC token +# AUDIT_SIEM_HTTP_CUSTOM_HEADERS= # Comma-separated Key:Value pairs + # **Rate Limiting** (see SECURITY_AUDIT.md and docs/API.md) # Protects against DoS attacks and API abuse by limiting request rates per IP/user # Enabled by default - highly recommended for production diff --git a/BUILD_DATE b/BUILD_DATE index ce1b9463..e91b5660 100644 --- a/BUILD_DATE +++ b/BUILD_DATE @@ -1 +1 @@ -2026-03-09T23:00:57Z +2026-03-10T21:51:18Z diff --git a/CHANGELOG.md b/CHANGELOG.md index 1568c12d..dccd7747 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 +## Unreleased + +### Testing + +- **views**: Add comprehensive coverage tests for 11 view modules + ([`c8987d9`](https://github.com/christianlouis/DocuElevate/commit/c8987d966bd949634d222a7809ec52c3ed258f81)) + +- **views**: Boost coverage for 11 view modules toward 100% + ([`6fb1df1`](https://github.com/christianlouis/DocuElevate/commit/6fb1df1aade8a8c98c27fe50dbe44045cba1724d)) + + +## v0.115.0 (2026-03-10) + + +## v0.114.1 (2026-03-10) + +### Bug Fixes + +- **db**: Add migration to create shared_links table for databases that skipped 025 + ([`289dcc3`](https://github.com/christianlouis/DocuElevate/commit/289dcc375c111c8d71bd04ef31f184a0e6a3f6f2)) + + ## v0.114.0 (2026-03-09) ### Bug Fixes diff --git a/GIT_SHA b/GIT_SHA index 082dbf0f..727fda95 100644 --- a/GIT_SHA +++ b/GIT_SHA @@ -1 +1 @@ -5fd3f06 +086793c diff --git a/RUNTIME_INFO b/RUNTIME_INFO index 0d52e60a..a6a0a496 100644 --- a/RUNTIME_INFO +++ b/RUNTIME_INFO @@ -1,10 +1,10 @@ DocuElevate Build Information ============================== -Version: 0.114.0 -Build Date: 2026-03-09T23:00:57Z -Git Commit: 5fd3f0661b8d86ba6b3f92481675d820aec0d53c -Git Short SHA: 5fd3f06 +Version: 0.115.0 +Build Date: 2026-03-10T21:51:18Z +Git Commit: 086793c05add0b48a938c9776110b9a1cce77d5e +Git Short SHA: 086793c Git Branch: main -Commit Date: 2026-03-10T00:00:39+01:00 -Build Timestamp: 2026-03-09T23:00:57Z +Commit Date: 2026-03-10T22:51:00+01:00 +Build Timestamp: 2026-03-10T21:51:18Z ============================== diff --git a/VERSION b/VERSION index 18455b77..bdc80994 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.114.0 +0.115.0 diff --git a/app/api/__init__.py b/app/api/__init__.py index aec13465..19aefb3c 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -8,6 +8,7 @@ from fastapi import APIRouter from app.api.admin_users import router as admin_users_router from app.api.api_tokens import router as api_tokens_router +from app.api.audit_logs import router as audit_logs_router from app.api.azure import router as azure_router from app.api.backup import router as backup_router from app.api.billing import router as billing_router @@ -83,4 +84,5 @@ router.include_router(imap_accounts_router) router.include_router(integrations_router) router.include_router(notifications_router) router.include_router(scheduled_jobs_router) +router.include_router(audit_logs_router) router.include_router(i18n_router) diff --git a/app/api/audit_logs.py b/app/api/audit_logs.py new file mode 100644 index 00000000..a4ed9d10 --- /dev/null +++ b/app/api/audit_logs.py @@ -0,0 +1,115 @@ +""" +Audit log REST API endpoints. + +Provides read-only access to the comprehensive audit log for admin users. +Events are append-only — there are no update or delete endpoints. +""" + +import logging +from datetime import datetime +from typing import Any + +from fastapi import APIRouter, Depends, Query, Request +from sqlalchemy.orm import Session + +from app.auth import require_login +from app.database import get_db +from app.utils.audit_service import count_events, query_events + +logger = logging.getLogger(__name__) + +router = APIRouter() + + +@router.get("/audit-logs") +@require_login +async def list_audit_logs( + request: Request, + db: Session = Depends(get_db), + action: str | None = Query(None, description="Filter by action (exact match)"), + user: str | None = Query(None, description="Filter by username"), + resource_type: str | None = Query(None, description="Filter by resource type"), + severity: str | None = Query(None, description="Filter by severity level"), + since: datetime | None = Query(None, description="Only events at or after this ISO-8601 timestamp"), + until: datetime | None = Query(None, description="Only events at or before this ISO-8601 timestamp"), + limit: int = Query(50, ge=1, le=500, description="Max rows to return"), + offset: int = Query(0, ge=0, description="Rows to skip for pagination"), +) -> dict[str, Any]: + """Return audit log entries with optional filtering and pagination. + + Requires authentication. Returns events in reverse chronological order. + """ + entries = query_events( + db, + action=action, + user=user, + resource_type=resource_type, + severity=severity, + since=since, + until=until, + limit=limit, + offset=offset, + ) + total = count_events( + db, + action=action, + user=user, + resource_type=resource_type, + severity=severity, + since=since, + until=until, + ) + return { + "items": [_serialize(e) for e in entries], + "total": total, + "limit": limit, + "offset": offset, + } + + +@router.get("/audit-logs/actions") +@require_login +async def list_distinct_actions( + request: Request, + db: Session = Depends(get_db), +) -> list[str]: + """Return the distinct action values present in the audit log.""" + from app.models import AuditLog + + rows = db.query(AuditLog.action).distinct().order_by(AuditLog.action).all() + return [r[0] for r in rows] + + +@router.get("/audit-logs/users") +@require_login +async def list_distinct_users( + request: Request, + db: Session = Depends(get_db), +) -> list[str]: + """Return the distinct user values present in the audit log.""" + from app.models import AuditLog + + rows = db.query(AuditLog.user).distinct().order_by(AuditLog.user).all() + return [r[0] for r in rows] + + +# ------------------------------------------------------------------ +# Helpers +# ------------------------------------------------------------------ + + +def _serialize(entry) -> dict[str, Any]: + """Convert an AuditLog row to a JSON-safe dict.""" + import json as _json + + return { + "id": entry.id, + "timestamp": entry.timestamp.isoformat() if entry.timestamp else None, + "user": entry.user, + "action": entry.action, + "resource_type": entry.resource_type, + "resource_id": entry.resource_id, + "ip_address": entry.ip_address, + "details": _json.loads(entry.details) if entry.details else None, + "severity": entry.severity, + } diff --git a/app/config.py b/app/config.py index ba6eeb25..51df35e0 100644 --- a/app/config.py +++ b/app/config.py @@ -849,6 +849,49 @@ class Settings(BaseSettings): ), ) + # SIEM / External Audit Log Forwarding + # Forward audit events to external SIEM systems for centralised monitoring. + audit_siem_enabled: bool = Field( + default=False, + description="Enable forwarding of audit events to an external SIEM system.", + ) + audit_siem_transport: str = Field( + default="syslog", + description=( + "Transport used to forward audit events. " + "Options: 'syslog' (RFC 5424 over UDP/TCP), 'http' (JSON POST to a webhook URL, " + "compatible with Splunk HEC, Logstash HTTP input, Grafana Loki, etc.)." + ), + ) + audit_siem_syslog_host: str = Field( + default="localhost", + description="Hostname or IP of the syslog receiver.", + ) + audit_siem_syslog_port: int = Field( + default=514, + description="Port of the syslog receiver.", + ) + audit_siem_syslog_protocol: str = Field( + default="udp", + description="Protocol for syslog transport: 'udp' or 'tcp'.", + ) + audit_siem_http_url: str = Field( + default="", + description=( + "HTTP endpoint URL for SIEM webhook delivery. " + "Supports Splunk HEC (https://splunk:8088/services/collector/event), " + "Logstash HTTP input, Grafana Loki push API, or any JSON-accepting endpoint." + ), + ) + audit_siem_http_token: str = Field( + default="", + description="Bearer / HEC token included in the Authorization header of SIEM HTTP requests.", + ) + audit_siem_http_custom_headers: str = Field( + default="", + description="Comma-separated 'Key:Value' pairs of extra headers for SIEM HTTP requests.", + ) + # UI / Appearance ui_default_color_scheme: str = Field( default="system", diff --git a/app/models.py b/app/models.py index 3a1703b5..4522d0b4 100644 --- a/app/models.py +++ b/app/models.py @@ -146,6 +146,27 @@ class SettingsAuditLog(Base): action = Column(String, nullable=False) # "update" or "delete" +class AuditLog(Base): + """Comprehensive audit log for compliance tracking. + + Records all significant actions: login/logout, document CRUD, settings + changes, and administrative operations. Rows are append-only; the API + and service layer never update or delete entries. + """ + + __tablename__ = "audit_logs" + + id = Column(Integer, primary_key=True, index=True) + timestamp = Column(DateTime(timezone=True), server_default=func.now(), nullable=False, index=True) + user = Column(String, nullable=False, index=True) # Username or "anonymous" / "system" + action = Column(String, nullable=False, index=True) # e.g. "login", "document.create", "settings.update" + resource_type = Column(String, nullable=True, index=True) # e.g. "document", "user", "settings" + resource_id = Column(String, nullable=True) # ID of the affected resource + ip_address = Column(String, nullable=True) # Client IP address + details = Column(Text, nullable=True) # JSON-encoded extra context + severity = Column(String(16), nullable=False, server_default="info") # info / warning / error / critical + + class SavedSearch(Base): """User-defined saved search filters for quick access to frequently used filter combinations.""" diff --git a/app/utils/audit_service.py b/app/utils/audit_service.py new file mode 100644 index 00000000..73a6af7c --- /dev/null +++ b/app/utils/audit_service.py @@ -0,0 +1,331 @@ +""" +Comprehensive audit-event service for DocuElevate. + +Provides helpers to **record** audit events (append-only database writes) +and to optionally **forward** them to external SIEM systems. + +Supported SIEM transports: +* **Syslog** – RFC 5424 structured-data messages over UDP or TCP. +* **HTTP** – JSON POST payloads compatible with Splunk HEC, Logstash + HTTP input, Grafana Loki push API, and any generic webhook endpoint. +""" + +import json +import logging +import re +import socket +import threading +from datetime import datetime, timezone +from typing import Any + +import httpx +from fastapi import Request +from sqlalchemy.orm import Session + +from app.config import settings +from app.middleware.audit_log import get_client_ip, get_username +from app.models import AuditLog + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Public helpers +# --------------------------------------------------------------------------- + + +def record_event( + db: Session, + *, + action: str, + user: str = "system", + resource_type: str | None = None, + resource_id: str | None = None, + ip_address: str | None = None, + details: dict[str, Any] | None = None, + severity: str = "info", +) -> AuditLog: + """Persist an audit event and optionally forward it to SIEM. + + Args: + db: Active SQLAlchemy session. + action: Short action identifier (e.g. ``"login"``, ``"document.create"``). + user: Username performing the action. + resource_type: Category of the affected resource (``"document"``, ``"user"`` …). + resource_id: Identifier of the affected resource. + ip_address: Client IP address (``None`` when not applicable). + details: Arbitrary key/value context serialised as JSON. + severity: One of ``info``, ``warning``, ``error``, ``critical``. + + Returns: + The newly created :class:`AuditLog` row. + """ + details_json = json.dumps(details, default=str) if details else None + + entry = AuditLog( + user=user, + action=action, + resource_type=resource_type, + resource_id=str(resource_id) if resource_id is not None else None, + ip_address=ip_address, + details=details_json, + severity=severity, + ) + db.add(entry) + db.commit() + db.refresh(entry) + + # Fire-and-forget SIEM forwarding in a background thread so we never + # block the request path. + if settings.audit_siem_enabled: + payload = _build_siem_payload(entry) + thread = threading.Thread(target=_forward_to_siem, args=(payload,), daemon=True) + thread.start() + + return entry + + +def record_event_from_request( + db: Session, + request: Request, + *, + action: str, + resource_type: str | None = None, + resource_id: str | None = None, + details: dict[str, Any] | None = None, + severity: str = "info", +) -> AuditLog: + """Convenience wrapper that extracts user and IP from a :class:`Request`. + + Args: + db: Active SQLAlchemy session. + request: The current HTTP request. + action: Short action identifier. + resource_type: Category of the affected resource. + resource_id: Identifier of the affected resource. + details: Arbitrary key/value context serialised as JSON. + severity: One of ``info``, ``warning``, ``error``, ``critical``. + + Returns: + The newly created :class:`AuditLog` row. + """ + return record_event( + db, + action=action, + user=get_username(request), + resource_type=resource_type, + resource_id=resource_id, + ip_address=get_client_ip(request), + details=details, + severity=severity, + ) + + +def query_events( + db: Session, + *, + action: str | None = None, + user: str | None = None, + resource_type: str | None = None, + severity: str | None = None, + since: datetime | None = None, + until: datetime | None = None, + limit: int = 200, + offset: int = 0, +) -> list[AuditLog]: + """Query audit log entries with optional filtering. + + Args: + db: Active SQLAlchemy session. + action: Filter by action string (exact match). + user: Filter by username (exact match). + resource_type: Filter by resource type (exact match). + severity: Filter by severity level (exact match). + since: Only events at or after this timestamp. + until: Only events at or before this timestamp. + limit: Maximum number of rows to return. + offset: Number of rows to skip (for pagination). + + Returns: + List of :class:`AuditLog` rows ordered by *timestamp descending*. + """ + q = db.query(AuditLog) + if action: + q = q.filter(AuditLog.action == action) + if user: + q = q.filter(AuditLog.user == user) + if resource_type: + q = q.filter(AuditLog.resource_type == resource_type) + if severity: + q = q.filter(AuditLog.severity == severity) + if since: + q = q.filter(AuditLog.timestamp >= since) + if until: + q = q.filter(AuditLog.timestamp <= until) + return q.order_by(AuditLog.timestamp.desc()).offset(offset).limit(limit).all() + + +def count_events( + db: Session, + *, + action: str | None = None, + user: str | None = None, + resource_type: str | None = None, + severity: str | None = None, + since: datetime | None = None, + until: datetime | None = None, +) -> int: + """Return the total count of events matching the given filters. + + Args: + db: Active SQLAlchemy session. + action: Filter by action string. + user: Filter by username. + resource_type: Filter by resource type. + severity: Filter by severity level. + since: Only events at or after this timestamp. + until: Only events at or before this timestamp. + + Returns: + Integer count. + """ + q = db.query(AuditLog) + if action: + q = q.filter(AuditLog.action == action) + if user: + q = q.filter(AuditLog.user == user) + if resource_type: + q = q.filter(AuditLog.resource_type == resource_type) + if severity: + q = q.filter(AuditLog.severity == severity) + if since: + q = q.filter(AuditLog.timestamp >= since) + if until: + q = q.filter(AuditLog.timestamp <= until) + return q.count() + + +# --------------------------------------------------------------------------- +# SIEM forwarding internals +# --------------------------------------------------------------------------- + +_SYSLOG_FACILITY_LOCAL0 = 16 +_SYSLOG_SEVERITY_MAP = { + "info": 6, + "warning": 4, + "error": 3, + "critical": 2, +} + + +def _build_siem_payload(entry: AuditLog) -> dict[str, Any]: + """Convert an :class:`AuditLog` row into a plain dict for SIEM delivery.""" + ts = entry.timestamp if entry.timestamp else datetime.now(timezone.utc) + return { + "id": entry.id, + "timestamp": ts.isoformat(), + "user": entry.user, + "action": entry.action, + "resource_type": entry.resource_type, + "resource_id": entry.resource_id, + "ip_address": entry.ip_address, + "details": entry.details, + "severity": entry.severity, + "source": "docuelevate", + } + + +def _forward_to_siem(payload: dict[str, Any]) -> None: + """Route a SIEM payload to the configured transport.""" + transport = settings.audit_siem_transport.lower() + try: + if transport == "syslog": + _send_syslog(payload) + elif transport == "http": + _send_http(payload) + else: + logger.warning("Unknown SIEM transport %r; skipping forwarding", transport) + except Exception: + logger.exception("Failed to forward audit event to SIEM (%s)", transport) + + +def _send_syslog(payload: dict[str, Any]) -> None: + """Send a RFC 5424 syslog message to the configured receiver.""" + severity_num = _SYSLOG_SEVERITY_MAP.get(payload.get("severity", "info"), 6) + priority = _SYSLOG_FACILITY_LOCAL0 * 8 + severity_num + ts = payload.get("timestamp", datetime.now(timezone.utc).isoformat()) + hostname = socket.gethostname() + app_name = "docuelevate" + msg_id = payload.get("action", "-") + + # Structured data (SD) element with key event fields. + sd = ( + f'[docuelevate@0 user="{payload.get("user", "-")}" ' + f'action="{payload.get("action", "-")}" ' + f'resource_type="{payload.get("resource_type", "-")}" ' + f'resource_id="{payload.get("resource_id", "-")}" ' + f'ip="{payload.get("ip_address", "-")}"]' + ) + message = json.dumps(payload, default=str) + syslog_msg = f"<{priority}>1 {ts} {hostname} {app_name} - {msg_id} {sd} {message}" + + proto = settings.audit_siem_syslog_protocol.lower() + host = settings.audit_siem_syslog_host + port = settings.audit_siem_syslog_port + + if proto == "tcp": + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(5) + sock.connect((host, port)) + sock.sendall(syslog_msg.encode("utf-8")) + else: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: + sock.settimeout(5) + sock.sendto(syslog_msg.encode("utf-8"), (host, port)) + + logger.debug("Syslog audit event sent to %s:%s (%s)", host, port, proto) + + +def _send_http(payload: dict[str, Any]) -> None: + """POST a JSON audit event to the configured HTTP endpoint.""" + url = settings.audit_siem_http_url + if not url: + logger.warning("SIEM HTTP URL not configured; skipping HTTP forwarding") + return + + headers: dict[str, str] = {"Content-Type": "application/json"} + token = settings.audit_siem_http_token + if token: + headers["Authorization"] = f"Bearer {token}" + + # Parse custom headers (comma-separated "Key:Value" pairs). + # Reject headers that could override security-critical ones already set, + # and validate that header names contain only RFC 7230 token characters. + _PROTECTED_HEADERS = {"authorization", "content-type", "host"} + _VALID_HEADER_NAME = re.compile(r"^[A-Za-z0-9!#$%&'*+\-.^_`|~]+$") + raw_custom = settings.audit_siem_http_custom_headers + if raw_custom: + for raw_pair in raw_custom.split(","): + pair = raw_pair.strip() + if ":" in pair: + k, _, v = pair.partition(":") + name = k.strip() + if not name or not _VALID_HEADER_NAME.match(name): + logger.warning("Skipping invalid SIEM custom header name: %r", name) + continue + if name.lower() in _PROTECTED_HEADERS: + logger.warning("Skipping protected SIEM custom header: %r", name) + continue + headers[name] = v.strip() + + # Wrap in Splunk HEC-style envelope when URL contains ``/services/collector``. + body: dict[str, Any] + if "/services/collector" in url: + body = {"event": payload, "sourcetype": "docuelevate:audit", "source": "docuelevate"} + else: + body = payload + + with httpx.Client(timeout=10) as client: + resp = client.post(url, json=body, headers=headers) + resp.raise_for_status() + + logger.debug("HTTP audit event forwarded to %s (status %s)", url, resp.status_code) diff --git a/app/utils/db_migrate.py b/app/utils/db_migrate.py index f95d97f0..3424009d 100644 --- a/app/utils/db_migrate.py +++ b/app/utils/db_migrate.py @@ -32,8 +32,10 @@ _TABLE_ORDER = [ "processing_logs", "application_settings", "settings_audit_log", + "audit_logs", "saved_searches", "webhook_configs", + "shared_links", ] diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index 7516f8f9..39ed812e 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -2065,6 +2065,78 @@ SETTING_METADATA = { "required": False, "restart_required": True, }, + "audit_siem_enabled": { + "category": "Security", + "description": "Enable forwarding of audit events to an external SIEM system (Syslog, Splunk, Logstash, etc.).", + "type": "boolean", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "audit_siem_transport": { + "category": "Security", + "description": ( + "Transport used to forward audit events. 'syslog' sends RFC 5424 messages over UDP/TCP. " + "'http' sends JSON POST payloads to a webhook URL (Splunk HEC, Logstash, Grafana Loki, etc.)." + ), + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + "options": ["syslog", "http"], + }, + "audit_siem_syslog_host": { + "category": "Security", + "description": "Hostname or IP of the syslog receiver.", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "audit_siem_syslog_port": { + "category": "Security", + "description": "Port of the syslog receiver. Default: 514.", + "type": "integer", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "audit_siem_syslog_protocol": { + "category": "Security", + "description": "Protocol for syslog transport: 'udp' or 'tcp'. Default: udp.", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + "options": ["udp", "tcp"], + }, + "audit_siem_http_url": { + "category": "Security", + "description": ( + "HTTP endpoint URL for SIEM webhook delivery. Supports Splunk HEC, " + "Logstash HTTP input, Grafana Loki push API, or any JSON-accepting endpoint." + ), + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "audit_siem_http_token": { + "category": "Security", + "description": "Bearer / HEC token included in the Authorization header of SIEM HTTP requests.", + "type": "string", + "sensitive": True, + "required": False, + "restart_required": False, + }, + "audit_siem_http_custom_headers": { + "category": "Security", + "description": "Comma-separated 'Key:Value' pairs of extra headers for SIEM HTTP requests.", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, # Rate Limiting "rate_limiting_enabled": { "category": "Security", diff --git a/app/views/__init__.py b/app/views/__init__.py index 5c098527..1b2b0443 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -6,6 +6,7 @@ from fastapi import APIRouter from app.views.admin_users import router as admin_users_router from app.views.api_tokens import router as api_tokens_router +from app.views.audit_logs import router as audit_logs_router from app.views.backup import router as backup_router from app.views.db_wizard import router as db_wizard_router from app.views.dropbox import router as dropbox_router @@ -60,4 +61,5 @@ router.include_router(imap_accounts_router) # Per-user IMAP ingestion accounts router.include_router(integrations_router) # Unified integrations dashboard router.include_router(notifications_router) # User notification dashboard router.include_router(scheduled_jobs_router) # Admin scheduled batch jobs +router.include_router(audit_logs_router) # Comprehensive audit log viewer router.include_router(help_router) # Built-in help / How-To docs diff --git a/app/views/audit_logs.py b/app/views/audit_logs.py new file mode 100644 index 00000000..a787fece --- /dev/null +++ b/app/views/audit_logs.py @@ -0,0 +1,46 @@ +""" +Audit log viewer UI — admin-only page with filtering and SIEM status. +""" + +import logging + +from fastapi import Depends, HTTPException, Request, status +from sqlalchemy.orm import Session + +from app.views.base import APIRouter, get_db, require_login, settings, templates +from app.views.settings import require_admin_access + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/admin/audit-logs") +@require_login +@require_admin_access +async def audit_logs_page(request: Request, db: Session = Depends(get_db)): + """Comprehensive audit log viewer with filtering controls. + + Displays a chronological log of all significant actions: logins, + document operations, settings changes, and admin actions. The + actual data is fetched client-side via the ``/api/audit-logs`` JSON + endpoint so that filters, pagination, and live refresh work without + full-page reloads. + """ + try: + siem_enabled = settings.audit_siem_enabled + siem_transport = settings.audit_siem_transport if siem_enabled else None + return templates.TemplateResponse( + "audit_logs.html", + { + "request": request, + "app_version": settings.version, + "siem_enabled": siem_enabled, + "siem_transport": siem_transport, + }, + ) + except Exception as e: + logger.error("Error loading audit logs page: %s", e) + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="Failed to load audit logs page", + ) diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index 009dc5a9..131324e4 100644 --- a/docs/ConfigurationGuide.md +++ b/docs/ConfigurationGuide.md @@ -398,6 +398,61 @@ 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. +### 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. + +| **Variable** | **Description** | **Default** | +|--------------------------------|---------------------------------------------------------------------------------------------------|-------------| +| `AUDIT_LOGGING_ENABLED` | Enable the HTTP request audit-logging middleware. | `true` | +| `AUDIT_LOG_INCLUDE_CLIENT_IP` | Include the client IP address in audit log entries. Disable for GDPR-sensitive deployments. | `true` | + +#### SIEM Integration + +Audit events can be forwarded in real time to external SIEM systems for centralised monitoring, alerting, and long-term retention. Two transports are supported: + +* **Syslog** – RFC 5424 structured-data messages over UDP or TCP. Works with rsyslog, syslog-ng, Graylog, Datadog, etc. +* **HTTP** – JSON POST payloads compatible with Splunk HEC, Logstash HTTP input, Grafana Loki push API, and any generic webhook. + +| **Variable** | **Description** | **Default** | +|-------------------------------------|---------------------------------------------------------------------------------------------------|---------------| +| `AUDIT_SIEM_ENABLED` | Enable forwarding of audit events to an external SIEM system. | `false` | +| `AUDIT_SIEM_TRANSPORT` | Transport: `syslog` or `http`. | `syslog` | +| `AUDIT_SIEM_SYSLOG_HOST` | Hostname or IP of the syslog receiver. | `localhost` | +| `AUDIT_SIEM_SYSLOG_PORT` | Port of the syslog receiver. | `514` | +| `AUDIT_SIEM_SYSLOG_PROTOCOL` | Protocol for syslog: `udp` or `tcp`. | `udp` | +| `AUDIT_SIEM_HTTP_URL` | HTTP endpoint URL for SIEM delivery (e.g. Splunk HEC, Logstash, Loki). | *(empty)* | +| `AUDIT_SIEM_HTTP_TOKEN` | Bearer / HEC token for the SIEM HTTP endpoint. | *(empty)* | +| `AUDIT_SIEM_HTTP_CUSTOM_HEADERS` | Comma-separated `Key:Value` extra headers for SIEM HTTP requests. | *(empty)* | + +**Example – Syslog to rsyslog:** + +```bash +AUDIT_SIEM_ENABLED=true +AUDIT_SIEM_TRANSPORT=syslog +AUDIT_SIEM_SYSLOG_HOST=syslog.internal.example.com +AUDIT_SIEM_SYSLOG_PORT=514 +AUDIT_SIEM_SYSLOG_PROTOCOL=udp +``` + +**Example – Splunk HEC:** + +```bash +AUDIT_SIEM_ENABLED=true +AUDIT_SIEM_TRANSPORT=http +AUDIT_SIEM_HTTP_URL=https://splunk.example.com:8088/services/collector/event +AUDIT_SIEM_HTTP_TOKEN=your-hec-token +``` + +**Example – Logstash HTTP input:** + +```bash +AUDIT_SIEM_ENABLED=true +AUDIT_SIEM_TRANSPORT=http +AUDIT_SIEM_HTTP_URL=https://logstash.example.com:8080 +AUDIT_SIEM_HTTP_TOKEN= +``` + ### Rate Limiting DocuElevate implements rate limiting to protect against DoS attacks and API abuse. **Rate limiting is enabled by default** and uses Redis for distributed rate limiting across multiple workers. diff --git a/frontend/templates/audit_logs.html b/frontend/templates/audit_logs.html new file mode 100644 index 00000000..92073816 --- /dev/null +++ b/frontend/templates/audit_logs.html @@ -0,0 +1,222 @@ +{% extends "base.html" %} + +{% block title %}Audit Logs - DocuElevate{% endblock %} + +{% block content %} +
+ + +
+
+

+ Audit Logs +

+

+ Comprehensive, append-only record of all significant actions. +

+
+
+ {% if siem_enabled %} + + SIEM: {{ siem_transport|upper }} + + {% else %} + + SIEM: Off + + {% endif %} + +
+
+ + +
+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+
+ + +
+ + Loading… +
+ + +
+
+ + + + + + + + + + + + + + + + + + +
TimestampSeverityUserActionResourceIPDetails
+ +

No audit events recorded yet.

+

Significant actions (logins, document operations, settings changes) will appear here.

+
+
+
+ + + + +
+ + +{% endblock %} diff --git a/frontend/templates/base.html b/frontend/templates/base.html index 1ccdff9f..416c40e5 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -171,6 +171,9 @@ {{ _("nav.backup_restore") }} + + Audit Logs + {{ _("nav.status") }} diff --git a/migrations/env.py b/migrations/env.py index 903382a5..67fc3ad3 100644 --- a/migrations/env.py +++ b/migrations/env.py @@ -20,13 +20,29 @@ from app.database import Base # Ensure all models are imported so Base.metadata is populated. from app.models import ( # noqa: F401 + ApiToken, ApplicationSettings, + AuditLog, + BackupRecord, DocumentMetadata, FileProcessingStep, FileRecord, + InAppNotification, + LocalUser, + Pipeline, + PipelineStep, ProcessingLog, SavedSearch, + ScheduledJob, SettingsAuditLog, + SharedLink, + SubscriptionPlan, + UserImapAccount, + UserIntegration, + UserNotificationPreference, + UserNotificationTarget, + UserProfile, + WebhookConfig, ) # Alembic Config object – provides access to values in alembic.ini. diff --git a/migrations/versions/027_ensure_shared_links_table.py b/migrations/versions/027_ensure_shared_links_table.py new file mode 100644 index 00000000..2a333582 --- /dev/null +++ b/migrations/versions/027_ensure_shared_links_table.py @@ -0,0 +1,62 @@ +"""Ensure shared_links table exists for databases that skipped migration 025. + +Databases that were already at revision 025_add_user_notifications or +026_add_scheduled_jobs before 025_add_shared_links was inserted into the +migration chain will never have had the ``shared_links`` table created. +This migration creates the table idempotently so those databases are +repaired on the next ``alembic upgrade head``. + +Revision ID: 027_ensure_shared_links_table +Revises: 026_add_scheduled_jobs +Create Date: 2026-03-09 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "027_ensure_shared_links_table" +down_revision: Union[str, None] = "026_add_scheduled_jobs" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create shared_links table if it does not already exist.""" + conn = op.get_bind() + inspector = sa.inspect(conn) + if "shared_links" not in inspector.get_table_names(): + op.create_table( + "shared_links", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("token", sa.String(64), nullable=False), + sa.Column("file_id", sa.Integer(), nullable=False), + sa.Column("owner_id", sa.String(), nullable=False), + sa.Column("label", sa.String(255), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("max_views", sa.Integer(), nullable=True), + sa.Column("view_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("password_hash", sa.String(128), nullable=True), + sa.Column("is_active", sa.Boolean(), nullable=False, server_default="1"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["file_id"], ["files.id"]), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("token"), + ) + op.create_index("ix_shared_links_id", "shared_links", ["id"]) + op.create_index("ix_shared_links_token", "shared_links", ["token"]) + op.create_index("ix_shared_links_file_id", "shared_links", ["file_id"]) + op.create_index("ix_shared_links_owner_id", "shared_links", ["owner_id"]) + + +def downgrade() -> None: + """Drop shared_links table only if this migration created it.""" + conn = op.get_bind() + inspector = sa.inspect(conn) + if "shared_links" in inspector.get_table_names(): + op.drop_index("ix_shared_links_owner_id", "shared_links") + op.drop_index("ix_shared_links_file_id", "shared_links") + op.drop_index("ix_shared_links_token", "shared_links") + op.drop_index("ix_shared_links_id", "shared_links") + op.drop_table("shared_links") diff --git a/migrations/versions/028_add_audit_logs.py b/migrations/versions/028_add_audit_logs.py new file mode 100644 index 00000000..58114e7a --- /dev/null +++ b/migrations/versions/028_add_audit_logs.py @@ -0,0 +1,47 @@ +"""Add audit_logs table for comprehensive compliance audit logging. + +Revision ID: 028_add_audit_logs +Revises: 027_ensure_shared_links_table +Create Date: 2026-03-09 +""" + +from typing import Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "028_add_audit_logs" +down_revision: Union[str, None] = "027_ensure_shared_links_table" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Create audit_logs table.""" + op.create_table( + "audit_logs", + sa.Column("id", sa.Integer(), nullable=False), + sa.Column("timestamp", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("user", sa.String(), nullable=False), + sa.Column("action", sa.String(), nullable=False), + sa.Column("resource_type", sa.String(), nullable=True), + sa.Column("resource_id", sa.String(), nullable=True), + sa.Column("ip_address", sa.String(), nullable=True), + sa.Column("details", sa.Text(), nullable=True), + sa.Column("severity", sa.String(16), nullable=False, server_default="info"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_audit_logs_id", "audit_logs", ["id"]) + op.create_index("ix_audit_logs_timestamp", "audit_logs", ["timestamp"]) + op.create_index("ix_audit_logs_user", "audit_logs", ["user"]) + op.create_index("ix_audit_logs_action", "audit_logs", ["action"]) + op.create_index("ix_audit_logs_resource_type", "audit_logs", ["resource_type"]) + + +def downgrade() -> None: + """Drop audit_logs table.""" + op.drop_index("ix_audit_logs_resource_type", "audit_logs") + op.drop_index("ix_audit_logs_action", "audit_logs") + op.drop_index("ix_audit_logs_user", "audit_logs") + op.drop_index("ix_audit_logs_timestamp", "audit_logs") + op.drop_index("ix_audit_logs_id", "audit_logs") + op.drop_table("audit_logs") diff --git a/migrations/versions/027_add_user_language_preference.py b/migrations/versions/029_add_user_language_preference.py similarity index 75% rename from migrations/versions/027_add_user_language_preference.py rename to migrations/versions/029_add_user_language_preference.py index 801801b3..632f0040 100644 --- a/migrations/versions/027_add_user_language_preference.py +++ b/migrations/versions/029_add_user_language_preference.py @@ -1,7 +1,7 @@ """Add preferred_language column to user_profiles for i18n support. -Revision ID: 027_add_user_language_preference -Revises: 026_add_scheduled_jobs +Revision ID: 029_add_user_language_preference +Revises: 028_add_audit_logs Create Date: 2026-03-09 """ @@ -10,8 +10,8 @@ from typing import Union import sqlalchemy as sa from alembic import op -revision: str = "027_add_user_language_preference" -down_revision: Union[str, None] = "026_add_scheduled_jobs" +revision: str = "029_add_user_language_preference" +down_revision: Union[str, None] = "028_add_audit_logs" depends_on: Union[str, None] = None diff --git a/tests/conftest.py b/tests/conftest.py index ae6db91e..cce347f7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -61,6 +61,7 @@ from app.main import app as fastapi_app # noqa: E402 # Import models to register them with SQLAlchemy Base from app.models import ( # noqa: F401, E402 ApiToken, + AuditLog, DocumentMetadata, FileRecord, Pipeline, diff --git a/tests/test_audit_logs.py b/tests/test_audit_logs.py new file mode 100644 index 00000000..1671fd47 --- /dev/null +++ b/tests/test_audit_logs.py @@ -0,0 +1,365 @@ +""" +Tests for the comprehensive audit logging feature. + +Covers the audit service (recording, querying, SIEM forwarding), +the REST API endpoints, and the admin viewer page. +""" + +import json +import socket +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.database import Base +from app.models import AuditLog + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def audit_db(): + """Fresh in-memory database with all tables created.""" + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + Session = sessionmaker(bind=engine) + session = Session() + yield session + session.close() + Base.metadata.drop_all(bind=engine) + + +# --------------------------------------------------------------------------- +# Model tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestAuditLogModel: + """Verify the AuditLog ORM model.""" + + def test_create_minimal_entry(self, audit_db): + """Minimal required fields can be persisted.""" + entry = AuditLog(user="alice", action="login") + audit_db.add(entry) + audit_db.commit() + audit_db.refresh(entry) + assert entry.id is not None + assert entry.user == "alice" + assert entry.action == "login" + assert entry.severity == "info" # server default + + def test_create_full_entry(self, audit_db): + """All columns persist correctly.""" + entry = AuditLog( + user="bob", + action="document.create", + resource_type="document", + resource_id="42", + ip_address="10.0.0.1", + details='{"filename": "invoice.pdf"}', + severity="warning", + ) + audit_db.add(entry) + audit_db.commit() + audit_db.refresh(entry) + assert entry.resource_type == "document" + assert entry.resource_id == "42" + assert entry.ip_address == "10.0.0.1" + assert json.loads(entry.details) == {"filename": "invoice.pdf"} + assert entry.severity == "warning" + + +# --------------------------------------------------------------------------- +# Service tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestAuditService: + """Verify the audit_service helper functions.""" + + @patch("app.utils.audit_service.settings") + def test_record_event(self, mock_settings, audit_db): + """record_event persists a row and returns the entry.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import record_event + + entry = record_event( + audit_db, + action="settings.update", + user="admin", + resource_type="settings", + resource_id="openai_model", + details={"old": "gpt-4", "new": "gpt-4o"}, + ) + assert entry.id is not None + assert entry.action == "settings.update" + assert entry.user == "admin" + + @patch("app.utils.audit_service.settings") + def test_query_events_no_filter(self, mock_settings, audit_db): + """query_events returns all events when no filter is supplied.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import query_events, record_event + + for i in range(5): + record_event(audit_db, action=f"action_{i}", user="sys") + results = query_events(audit_db) + assert len(results) == 5 + + @patch("app.utils.audit_service.settings") + def test_query_events_filter_action(self, mock_settings, audit_db): + """query_events filters by action.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import query_events, record_event + + record_event(audit_db, action="login", user="alice") + record_event(audit_db, action="logout", user="alice") + results = query_events(audit_db, action="login") + assert len(results) == 1 + assert results[0].action == "login" + + @patch("app.utils.audit_service.settings") + def test_query_events_filter_user(self, mock_settings, audit_db): + """query_events filters by user.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import query_events, record_event + + record_event(audit_db, action="login", user="alice") + record_event(audit_db, action="login", user="bob") + results = query_events(audit_db, user="bob") + assert len(results) == 1 + + @patch("app.utils.audit_service.settings") + def test_query_events_filter_severity(self, mock_settings, audit_db): + """query_events filters by severity.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import query_events, record_event + + record_event(audit_db, action="fail", user="sys", severity="error") + record_event(audit_db, action="ok", user="sys", severity="info") + results = query_events(audit_db, severity="error") + assert len(results) == 1 + assert results[0].severity == "error" + + @patch("app.utils.audit_service.settings") + def test_count_events(self, mock_settings, audit_db): + """count_events returns the correct total.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import count_events, record_event + + for _ in range(3): + record_event(audit_db, action="ping", user="sys") + assert count_events(audit_db) == 3 + assert count_events(audit_db, action="ping") == 3 + assert count_events(audit_db, action="pong") == 0 + + @patch("app.utils.audit_service.settings") + def test_query_events_pagination(self, mock_settings, audit_db): + """query_events respects limit and offset.""" + mock_settings.audit_siem_enabled = False + from app.utils.audit_service import query_events, record_event + + for i in range(10): + record_event(audit_db, action=f"a{i}", user="sys") + page1 = query_events(audit_db, limit=3, offset=0) + page2 = query_events(audit_db, limit=3, offset=3) + assert len(page1) == 3 + assert len(page2) == 3 + assert page1[0].id != page2[0].id + + +# --------------------------------------------------------------------------- +# SIEM forwarding tests +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestSIEMForwarding: + """Verify SIEM transport helpers.""" + + @patch("app.utils.audit_service.settings") + def test_build_siem_payload(self, mock_settings): + """_build_siem_payload returns a dict with all expected keys.""" + from app.utils.audit_service import _build_siem_payload + + entry = AuditLog( + id=1, + user="admin", + action="login", + resource_type="session", + timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + severity="info", + ) + payload = _build_siem_payload(entry) + assert payload["user"] == "admin" + assert payload["action"] == "login" + assert payload["source"] == "docuelevate" + assert "timestamp" in payload + + @patch("app.utils.audit_service.settings") + @patch("app.utils.audit_service.socket") + def test_send_syslog_udp(self, mock_socket_mod, mock_settings): + """_send_syslog sends a UDP datagram to the configured host.""" + mock_settings.audit_siem_syslog_protocol = "udp" + mock_settings.audit_siem_syslog_host = "127.0.0.1" + mock_settings.audit_siem_syslog_port = 5140 + + mock_sock = MagicMock() + mock_socket_mod.AF_INET = socket.AF_INET + mock_socket_mod.SOCK_DGRAM = socket.SOCK_DGRAM + mock_socket_mod.gethostname.return_value = "test-host" + mock_socket_mod.socket.return_value.__enter__ = MagicMock(return_value=mock_sock) + mock_socket_mod.socket.return_value.__exit__ = MagicMock(return_value=False) + + from app.utils.audit_service import _send_syslog + + _send_syslog({"user": "test", "action": "login", "severity": "info", "timestamp": "2026-01-01T00:00:00"}) + mock_sock.sendto.assert_called_once() + + @patch("app.utils.audit_service.settings") + @patch("app.utils.audit_service.httpx") + def test_send_http_generic(self, mock_httpx, mock_settings): + """_send_http POSTs JSON to a generic endpoint.""" + mock_settings.audit_siem_http_url = "https://siem.example.com/ingest" + mock_settings.audit_siem_http_token = "my-token" + mock_settings.audit_siem_http_custom_headers = "" + + mock_client = MagicMock() + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_client.post.return_value = mock_resp + mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client) + mock_httpx.Client.return_value.__exit__ = MagicMock(return_value=False) + + from app.utils.audit_service import _send_http + + _send_http({"user": "test", "action": "login"}) + mock_client.post.assert_called_once() + call_kwargs = mock_client.post.call_args + assert call_kwargs.kwargs["headers"]["Authorization"] == "Bearer my-token" + + @patch("app.utils.audit_service.settings") + @patch("app.utils.audit_service.httpx") + def test_send_http_splunk_hec(self, mock_httpx, mock_settings): + """_send_http wraps payload in Splunk HEC envelope when URL contains /services/collector.""" + mock_settings.audit_siem_http_url = "https://splunk:8088/services/collector/event" + mock_settings.audit_siem_http_token = "hec-token" + mock_settings.audit_siem_http_custom_headers = "" + + mock_client = MagicMock() + mock_resp = MagicMock() + mock_resp.status_code = 200 + mock_client.post.return_value = mock_resp + mock_httpx.Client.return_value.__enter__ = MagicMock(return_value=mock_client) + mock_httpx.Client.return_value.__exit__ = MagicMock(return_value=False) + + from app.utils.audit_service import _send_http + + _send_http({"user": "test", "action": "login"}) + call_kwargs = mock_client.post.call_args + body = call_kwargs.kwargs["json"] + assert "event" in body + assert body["sourcetype"] == "docuelevate:audit" + + +# --------------------------------------------------------------------------- +# API endpoint tests +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestAuditLogAPI: + """Test /api/audit-logs REST endpoints.""" + + def test_list_audit_logs_empty(self, client): + """GET /api/audit-logs returns empty list when no events exist.""" + resp = client.get("/api/audit-logs") + assert resp.status_code == 200 + data = resp.json() + assert data["items"] == [] + assert data["total"] == 0 + + def test_list_audit_logs_with_data(self, client, db_session): + """GET /api/audit-logs returns recorded events.""" + entry = AuditLog(user="tester", action="test.action", severity="info") + db_session.add(entry) + db_session.commit() + + resp = client.get("/api/audit-logs") + assert resp.status_code == 200 + data = resp.json() + assert data["total"] == 1 + assert data["items"][0]["action"] == "test.action" + + def test_list_audit_logs_filter_by_action(self, client, db_session): + """GET /api/audit-logs?action=x filters correctly.""" + db_session.add(AuditLog(user="a", action="login", severity="info")) + db_session.add(AuditLog(user="a", action="logout", severity="info")) + db_session.commit() + + resp = client.get("/api/audit-logs?action=login") + assert resp.status_code == 200 + data = resp.json() + assert data["total"] == 1 + + def test_list_distinct_actions(self, client, db_session): + """GET /api/audit-logs/actions returns distinct action values.""" + db_session.add(AuditLog(user="a", action="login", severity="info")) + db_session.add(AuditLog(user="b", action="login", severity="info")) + db_session.add(AuditLog(user="a", action="logout", severity="info")) + db_session.commit() + + resp = client.get("/api/audit-logs/actions") + assert resp.status_code == 200 + actions = resp.json() + assert set(actions) == {"login", "logout"} + + def test_list_distinct_users(self, client, db_session): + """GET /api/audit-logs/users returns distinct user values.""" + db_session.add(AuditLog(user="alice", action="x", severity="info")) + db_session.add(AuditLog(user="bob", action="x", severity="info")) + db_session.commit() + + resp = client.get("/api/audit-logs/users") + assert resp.status_code == 200 + users = resp.json() + assert set(users) == {"alice", "bob"} + + def test_list_audit_logs_pagination(self, client, db_session): + """GET /api/audit-logs supports limit/offset pagination.""" + for i in range(5): + db_session.add(AuditLog(user="u", action=f"a{i}", severity="info")) + db_session.commit() + + resp = client.get("/api/audit-logs?limit=2&offset=0") + data = resp.json() + assert len(data["items"]) == 2 + assert data["total"] == 5 + + +# --------------------------------------------------------------------------- +# View tests +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestAuditLogView: + """Test the admin audit-log viewer page.""" + + def test_audit_logs_page_loads(self, client): + """GET /admin/audit-logs returns 200 and renders the template.""" + resp = client.get("/admin/audit-logs") + assert resp.status_code == 200 + assert "Audit Logs" in resp.text diff --git a/tests/test_database.py b/tests/test_database.py index a49dd0a5..14e803c0 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -124,6 +124,85 @@ class TestInitDb: test_engine.dispose() + def test_init_db_creates_shared_links_for_database_missing_table(self, tmp_path): + """Regression test: databases at revision 026 that skipped 025_add_shared_links. + + Migration 025_add_shared_links was inserted into the chain between + 024_add_api_tokens and 025_add_user_notifications after some databases + had already been migrated past that point. Migration 027 creates the + table idempotently so those databases are repaired. + """ + from sqlalchemy import create_engine, text + from sqlalchemy import inspect as sa_inspect + + db_path = str(tmp_path / "regression_shared_links.db") + test_engine = create_engine(f"sqlite:///{db_path}") + + # Set up a database at revision 026 but WITHOUT the shared_links table. + # This simulates a DB that was migrated before 025_add_shared_links + # was inserted into the chain. + with test_engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE files (" + "id INTEGER PRIMARY KEY, filehash VARCHAR NOT NULL, " + "original_filename VARCHAR, local_filename VARCHAR NOT NULL, " + "original_file_path VARCHAR, processed_file_path VARCHAR, " + "file_size INTEGER NOT NULL, mime_type VARCHAR, " + "is_duplicate BOOLEAN DEFAULT 0 NOT NULL, duplicate_of_id INTEGER, " + "ocr_text TEXT, ai_metadata TEXT, document_title VARCHAR, " + "ocr_quality_score INTEGER, created_at DATETIME DEFAULT CURRENT_TIMESTAMP)" + ) + ) + conn.execute( + text( + "CREATE TABLE processing_logs (" + "id INTEGER PRIMARY KEY, file_id INTEGER, task_id VARCHAR, " + "step_name VARCHAR, status VARCHAR, message VARCHAR, detail TEXT, " + "timestamp DATETIME DEFAULT CURRENT_TIMESTAMP)" + ) + ) + conn.execute( + text( + "CREATE TABLE file_processing_steps (" + "id INTEGER PRIMARY KEY, file_id INTEGER NOT NULL, " + "step_name VARCHAR NOT NULL, status VARCHAR NOT NULL, " + "started_at DATETIME, completed_at DATETIME, error_message TEXT, " + "created_at DATETIME DEFAULT CURRENT_TIMESTAMP, " + "updated_at DATETIME DEFAULT CURRENT_TIMESTAMP)" + ) + ) + conn.execute( + text( + "CREATE TABLE saved_searches (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, user_id VARCHAR NOT NULL, " + "name VARCHAR NOT NULL, filters TEXT NOT NULL, " + "created_at DATETIME DEFAULT CURRENT_TIMESTAMP, " + "updated_at DATETIME DEFAULT CURRENT_TIMESTAMP, " + "UNIQUE (user_id, name))" + ) + ) + conn.execute(text("CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL)")) + conn.execute(text("INSERT INTO alembic_version VALUES ('026_add_scheduled_jobs')")) + + with patch("app.database.engine", test_engine), patch("app.database.DB_URL", f"sqlite:///{db_path}"): + init_db() + + inspector = sa_inspect(test_engine) + table_names = inspector.get_table_names() + assert "shared_links" in table_names + + # Verify the shared_links table has the expected columns. + columns = {col["name"] for col in inspector.get_columns("shared_links")} + assert "id" in columns + assert "token" in columns + assert "file_id" in columns + assert "owner_id" in columns + assert "expires_at" in columns + assert "is_active" in columns + + test_engine.dispose() + @pytest.mark.unit class TestGetDb: diff --git a/tests/test_views_coverage_boost.py b/tests/test_views_coverage_boost.py new file mode 100644 index 00000000..cd639b9b --- /dev/null +++ b/tests/test_views_coverage_boost.py @@ -0,0 +1,608 @@ +"""Tests to boost code coverage for all view modules below 100%. + +Covers: api_tokens, notifications, shared_links, share, plans, +imap_accounts, integrations, general, filemanager, files, help. +""" + +import os +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.config import settings as app_settings +from app.database import Base, get_db +from app.main import app + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def _fresh_db(): + """Yield a fresh in-memory SQLite session.""" + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(bind=engine) + session = sessionmaker(autocommit=False, autoflush=False, bind=engine)() + try: + yield session + finally: + session.close() + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def client_fresh(_fresh_db) -> TestClient: + """TestClient backed by a fresh database.""" + + def _override(): + try: + yield _fresh_db + finally: + pass + + app.dependency_overrides[get_db] = _override + with TestClient(app, base_url="http://localhost") as tc: + yield tc + app.dependency_overrides.clear() + + +# =================================================================== +# 1. Simple template-render views (api_tokens, notifications, +# shared_links, share, plans) +# =================================================================== + + +class TestApiTokensView: + """GET /api-tokens should render the management page.""" + + @pytest.mark.unit + def test_api_tokens_page_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/api-tokens") + assert resp.status_code == 200 + assert "API Tokens" in resp.text + + +class TestNotificationsView: + """GET /notifications should render the dashboard.""" + + @pytest.mark.unit + def test_notifications_page_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/notifications") + assert resp.status_code == 200 + assert "Notifications" in resp.text + + +class TestSharedLinksView: + """GET /shared-links should render the management page.""" + + @pytest.mark.unit + def test_shared_links_page_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/shared-links") + assert resp.status_code == 200 + assert "Shared Links" in resp.text + + +class TestShareView: + """GET /share/{token} should render the public share landing page.""" + + @pytest.mark.unit + def test_share_page_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/share/abc123") + assert resp.status_code == 200 + # Token should be passed to the template + assert "abc123" in resp.text + + +class TestPlansViews: + """GET /admin/plans and /admin/stripe-wizard should render pages.""" + + @pytest.mark.unit + def test_plan_designer_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/admin/plans") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_stripe_wizard_returns_200(self, client_fresh: TestClient): + resp = client_fresh.get("/admin/stripe-wizard") + assert resp.status_code == 200 + + +# =================================================================== +# 2. imap_accounts view (30.95 % → 100 %) +# =================================================================== + + +class TestImapAccountsView: + """Tests for /imap-accounts view.""" + + @pytest.mark.unit + def test_imap_accounts_page_no_owner(self, client_fresh: TestClient): + """When no owner_id is resolved, page renders with defaults.""" + resp = client_fresh.get("/imap-accounts") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_imap_accounts_page_with_owner(self, _fresh_db, client_fresh: TestClient): + """When session has a user, the page queries IMAP accounts.""" + with patch("app.views.imap_accounts.get_current_owner_id", return_value="testuser"): + with patch( + "app.views.imap_accounts.get_user_tier_id", + return_value="starter", + ): + with patch( + "app.views.imap_accounts.get_tier", + return_value={"id": "starter", "name": "Starter", "max_mailboxes": 3}, + ): + resp = client_fresh.get("/imap-accounts") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_get_max_mailboxes_free_tier(self): + """Free tier should return 0 mailboxes.""" + from app.views.imap_accounts import _get_max_mailboxes + + assert _get_max_mailboxes({"id": "free", "max_mailboxes": 0}) == 0 + + @pytest.mark.unit + def test_get_max_mailboxes_unlimited(self): + """When max_mailboxes is 0 on a non-free tier, it means unlimited.""" + from app.views.imap_accounts import _get_max_mailboxes + + assert _get_max_mailboxes({"id": "power", "max_mailboxes": 0}) is None + + @pytest.mark.unit + def test_get_max_mailboxes_limited(self): + """When max_mailboxes > 0, return that value.""" + from app.views.imap_accounts import _get_max_mailboxes + + assert _get_max_mailboxes({"id": "starter", "max_mailboxes": 5}) == 5 + + +# =================================================================== +# 3. integrations view (82.09 % → 100 %) +# =================================================================== + + +class TestIntegrationsView: + """Tests for /integrations view.""" + + @pytest.mark.unit + def test_integrations_dashboard_no_owner(self, client_fresh: TestClient): + """When no owner, the dashboard renders with zero-count defaults.""" + resp = client_fresh.get("/integrations") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_integrations_dashboard_with_owner(self, _fresh_db, client_fresh: TestClient): + """When an owner_id is resolved, DB queries run and tier is fetched.""" + with patch("app.views.integrations.get_current_owner_id", return_value="testuser"): + with patch("app.views.integrations.get_user_tier_id", return_value="power"): + with patch( + "app.views.integrations.get_tier", + return_value={ + "id": "power", + "name": "Power", + "max_storage_destinations": 0, + "max_mailboxes": 0, + }, + ): + resp = client_fresh.get("/integrations") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_integrations_dashboard_generic_exception(self, client_fresh: TestClient): + """A non-HTTP exception in the dashboard returns 500.""" + with patch( + "app.views.integrations.get_current_owner_id", + side_effect=RuntimeError("boom"), + ): + resp = client_fresh.get("/integrations") + assert resp.status_code == 500 + + @pytest.mark.unit + def test_integrations_dashboard_http_exception_passthrough(self, client_fresh: TestClient): + """An HTTPException inside the dashboard is re-raised, not wrapped in 500.""" + from fastapi import HTTPException + + with patch( + "app.views.integrations.get_current_owner_id", + side_effect=HTTPException(status_code=403, detail="Forbidden"), + ): + resp = client_fresh.get("/integrations") + assert resp.status_code == 403 + + @pytest.mark.unit + def test_get_max_destinations_free_default(self): + from app.views.integrations import _get_max_destinations + + assert _get_max_destinations({"id": "free", "max_storage_destinations": 0}) == 1 + + @pytest.mark.unit + def test_get_max_destinations_free_with_value(self): + from app.views.integrations import _get_max_destinations + + assert _get_max_destinations({"id": "free", "max_storage_destinations": 3}) == 3 + + @pytest.mark.unit + def test_get_max_destinations_unlimited(self): + from app.views.integrations import _get_max_destinations + + assert _get_max_destinations({"id": "power", "max_storage_destinations": 0}) is None + + @pytest.mark.unit + def test_get_max_destinations_limited(self): + from app.views.integrations import _get_max_destinations + + assert _get_max_destinations({"id": "starter", "max_storage_destinations": 5}) == 5 + + @pytest.mark.unit + def test_get_max_sources_free(self): + from app.views.integrations import _get_max_sources + + assert _get_max_sources({"id": "free", "max_mailboxes": 0}) == 0 + + @pytest.mark.unit + def test_get_max_sources_unlimited(self): + from app.views.integrations import _get_max_sources + + assert _get_max_sources({"id": "power", "max_mailboxes": 0}) is None + + @pytest.mark.unit + def test_get_max_sources_limited(self): + from app.views.integrations import _get_max_sources + + assert _get_max_sources({"id": "starter", "max_mailboxes": 2}) == 2 + + +# =================================================================== +# 4. general view (88.68 % → 100 %) +# =================================================================== + + +class TestGeneralViewMultiUser: + """Cover the multi_user_enabled subscription branch (lines 96-105).""" + + @staticmethod + def _signed_session(user_data: dict) -> str: + """Create a signed Starlette session cookie containing user_data.""" + import json + from base64 import b64encode + + from itsdangerous import TimestampSigner + + secret = os.environ.get( + "SESSION_SECRET", + "test_secret_key_for_testing_must_be_at_least_32_characters_long", + ) + signer = TimestampSigner(secret) + data = {"user": user_data} + return signer.sign(b64encode(json.dumps(data).encode("utf-8"))).decode("utf-8") + + @pytest.mark.unit + def test_home_page_multi_user_with_subscription(self, _fresh_db, client_fresh: TestClient): + """When multi_user_enabled is True and user has owner_id, subscription info is fetched. + + Lines 96-103: Exercises the subscription lookup path. + """ + cookie_val = self._signed_session({"username": "testuser", "email": "test@example.com", "is_admin": False}) + tier_mock = { + "id": "starter", + "name": "Starter", + "lifetime_file_limit": 1000, + "daily_upload_limit": 50, + "monthly_upload_limit": 500, + } + usage_mock = {"lifetime": 10, "today": 2, "month": 8} + with ( + patch.object(app_settings, "multi_user_enabled", True), + patch("app.utils.setup_wizard.is_setup_required", return_value=False), + patch("app.views.general.get_provider_status", return_value={}), + patch("app.views.general.validate_storage_configs", return_value={}), + patch("app.utils.subscription.get_user_tier_id", return_value="starter"), + patch("app.utils.subscription.get_tier", return_value=tier_mock), + patch("app.utils.subscription.get_user_usage", return_value=usage_mock), + ): + client_fresh.cookies.set("session", cookie_val) + resp = client_fresh.get("/?setup=complete") + assert resp.status_code == 200 + + @pytest.mark.unit + def test_home_page_multi_user_subscription_error(self, _fresh_db, client_fresh: TestClient): + """When subscription lookup fails, error is logged but page still renders. + + Lines 104-105: Exercises the exception handling branch. + """ + cookie_val = self._signed_session({"username": "testuser", "email": "test@example.com", "is_admin": False}) + with ( + patch.object(app_settings, "multi_user_enabled", True), + patch("app.utils.setup_wizard.is_setup_required", return_value=False), + patch("app.views.general.get_provider_status", return_value={}), + patch("app.views.general.validate_storage_configs", return_value={}), + patch("app.utils.subscription.get_user_tier_id", side_effect=RuntimeError("boom")), + ): + client_fresh.cookies.set("session", cookie_val) + resp = client_fresh.get("/?setup=complete") + assert resp.status_code == 200 + + +# =================================================================== +# 5. filemanager view (96.63 % → 100 %) +# =================================================================== + + +class TestFilemanagerCoverageGaps: + """Cover the remaining gaps in filemanager.py.""" + + @pytest.mark.unit + def test_format_size_petabytes(self): + """Line 43: _format_size should return PB for very large sizes.""" + from app.views.filemanager import _format_size + + # 1 PB = 1024^5 bytes + one_pb = 1024**5 + result = _format_size(one_pb) + assert "PB" in result + assert "1.0 PB" == result + + @pytest.mark.unit + def test_format_size_multiple_petabytes(self): + """Large values above 1 PB.""" + from app.views.filemanager import _format_size + + result = _format_size(5 * 1024**5) + assert "PB" in result + + @pytest.mark.unit + def test_scan_dir_with_broken_symlink(self, tmp_path): + """Lines 104-106: files that cannot be stat'd are skipped with a warning. + + Using a broken symlink to trigger OSError on stat(). + """ + from app.views.filemanager import _scan_dir + + # Create a broken symlink — stat() will raise FileNotFoundError (subclass of OSError) + broken_link = tmp_path / "broken_link.txt" + broken_link.symlink_to("/nonexistent/target/file") + + # Also create a valid file so we can verify it's included + valid_file = tmp_path / "valid.txt" + valid_file.write_text("hello") + + db_paths: set[str] = set() + entries = _scan_dir(tmp_path, tmp_path, db_paths) + + # The broken symlink should be skipped, the valid file should be included + entry_names = [e["name"] for e in entries] + assert "broken_link.txt" not in entry_names + assert "valid.txt" in entry_names + + @pytest.mark.unit + def test_scan_dir_oserror(self, tmp_path): + """OSError during stat in _scan_dir is caught and file is skipped. + + We create a second broken symlink for this test. + """ + from app.views.filemanager import _scan_dir + + broken_link = tmp_path / "also_broken.txt" + broken_link.symlink_to("/another/nonexistent/path") + + valid_file = tmp_path / "good.txt" + valid_file.write_text("ok") + db_paths: set[str] = set() + + entries = _scan_dir(tmp_path, tmp_path, db_paths) + entry_names = [e["name"] for e in entries] + assert "also_broken.txt" not in entry_names + assert "good.txt" in entry_names + + @pytest.mark.unit + def test_walk_all_files_with_broken_symlink(self, tmp_path): + """Lines 146-147: files that fail stat during walk are skipped. + + Using a broken symlink to trigger OSError. + """ + from app.views.filemanager import _walk_all_files + + broken_link = tmp_path / "broken.pdf" + broken_link.symlink_to("/nonexistent/target/file") + + valid_file = tmp_path / "valid.pdf" + valid_file.write_text("content") + + db_paths: set[str] = set() + entries = _walk_all_files(tmp_path, db_paths) + + entry_names = [e["name"] for e in entries] + assert "broken.pdf" not in entry_names + assert "valid.pdf" in entry_names + + +# =================================================================== +# 6. files view (99.21 % → 100 %) +# =================================================================== + + +class TestFilesViewCoverageGaps: + """Cover the remaining branches in files.py.""" + + @pytest.mark.unit + def test_compute_processing_flow_with_pipeline_steps(self): + """Lines 504-515: pipeline_steps filtering in _compute_processing_flow.""" + from app.views.files import _compute_processing_flow + + # Create mock pipeline steps + ps1 = SimpleNamespace(enabled=True, step_type="ocr") + ps2 = SimpleNamespace(enabled=False, step_type="extract_metadata") + ps3 = SimpleNamespace(enabled=True, step_type="send_to_destinations") + + # Create mock logs with all required attributes including task_id + log1 = SimpleNamespace( + step_name="create_file_record", + status="completed", + message="ok", + timestamp=None, + started_at=None, + completed_at=None, + task_id="task-001", + ) + log2 = SimpleNamespace( + step_name="check_text", + status="completed", + message="ok", + timestamp=None, + started_at=None, + completed_at=None, + task_id="task-002", + ) + + result = _compute_processing_flow([log1, log2], pipeline_steps=[ps1, ps2, ps3]) + + # _compute_processing_flow returns a list of stage dicts + stage_keys = [s["key"] for s in result] + assert "create_file_record" in stage_keys # always shown + assert "check_text" in stage_keys # OCR step type + ran + # extract_metadata is disabled, so its stages should NOT be included + assert "extract_metadata_with_gpt" not in stage_keys + + @pytest.mark.unit + def test_compute_processing_flow_with_pipeline_steps_none(self): + """When pipeline_steps is None, all stages are shown.""" + from app.views.files import _compute_processing_flow + + result = _compute_processing_flow([], pipeline_steps=None) + stage_keys = [s["key"] for s in result] + assert "create_file_record" in stage_keys + assert "extract_metadata_with_gpt" in stage_keys + + @pytest.mark.unit + def test_compute_processing_flow_with_empty_pipeline_steps(self): + """When pipeline_steps is empty list, only always-show + ran stages remain.""" + from app.views.files import _compute_processing_flow + + result = _compute_processing_flow([], pipeline_steps=[]) + stage_keys = [s["key"] for s in result] + assert "create_file_record" in stage_keys + # Other stages should be filtered out + assert "convert_to_pdf" not in stage_keys + + @pytest.mark.unit + def test_compute_processing_flow_dedup_enabled(self): + """When dedup is enabled and shown, check_for_duplicates stage appears.""" + from app.views.files import _compute_processing_flow + + with ( + patch.object(app_settings, "enable_deduplication", True), + patch.object(app_settings, "show_deduplication_step", True), + ): + result = _compute_processing_flow([], pipeline_steps=None) + stage_keys = [s["key"] for s in result] + assert "check_for_duplicates" in stage_keys + + @pytest.mark.unit + def test_file_detail_safe_exists_value_error(self, _fresh_db, client_fresh: TestClient): + """Test that _safe_exists handles ValueError from commonpath gracefully. + + Lines 240-241: When os.path.commonpath raises ValueError (e.g., paths + on different drives on Windows), _safe_exists returns False. + """ + from app.models import FileRecord + + # Create a file record with all required fields + rec = FileRecord( + original_filename="test.pdf", + local_filename="/tmp/test_local.pdf", + original_file_path="/tmp/test_original.pdf", + processed_file_path="/tmp/test_processed.pdf", + file_size=100, + mime_type="application/pdf", + filehash="abc123def456", + ) + _fresh_db.add(rec) + _fresh_db.commit() + _fresh_db.refresh(rec) + + # Patch commonpath to raise ValueError + with patch("os.path.commonpath", side_effect=ValueError("different drives")): + resp = client_fresh.get(f"/files/{rec.id}") + + assert resp.status_code == 200 + + +# =================================================================== +# 7. help view (96 % → 100 %) +# =================================================================== + + +class TestHelpViewCoverageGaps: + """Cover the missing branch in help.py (34->37).""" + + @pytest.mark.unit + def test_help_page_no_session_attr(self, client_fresh: TestClient): + """When no session user is set, defaults are used for Zammad widgets.""" + resp = client_fresh.get("/help") + assert resp.status_code == 200 + + @pytest.mark.unit + @pytest.mark.asyncio + async def test_help_page_request_without_session(self): + """Direct function call where request has no session attribute. + + Branch 34->37: when hasattr(request, 'session') is False. + """ + from app.views.help import help_center + + # Create a mock request without session attribute + mock_request = MagicMock(spec=[]) # spec=[] means no attributes + # help_center checks hasattr(request, "session") + # With spec=[], hasattr will return False + + with patch("app.views.help.templates") as mock_templates: + mock_templates.TemplateResponse.return_value = "ok" + await help_center(mock_request) + + # Template should be called with empty user context + call_args = mock_templates.TemplateResponse.call_args + ctx = call_args[0][1] if len(call_args[0]) > 1 else call_args[1].get("context", {}) + assert ctx["user_name"] == "" + assert ctx["user_email"] == "" + assert ctx["user_id"] == "" + + @pytest.mark.unit + def test_help_page_with_logged_in_user(self, client_fresh: TestClient): + """When session has user data, Zammad widget fields are populated. + + Covers lines 41-43 (user_name, user_email, user_id extraction). + """ + import json + from base64 import b64encode + + from itsdangerous import TimestampSigner + + secret = os.environ.get( + "SESSION_SECRET", + "test_secret_key_for_testing_must_be_at_least_32_characters_long", + ) + signer = TimestampSigner(secret) + session_data = { + "user": { + "name": "Jane Doe", + "email": "jane@example.com", + "preferred_username": "janedoe", + } + } + cookie_val = signer.sign(b64encode(json.dumps(session_data).encode("utf-8"))).decode("utf-8") + client_fresh.cookies.set("session", cookie_val) + + resp = client_fresh.get("/help") + assert resp.status_code == 200