fix: merge main into i18n branch and relink migration to 029
- 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>
This commit is contained in:
copilot-swe-agent[bot]
@@ -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,
|
||||
|
||||
@@ -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
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user