fix(migrations): merge main and rechain compliance migration as 031 after 030_add_mobile_devices
Merge main branch into compliance templates feature branch. Main had advanced with migrations 027-030 (ensure_shared_links, audit_logs, user_language_preference, mobile_devices) since this branch forked. Our compliance migration was 027 with down_revision 026, which conflicted with main's 027_ensure_shared_links_table. Changes: - Merge main (including i18n, audit logs, mobile, GraphQL features) - Resolve conflicts in app/api/__init__.py, app/models.py, tests/conftest.py - Rename 027_add_compliance_templates → 031_add_compliance_templates - Rechain: down_revision 026_add_scheduled_jobs → 030_add_mobile_devices - Add ComplianceTemplate to migrations/env.py imports - Alembic now has single head: 031_add_compliance_templates
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,
|
||||
ComplianceTemplate,
|
||||
DocumentMetadata,
|
||||
FileRecord,
|
||||
|
||||
@@ -0,0 +1,518 @@
|
||||
"""Tests for the mobile API endpoints (app/api/mobile.py)."""
|
||||
|
||||
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.database import Base, get_db
|
||||
from app.models import ApiToken, MobileDevice
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test data
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_OWNER = "mobile_user@example.com"
|
||||
_OTHER_OWNER = "other@example.com"
|
||||
_EXPO_TOKEN = "ExponentPushToken[test-token-abc123]"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mob_engine():
|
||||
"""In-memory SQLite engine for mobile tests."""
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def mob_session(mob_engine):
|
||||
"""DB session scoped to one test."""
|
||||
Session = sessionmaker(bind=mob_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
def _make_client(mob_engine, owner_id: str = _OWNER) -> TestClient:
|
||||
"""Return a TestClient with *owner_id* injected as the authenticated user."""
|
||||
from app.api.mobile import _get_owner_id
|
||||
from app.main import app
|
||||
|
||||
Session = sessionmaker(bind=mob_engine)
|
||||
|
||||
def _override_get_db():
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def _override_owner():
|
||||
return owner_id
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
app.dependency_overrides[_get_owner_id] = _override_owner
|
||||
|
||||
client = TestClient(app, base_url="http://localhost", raise_server_exceptions=False)
|
||||
return client
|
||||
|
||||
|
||||
def _cleanup(app):
|
||||
"""Remove dependency overrides after test."""
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – /mobile/generate-token
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGenerateMobileToken:
|
||||
"""Tests for POST /api/mobile/generate-token."""
|
||||
|
||||
def test_generate_token_success(self, mob_engine):
|
||||
"""Generating a mobile token returns a token string and metadata."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post(
|
||||
"/api/mobile/generate-token",
|
||||
json={"device_name": "John's iPhone"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["token"].startswith("de_")
|
||||
assert data["token_id"] > 0
|
||||
assert "Mobile App" in data["name"]
|
||||
assert "John's iPhone" in data["name"]
|
||||
assert "created_at" in data
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_generate_token_default_device_name(self, mob_engine):
|
||||
"""A default device name is used if none is provided."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post("/api/mobile/generate-token", json={})
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert "Mobile App" in data["name"]
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_generate_token_persisted_in_db(self, mob_engine, mob_session):
|
||||
"""The generated token is stored in the api_tokens table."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post(
|
||||
"/api/mobile/generate-token",
|
||||
json={"device_name": "Test Device"},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
token_id = resp.json()["token_id"]
|
||||
|
||||
db_token = mob_session.get(ApiToken, token_id)
|
||||
assert db_token is not None
|
||||
assert db_token.owner_id == _OWNER
|
||||
assert "Mobile App" in db_token.name
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_generate_token_unauthenticated(self, mob_engine):
|
||||
"""Unauthenticated requests are rejected with 401."""
|
||||
from app.api.mobile import _get_owner_id
|
||||
from app.main import app
|
||||
|
||||
Session = sessionmaker(bind=mob_engine)
|
||||
|
||||
def _override_get_db():
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
def _raise_401():
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
app.dependency_overrides[_get_owner_id] = _raise_401
|
||||
|
||||
client = TestClient(app, base_url="http://localhost", raise_server_exceptions=False)
|
||||
try:
|
||||
resp = client.post("/api/mobile/generate-token", json={"device_name": "Test"})
|
||||
assert resp.status_code == 401
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – /mobile/register-device
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRegisterDevice:
|
||||
"""Tests for POST /api/mobile/register-device."""
|
||||
|
||||
def test_register_new_device(self, mob_engine, mob_session):
|
||||
"""Registering a new device persists it in mobile_devices."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post(
|
||||
"/api/mobile/register-device",
|
||||
json={
|
||||
"push_token": _EXPO_TOKEN,
|
||||
"device_name": "Test iPhone",
|
||||
"platform": "ios",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["id"] > 0
|
||||
assert data["platform"] == "ios"
|
||||
assert data["is_active"] is True
|
||||
assert "ExponentPushToken" in data["push_token_preview"]
|
||||
|
||||
device = mob_session.get(MobileDevice, data["id"])
|
||||
assert device is not None
|
||||
assert device.push_token == _EXPO_TOKEN
|
||||
assert device.owner_id == _OWNER
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_register_same_token_is_idempotent(self, mob_engine, mob_session):
|
||||
"""Re-registering the same token reactivates the existing record."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp1 = client.post(
|
||||
"/api/mobile/register-device",
|
||||
json={"push_token": _EXPO_TOKEN, "platform": "ios"},
|
||||
)
|
||||
assert resp1.status_code == 201
|
||||
id1 = resp1.json()["id"]
|
||||
|
||||
resp2 = client.post(
|
||||
"/api/mobile/register-device",
|
||||
json={"push_token": _EXPO_TOKEN, "device_name": "Updated Name", "platform": "ios"},
|
||||
)
|
||||
assert resp2.status_code == 201
|
||||
id2 = resp2.json()["id"]
|
||||
|
||||
assert id1 == id2 # Same record reused
|
||||
|
||||
devices = mob_session.query(MobileDevice).filter(MobileDevice.owner_id == _OWNER).all()
|
||||
assert len(devices) == 1
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_register_invalid_platform(self, mob_engine):
|
||||
"""An invalid platform value is rejected with 422."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post(
|
||||
"/api/mobile/register-device",
|
||||
json={"push_token": _EXPO_TOKEN, "platform": "windows"},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_register_android_device(self, mob_engine):
|
||||
"""Android devices can be registered."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.post(
|
||||
"/api/mobile/register-device",
|
||||
json={
|
||||
"push_token": "ExponentPushToken[android-token-xyz]",
|
||||
"device_name": "Pixel 8",
|
||||
"platform": "android",
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
assert resp.json()["platform"] == "android"
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – /mobile/devices
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestListDevices:
|
||||
"""Tests for GET /api/mobile/devices."""
|
||||
|
||||
def test_list_devices_empty(self, mob_engine):
|
||||
"""An empty list is returned when no devices are registered."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.get("/api/mobile/devices")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_list_devices_returns_own_devices_only(self, mob_engine, mob_session):
|
||||
"""Only the current user's devices are returned."""
|
||||
from app.main import app
|
||||
|
||||
# Add devices for two different owners directly
|
||||
mob_session.add(
|
||||
MobileDevice(
|
||||
owner_id=_OWNER,
|
||||
push_token="ExponentPushToken[owner-token-12345]",
|
||||
platform="ios",
|
||||
)
|
||||
)
|
||||
mob_session.add(
|
||||
MobileDevice(
|
||||
owner_id=_OTHER_OWNER,
|
||||
push_token="ExponentPushToken[other-token-67890]",
|
||||
platform="android",
|
||||
)
|
||||
)
|
||||
mob_session.commit()
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.get("/api/mobile/devices")
|
||||
assert resp.status_code == 200
|
||||
devices = resp.json()
|
||||
assert len(devices) == 1
|
||||
# The push_token_preview is the first 20 chars + "…"
|
||||
assert devices[0]["push_token_preview"].startswith("ExponentPushToken[ow")
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – DELETE /mobile/devices/{device_id}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDeactivateDevice:
|
||||
"""Tests for DELETE /api/mobile/devices/{device_id}."""
|
||||
|
||||
def test_deactivate_own_device(self, mob_engine, mob_session):
|
||||
"""Deactivating a device sets is_active to False."""
|
||||
from app.main import app
|
||||
|
||||
device = MobileDevice(
|
||||
owner_id=_OWNER,
|
||||
push_token=_EXPO_TOKEN,
|
||||
platform="ios",
|
||||
is_active=True,
|
||||
)
|
||||
mob_session.add(device)
|
||||
mob_session.commit()
|
||||
mob_session.refresh(device)
|
||||
device_id = device.id
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.delete(f"/api/mobile/devices/{device_id}")
|
||||
assert resp.status_code == 204
|
||||
|
||||
mob_session.expire_all()
|
||||
updated = mob_session.get(MobileDevice, device_id)
|
||||
assert updated is not None
|
||||
assert updated.is_active is False
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_deactivate_other_users_device_returns_404(self, mob_engine, mob_session):
|
||||
"""Attempting to deactivate another user's device returns 404."""
|
||||
from app.main import app
|
||||
|
||||
device = MobileDevice(
|
||||
owner_id=_OTHER_OWNER,
|
||||
push_token="ExponentPushToken[other-token]",
|
||||
platform="ios",
|
||||
is_active=True,
|
||||
)
|
||||
mob_session.add(device)
|
||||
mob_session.commit()
|
||||
mob_session.refresh(device)
|
||||
device_id = device.id
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.delete(f"/api/mobile/devices/{device_id}")
|
||||
assert resp.status_code == 404
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_deactivate_nonexistent_device_returns_404(self, mob_engine):
|
||||
"""Deactivating a device that does not exist returns 404."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.delete("/api/mobile/devices/99999")
|
||||
assert resp.status_code == 404
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – /mobile/whoami
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestWhoAmI:
|
||||
"""Tests for GET /api/mobile/whoami."""
|
||||
|
||||
def test_whoami_with_no_profile(self, mob_engine):
|
||||
"""Returns owner_id and inferred email even when no UserProfile exists."""
|
||||
from app.main import app
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.get("/api/mobile/whoami")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["owner_id"] == _OWNER
|
||||
assert data["display_name"] is None
|
||||
# _OWNER contains "@" so email is inferred from owner_id
|
||||
assert data["email"] == _OWNER
|
||||
assert data["avatar_url"] is not None # Gravatar URL from email
|
||||
assert data["is_admin"] is False
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
def test_whoami_with_profile(self, mob_engine, mob_session):
|
||||
"""Returns full profile data when a UserProfile record exists."""
|
||||
from app.main import app
|
||||
from app.models import UserProfile
|
||||
|
||||
profile = UserProfile(
|
||||
user_id=_OWNER,
|
||||
display_name="Alice Test",
|
||||
)
|
||||
mob_session.add(profile)
|
||||
mob_session.commit()
|
||||
|
||||
client = _make_client(mob_engine)
|
||||
try:
|
||||
resp = client.get("/api/mobile/whoami")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["owner_id"] == _OWNER
|
||||
assert data["display_name"] == "Alice Test"
|
||||
# owner_id contains "@" so email is inferred from it
|
||||
assert data["email"] == _OWNER
|
||||
assert data["avatar_url"] is not None # Gravatar URL
|
||||
assert data["is_admin"] is False
|
||||
finally:
|
||||
_cleanup(app)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests – push notification utility
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestPushNotificationUtility:
|
||||
"""Tests for app/utils/push_notification.py."""
|
||||
|
||||
def test_send_expo_push_empty_tokens(self):
|
||||
"""send_expo_push_notification with no tokens returns empty list."""
|
||||
from app.utils.push_notification import send_expo_push_notification
|
||||
|
||||
result = send_expo_push_notification([], "Title", "Body")
|
||||
assert result == []
|
||||
|
||||
def test_send_expo_push_calls_expo_api(self):
|
||||
"""send_expo_push_notification POSTs to the Expo push API."""
|
||||
from app.utils.push_notification import send_expo_push_notification
|
||||
|
||||
mock_response = MagicMock()
|
||||
mock_response.json.return_value = {"data": [{"status": "ok"}]}
|
||||
mock_response.raise_for_status = MagicMock()
|
||||
|
||||
with patch("app.utils.push_notification.httpx.post", return_value=mock_response) as mock_post:
|
||||
result = send_expo_push_notification(
|
||||
tokens=["ExponentPushToken[abc]"],
|
||||
title="Test",
|
||||
body="Message",
|
||||
)
|
||||
|
||||
mock_post.assert_called_once()
|
||||
call_kwargs = mock_post.call_args
|
||||
assert "exp.host" in call_kwargs[0][0]
|
||||
payload = call_kwargs[1]["json"]
|
||||
assert len(payload) == 1
|
||||
assert payload[0]["to"] == "ExponentPushToken[abc]"
|
||||
assert payload[0]["title"] == "Test"
|
||||
|
||||
def test_send_push_to_owner_no_devices(self, mob_engine):
|
||||
"""send_push_to_owner silently does nothing when no devices are registered."""
|
||||
from app.utils.push_notification import send_push_to_owner
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.return_value.filter.return_value.all.return_value = []
|
||||
mock_session.close = MagicMock()
|
||||
|
||||
with patch("app.utils.push_notification.SessionLocal", return_value=mock_session):
|
||||
with patch("app.utils.push_notification.send_expo_push_notification") as mock_send:
|
||||
send_push_to_owner("user@example.com", "Title", "Body")
|
||||
mock_send.assert_not_called()
|
||||
mock_session.close.assert_called_once()
|
||||
|
||||
def test_send_push_to_owner_with_devices(self):
|
||||
"""send_push_to_owner calls send_expo_push_notification with device tokens."""
|
||||
from app.utils.push_notification import send_push_to_owner
|
||||
|
||||
fake_device = MagicMock()
|
||||
fake_device.push_token = "ExponentPushToken[device1]"
|
||||
fake_device.is_active = True
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.query.return_value.filter.return_value.all.return_value = [fake_device]
|
||||
mock_session.close = MagicMock()
|
||||
|
||||
with patch("app.utils.push_notification.SessionLocal", return_value=mock_session):
|
||||
with patch("app.utils.push_notification.send_expo_push_notification") as mock_send:
|
||||
mock_send.return_value = [{"status": "ok"}]
|
||||
send_push_to_owner("user@example.com", "Processed!", "Your doc is ready.")
|
||||
|
||||
mock_send.assert_called_once()
|
||||
call_kwargs = mock_send.call_args[1]
|
||||
assert "ExponentPushToken[device1]" in call_kwargs["tokens"]
|
||||
@@ -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,419 @@
|
||||
"""
|
||||
Tests for the GraphQL API endpoint at /graphql.
|
||||
|
||||
Covers:
|
||||
- Schema introspection (endpoint availability + GraphiQL)
|
||||
- Query: documents (list, single, auth-gated)
|
||||
- Query: pipelines (list, single)
|
||||
- Query: settings (admin-only)
|
||||
- Query: users (admin-only)
|
||||
- Pagination and limit clamping
|
||||
- Sensitive setting keys are excluded
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.models import ApplicationSettings, FileRecord, Pipeline, PipelineStep, UserProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def gql(client: TestClient, query: str, variables: dict | None = None) -> dict:
|
||||
"""Execute a GraphQL POST request and return the parsed JSON body."""
|
||||
payload: dict = {"query": query}
|
||||
if variables:
|
||||
payload["variables"] = variables
|
||||
response = client.post("/graphql", json=payload)
|
||||
assert response.status_code == 200, f"Unexpected status {response.status_code}: {response.text}"
|
||||
return response.json()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def file_record(db_session) -> FileRecord:
|
||||
rec = FileRecord(
|
||||
owner_id="user1",
|
||||
original_filename="invoice.pdf",
|
||||
local_filename="/workdir/tmp/invoice.pdf",
|
||||
file_size=1024,
|
||||
mime_type="application/pdf",
|
||||
filehash="abc123",
|
||||
)
|
||||
db_session.add(rec)
|
||||
db_session.commit()
|
||||
db_session.refresh(rec)
|
||||
return rec
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def pipeline_record(db_session) -> Pipeline:
|
||||
p = Pipeline(
|
||||
owner_id="user1",
|
||||
name="Test Pipeline",
|
||||
description="A pipeline for tests",
|
||||
is_default=False,
|
||||
is_active=True,
|
||||
)
|
||||
db_session.add(p)
|
||||
db_session.commit()
|
||||
db_session.refresh(p)
|
||||
|
||||
step = PipelineStep(
|
||||
pipeline_id=p.id,
|
||||
position=0,
|
||||
step_type="ocr",
|
||||
label="Run OCR",
|
||||
enabled=True,
|
||||
)
|
||||
db_session.add(step)
|
||||
db_session.commit()
|
||||
return p
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setting_record(db_session) -> ApplicationSettings:
|
||||
s = ApplicationSettings(key="max_upload_size", value="104857600")
|
||||
db_session.add(s)
|
||||
db_session.commit()
|
||||
db_session.refresh(s)
|
||||
return s
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def user_profile(db_session) -> UserProfile:
|
||||
profile = UserProfile(
|
||||
user_id="user1",
|
||||
display_name="Test User",
|
||||
is_blocked=False,
|
||||
subscription_tier="free",
|
||||
onboarding_completed=False,
|
||||
)
|
||||
db_session.add(profile)
|
||||
db_session.commit()
|
||||
db_session.refresh(profile)
|
||||
return profile
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: endpoint availability
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGraphQLEndpoint:
|
||||
"""Verify the /graphql endpoint is reachable and introspectable."""
|
||||
|
||||
def test_graphql_post_exists(self, client: TestClient):
|
||||
"""POST /graphql returns 200 for a valid introspection query."""
|
||||
result = gql(client, "{ __schema { queryType { name } } }")
|
||||
assert "data" in result
|
||||
assert result["data"]["__schema"]["queryType"]["name"] == "Query"
|
||||
|
||||
def test_graphql_get_returns_graphiql(self, client: TestClient):
|
||||
"""GET /graphql returns the GraphiQL playground HTML."""
|
||||
response = client.get("/graphql", headers={"Accept": "text/html"})
|
||||
assert response.status_code == 200
|
||||
assert "graphiql" in response.text.lower() or "graphql" in response.text.lower()
|
||||
|
||||
def test_graphql_schema_has_expected_types(self, client: TestClient):
|
||||
"""Schema exposes DocumentType, PipelineType, SettingType, UserType."""
|
||||
result = gql(
|
||||
client,
|
||||
"""
|
||||
{
|
||||
__schema {
|
||||
types { name }
|
||||
}
|
||||
}
|
||||
""",
|
||||
)
|
||||
type_names = {t["name"] for t in result["data"]["__schema"]["types"]}
|
||||
for expected in ("DocumentType", "PipelineType", "SettingType", "UserType"):
|
||||
assert expected in type_names, f"{expected} not found in schema"
|
||||
|
||||
def test_graphql_query_fields(self, client: TestClient):
|
||||
"""Root Query has documents, document, pipelines, pipeline, settings, users, user fields."""
|
||||
result = gql(
|
||||
client,
|
||||
"""
|
||||
{
|
||||
__type(name: "Query") {
|
||||
fields { name }
|
||||
}
|
||||
}
|
||||
""",
|
||||
)
|
||||
field_names = {f["name"] for f in result["data"]["__type"]["fields"]}
|
||||
for expected in ("documents", "document", "pipelines", "pipeline", "settings", "users", "user"):
|
||||
assert expected in field_names, f"Query field '{expected}' missing from schema"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: documents queries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestDocumentsQuery:
|
||||
"""Tests for the documents and document queries."""
|
||||
|
||||
def test_list_documents_empty(self, client: TestClient):
|
||||
result = gql(client, "{ documents { id originalFilename } }")
|
||||
assert "errors" not in result
|
||||
assert result["data"]["documents"] == []
|
||||
|
||||
def test_list_documents_returns_records(self, client: TestClient, file_record: FileRecord):
|
||||
result = gql(client, "{ documents { id originalFilename mimeType fileSize } }")
|
||||
assert "errors" not in result
|
||||
docs = result["data"]["documents"]
|
||||
assert len(docs) == 1
|
||||
assert docs[0]["id"] == file_record.id
|
||||
assert docs[0]["originalFilename"] == "invoice.pdf"
|
||||
assert docs[0]["mimeType"] == "application/pdf"
|
||||
assert docs[0]["fileSize"] == 1024
|
||||
|
||||
def test_get_single_document(self, client: TestClient, file_record: FileRecord):
|
||||
result = gql(
|
||||
client,
|
||||
"query($id: Int!) { document(id: $id) { id originalFilename } }",
|
||||
variables={"id": file_record.id},
|
||||
)
|
||||
assert "errors" not in result
|
||||
assert result["data"]["document"]["id"] == file_record.id
|
||||
|
||||
def test_get_nonexistent_document_returns_null(self, client: TestClient):
|
||||
result = gql(client, "{ document(id: 99999) { id } }")
|
||||
assert "errors" not in result
|
||||
assert result["data"]["document"] is None
|
||||
|
||||
def test_documents_pagination(self, client: TestClient, db_session):
|
||||
for i in range(5):
|
||||
db_session.add(
|
||||
FileRecord(
|
||||
owner_id="user1",
|
||||
original_filename=f"doc{i}.pdf",
|
||||
local_filename=f"/workdir/tmp/doc{i}.pdf",
|
||||
file_size=100,
|
||||
filehash=f"hash{i}",
|
||||
)
|
||||
)
|
||||
db_session.commit()
|
||||
|
||||
result_page1 = gql(client, "{ documents(limit: 2, offset: 0) { id } }")
|
||||
result_page2 = gql(client, "{ documents(limit: 2, offset: 2) { id } }")
|
||||
assert "errors" not in result_page1
|
||||
assert "errors" not in result_page2
|
||||
assert len(result_page1["data"]["documents"]) == 2
|
||||
assert len(result_page2["data"]["documents"]) == 2
|
||||
|
||||
def test_documents_limit_clamped_to_100(self, client: TestClient, db_session):
|
||||
# Requesting more than 100 should be silently clamped to 100
|
||||
for i in range(5):
|
||||
db_session.add(
|
||||
FileRecord(
|
||||
owner_id="user1",
|
||||
original_filename=f"big{i}.pdf",
|
||||
local_filename=f"/workdir/tmp/big{i}.pdf",
|
||||
file_size=100,
|
||||
filehash=f"bighash{i}",
|
||||
)
|
||||
)
|
||||
db_session.commit()
|
||||
result = gql(client, "{ documents(limit: 999) { id } }")
|
||||
assert "errors" not in result
|
||||
# Just verify it doesn't error and returns something
|
||||
assert isinstance(result["data"]["documents"], list)
|
||||
|
||||
def test_documents_filter_by_owner(self, client: TestClient, db_session):
|
||||
db_session.add(
|
||||
FileRecord(
|
||||
owner_id="alice",
|
||||
original_filename="alice.pdf",
|
||||
local_filename="/workdir/tmp/alice.pdf",
|
||||
file_size=100,
|
||||
filehash="alicehash",
|
||||
)
|
||||
)
|
||||
db_session.add(
|
||||
FileRecord(
|
||||
owner_id="bob",
|
||||
original_filename="bob.pdf",
|
||||
local_filename="/workdir/tmp/bob.pdf",
|
||||
file_size=200,
|
||||
filehash="bobhash",
|
||||
)
|
||||
)
|
||||
db_session.commit()
|
||||
|
||||
result = gql(client, '{ documents(ownerId: "alice") { id originalFilename } }')
|
||||
assert "errors" not in result
|
||||
docs = result["data"]["documents"]
|
||||
assert all(d["originalFilename"] == "alice.pdf" for d in docs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: pipelines queries
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestPipelinesQuery:
|
||||
"""Tests for the pipelines and pipeline queries."""
|
||||
|
||||
def test_list_pipelines_empty(self, client: TestClient):
|
||||
result = gql(client, "{ pipelines { id name } }")
|
||||
assert "errors" not in result
|
||||
assert result["data"]["pipelines"] == []
|
||||
|
||||
def test_list_pipelines_with_steps(self, client: TestClient, pipeline_record: Pipeline):
|
||||
result = gql(
|
||||
client,
|
||||
"""
|
||||
{
|
||||
pipelines {
|
||||
id name description isDefault isActive
|
||||
steps { id stepType position enabled }
|
||||
}
|
||||
}
|
||||
""",
|
||||
)
|
||||
assert "errors" not in result
|
||||
pipelines = result["data"]["pipelines"]
|
||||
assert len(pipelines) == 1
|
||||
assert pipelines[0]["name"] == "Test Pipeline"
|
||||
assert len(pipelines[0]["steps"]) == 1
|
||||
assert pipelines[0]["steps"][0]["stepType"] == "ocr"
|
||||
|
||||
def test_get_single_pipeline(self, client: TestClient, pipeline_record: Pipeline):
|
||||
result = gql(
|
||||
client,
|
||||
"query($id: Int!) { pipeline(id: $id) { id name steps { stepType } } }",
|
||||
variables={"id": pipeline_record.id},
|
||||
)
|
||||
assert "errors" not in result
|
||||
assert result["data"]["pipeline"]["id"] == pipeline_record.id
|
||||
assert result["data"]["pipeline"]["steps"][0]["stepType"] == "ocr"
|
||||
|
||||
def test_get_nonexistent_pipeline_returns_null(self, client: TestClient):
|
||||
result = gql(client, "{ pipeline(id: 99999) { id } }")
|
||||
assert "errors" not in result
|
||||
assert result["data"]["pipeline"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: settings query (admin-only)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSettingsQuery:
|
||||
"""Tests for the settings query."""
|
||||
|
||||
def test_settings_returns_data_when_no_auth(self, client: TestClient, setting_record: ApplicationSettings):
|
||||
"""When AUTH_ENABLED=False, settings are accessible (no auth required)."""
|
||||
result = gql(client, "{ settings { key value } }")
|
||||
assert "errors" not in result
|
||||
keys = [s["key"] for s in result["data"]["settings"]]
|
||||
assert "max_upload_size" in keys
|
||||
|
||||
def test_sensitive_settings_excluded(self, client: TestClient, db_session):
|
||||
"""Sensitive setting keys must never appear in the response."""
|
||||
sensitive_keys = [
|
||||
"openai_api_key",
|
||||
"session_secret",
|
||||
"azure_ai_key",
|
||||
"smtp_password",
|
||||
]
|
||||
for key in sensitive_keys:
|
||||
db_session.add(ApplicationSettings(key=key, value="super-secret"))
|
||||
db_session.commit()
|
||||
|
||||
result = gql(client, "{ settings { key value } }")
|
||||
assert "errors" not in result
|
||||
returned_keys = {s["key"] for s in result["data"]["settings"]}
|
||||
for key in sensitive_keys:
|
||||
assert key not in returned_keys, f"Sensitive key '{key}' was returned by GraphQL settings query"
|
||||
|
||||
def test_settings_auth_required_when_auth_enabled(self, client: TestClient):
|
||||
"""When AUTH_ENABLED=True and no user, settings query must return an error."""
|
||||
from app.config import settings as app_settings
|
||||
|
||||
with patch.object(app_settings, "auth_enabled", True):
|
||||
result = gql(client, "{ settings { key } }")
|
||||
# Should have errors because no user is authenticated
|
||||
assert "errors" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: users query (admin-only)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUsersQuery:
|
||||
"""Tests for the users and user queries."""
|
||||
|
||||
def test_users_returns_profiles_when_no_auth(self, client: TestClient, user_profile: UserProfile):
|
||||
"""When AUTH_ENABLED=False, users are accessible."""
|
||||
result = gql(client, "{ users { userId displayName subscriptionTier } }")
|
||||
assert "errors" not in result
|
||||
users = result["data"]["users"]
|
||||
assert any(u["userId"] == "user1" for u in users)
|
||||
|
||||
def test_get_user_by_id(self, client: TestClient, user_profile: UserProfile):
|
||||
result = gql(
|
||||
client,
|
||||
'query { user(userId: "user1") { userId displayName isBlocked } }',
|
||||
)
|
||||
assert "errors" not in result
|
||||
assert result["data"]["user"]["userId"] == "user1"
|
||||
assert result["data"]["user"]["displayName"] == "Test User"
|
||||
assert result["data"]["user"]["isBlocked"] is False
|
||||
|
||||
def test_get_nonexistent_user_returns_null(self, client: TestClient):
|
||||
result = gql(client, '{ user(userId: "nobody") { userId } }')
|
||||
assert "errors" not in result
|
||||
assert result["data"]["user"] is None
|
||||
|
||||
def test_users_auth_required_when_auth_enabled(self, client: TestClient):
|
||||
"""When AUTH_ENABLED=True and no user, users query must return an error."""
|
||||
from app.config import settings as app_settings
|
||||
|
||||
with patch.object(app_settings, "auth_enabled", True):
|
||||
result = gql(client, "{ users { userId } }")
|
||||
assert "errors" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: auth enforcement
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGraphQLAuth:
|
||||
"""Verify auth is enforced for all queries when AUTH_ENABLED=True."""
|
||||
|
||||
def test_documents_auth_required_when_auth_enabled(self, client: TestClient):
|
||||
from app.config import settings as app_settings
|
||||
|
||||
with patch.object(app_settings, "auth_enabled", True):
|
||||
result = gql(client, "{ documents { id } }")
|
||||
assert "errors" in result
|
||||
|
||||
def test_pipelines_auth_required_when_auth_enabled(self, client: TestClient):
|
||||
from app.config import settings as app_settings
|
||||
|
||||
with patch.object(app_settings, "auth_enabled", True):
|
||||
result = gql(client, "{ pipelines { id } }")
|
||||
assert "errors" in result
|
||||
@@ -0,0 +1,447 @@
|
||||
"""Tests for the i18n (internationalization) and l10n (localization) utilities."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.utils.i18n import (
|
||||
DEFAULT_LANGUAGE,
|
||||
SUPPORTED_LANGUAGE_CODES,
|
||||
SUPPORTED_LANGUAGES,
|
||||
_parse_accept_language,
|
||||
detect_language,
|
||||
format_date,
|
||||
format_datetime,
|
||||
format_number,
|
||||
get_language_info,
|
||||
reload_translations,
|
||||
translate,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Translation file integrity
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTranslationFiles:
|
||||
"""Verify that all translation JSON files are valid and complete."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_cache(self) -> None:
|
||||
"""Clear translation cache before each test."""
|
||||
reload_translations()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_all_translation_files_exist(self) -> None:
|
||||
"""Every supported language must have a corresponding JSON file."""
|
||||
translations_dir = Path(__file__).resolve().parent.parent / "frontend" / "translations"
|
||||
for lang in SUPPORTED_LANGUAGES:
|
||||
filepath = translations_dir / f"{lang['code']}.json"
|
||||
assert filepath.is_file(), f"Missing translation file for {lang['code']}"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_all_translation_files_are_valid_json(self) -> None:
|
||||
"""All translation files must be parseable JSON."""
|
||||
translations_dir = Path(__file__).resolve().parent.parent / "frontend" / "translations"
|
||||
for lang in SUPPORTED_LANGUAGES:
|
||||
filepath = translations_dir / f"{lang['code']}.json"
|
||||
data = json.loads(filepath.read_text(encoding="utf-8"))
|
||||
assert isinstance(data, dict), f"{lang['code']}.json must be a dict"
|
||||
assert len(data) > 0, f"{lang['code']}.json must not be empty"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_all_languages_have_same_keys(self) -> None:
|
||||
"""All translation files should have the same set of keys as English."""
|
||||
translations_dir = Path(__file__).resolve().parent.parent / "frontend" / "translations"
|
||||
en_path = translations_dir / "en.json"
|
||||
en_keys = set(json.loads(en_path.read_text(encoding="utf-8")).keys())
|
||||
|
||||
for lang in SUPPORTED_LANGUAGES:
|
||||
if lang["code"] == "en":
|
||||
continue
|
||||
filepath = translations_dir / f"{lang['code']}.json"
|
||||
lang_keys = set(json.loads(filepath.read_text(encoding="utf-8")).keys())
|
||||
missing = en_keys - lang_keys
|
||||
assert not missing, f"{lang['code']}.json missing keys: {missing}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Core translate() function
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestTranslate:
|
||||
"""Tests for the translate() function."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_cache(self) -> None:
|
||||
reload_translations()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_english_key(self) -> None:
|
||||
"""English keys should resolve to English text."""
|
||||
result = translate("nav.dashboard", "en")
|
||||
assert result == "Dashboard"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_german_key(self) -> None:
|
||||
"""German locale should return German text."""
|
||||
result = translate("nav.dashboard", "de")
|
||||
assert result == "Übersicht"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_french_key(self) -> None:
|
||||
"""French locale should return French text."""
|
||||
result = translate("nav.dashboard", "fr")
|
||||
assert result == "Tableau de bord"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_chinese_key(self) -> None:
|
||||
"""Chinese locale should return Chinese text."""
|
||||
result = translate("nav.dashboard", "zh")
|
||||
assert result == "仪表盘"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_fallback_to_english(self) -> None:
|
||||
"""Unknown locale falls back to English."""
|
||||
result = translate("nav.dashboard", "xx")
|
||||
assert result == "Dashboard"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_missing_key_returns_key(self) -> None:
|
||||
"""Missing key falls back to the key itself."""
|
||||
result = translate("nonexistent.key", "en")
|
||||
assert result == "nonexistent.key"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_none_locale_uses_default(self) -> None:
|
||||
"""None locale defaults to English."""
|
||||
result = translate("nav.dashboard", None)
|
||||
assert result == "Dashboard"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_with_kwargs(self) -> None:
|
||||
"""Placeholders should be interpolated via kwargs."""
|
||||
result = translate("footer.copyright", "en", year="2025")
|
||||
assert result == "DocuElevate 2025"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_translate_with_kwargs_german(self) -> None:
|
||||
"""Placeholder interpolation in German."""
|
||||
result = translate("language.changed", "de", language="English")
|
||||
assert result == "Sprache geändert zu English"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Accept-Language header parsing
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestParseAcceptLanguage:
|
||||
"""Tests for parsing the Accept-Language HTTP header."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_simple_language(self) -> None:
|
||||
assert _parse_accept_language("de") == "de"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_language_with_region(self) -> None:
|
||||
assert _parse_accept_language("de-DE") == "de"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_multiple_languages_quality(self) -> None:
|
||||
result = _parse_accept_language("fr;q=0.9, de;q=1.0, en;q=0.8")
|
||||
assert result == "de"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_unsupported_language_fallback(self) -> None:
|
||||
result = _parse_accept_language("ja, ko")
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_empty_header(self) -> None:
|
||||
assert _parse_accept_language("") is None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_complex_accept_language(self) -> None:
|
||||
header = "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7"
|
||||
result = _parse_accept_language(header)
|
||||
assert result == "zh"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Language detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDetectLanguage:
|
||||
"""Tests for detecting language from request context."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_session_preference_takes_priority(self) -> None:
|
||||
request = MagicMock()
|
||||
request.session = {"preferred_language": "de"}
|
||||
request.cookies = {}
|
||||
request.headers = {}
|
||||
assert detect_language(request) == "de"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cookie_fallback(self) -> None:
|
||||
request = MagicMock()
|
||||
request.session = {}
|
||||
request.cookies = {"docuelevate_lang": "fr"}
|
||||
request.headers = {}
|
||||
assert detect_language(request) == "fr"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_accept_language_fallback(self) -> None:
|
||||
request = MagicMock()
|
||||
request.session = {}
|
||||
request.cookies = {}
|
||||
request.headers = {"accept-language": "es-ES,es;q=0.9"}
|
||||
assert detect_language(request) == "es"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_default_fallback(self) -> None:
|
||||
request = MagicMock()
|
||||
request.session = {}
|
||||
request.cookies = {}
|
||||
request.headers = {}
|
||||
assert detect_language(request) == DEFAULT_LANGUAGE
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_invalid_session_language_ignored(self) -> None:
|
||||
request = MagicMock()
|
||||
request.session = {"preferred_language": "invalid"}
|
||||
request.cookies = {"docuelevate_lang": "it"}
|
||||
request.headers = {}
|
||||
assert detect_language(request) == "it"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Localization helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestL10nFormatters:
|
||||
"""Tests for locale-aware formatting functions."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_date_english(self) -> None:
|
||||
d = date(2025, 3, 15)
|
||||
result = format_date(d, "en")
|
||||
assert "March" in result
|
||||
assert "15" in result
|
||||
assert "2025" in result
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_date_german(self) -> None:
|
||||
d = date(2025, 3, 15)
|
||||
result = format_date(d, "de")
|
||||
assert "15." in result
|
||||
assert "2025" in result
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_date_short(self) -> None:
|
||||
d = date(2025, 3, 15)
|
||||
result = format_date(d, "en", short=True)
|
||||
assert result == "03/15/2025"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_date_short_german(self) -> None:
|
||||
d = date(2025, 3, 15)
|
||||
result = format_date(d, "de", short=True)
|
||||
assert result == "15.03.2025"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_date_none(self) -> None:
|
||||
assert format_date(None) == ""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_datetime_none(self) -> None:
|
||||
assert format_datetime(None) == ""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_number_english(self) -> None:
|
||||
result = format_number(1234567, "en")
|
||||
assert result == "1,234,567"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_number_german(self) -> None:
|
||||
result = format_number(1234567, "de")
|
||||
assert result == "1.234.567"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_number_float_english(self) -> None:
|
||||
result = format_number(1234.56, "en")
|
||||
assert result == "1,234.56"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_number_float_german(self) -> None:
|
||||
result = format_number(1234.56, "de")
|
||||
assert result == "1.234,56"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_format_datetime_chinese(self) -> None:
|
||||
dt = datetime(2025, 3, 15, 14, 30)
|
||||
result = format_datetime(dt, "zh")
|
||||
assert "2025" in result
|
||||
assert "03" in result
|
||||
assert "15" in result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# get_language_info()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetLanguageInfo:
|
||||
"""Tests for get_language_info() utility."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_known_language(self) -> None:
|
||||
info = get_language_info("de")
|
||||
assert info is not None
|
||||
assert info["name"] == "German"
|
||||
assert info["native"] == "Deutsch"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_unknown_language(self) -> None:
|
||||
assert get_language_info("xx") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SUPPORTED_LANGUAGES metadata
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSupportedLanguages:
|
||||
"""Tests for language metadata constants."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_ten_languages_supported(self) -> None:
|
||||
assert len(SUPPORTED_LANGUAGES) == 31
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_supported_codes_set(self) -> None:
|
||||
expected = {
|
||||
"en",
|
||||
"de",
|
||||
"fr",
|
||||
"es",
|
||||
"it",
|
||||
"pt",
|
||||
"nl",
|
||||
"pl",
|
||||
"zh",
|
||||
"ru",
|
||||
"nb",
|
||||
"da",
|
||||
"sv",
|
||||
"fi",
|
||||
"is",
|
||||
"ga",
|
||||
"lb",
|
||||
"ca",
|
||||
"cs",
|
||||
"sk",
|
||||
"hu",
|
||||
"sl",
|
||||
"hr",
|
||||
"ro",
|
||||
"bg",
|
||||
"el",
|
||||
"et",
|
||||
"lv",
|
||||
"lt",
|
||||
"tr",
|
||||
"uk",
|
||||
}
|
||||
assert SUPPORTED_LANGUAGE_CODES == expected
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_default_language_is_english(self) -> None:
|
||||
assert DEFAULT_LANGUAGE == "en"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# API endpoint tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestI18nAPI:
|
||||
"""Tests for the i18n API endpoints."""
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_list_languages(self, client: TestClient) -> None:
|
||||
"""GET /api/i18n/languages should return all supported languages."""
|
||||
response = client.get("/api/i18n/languages")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "languages" in data
|
||||
assert len(data["languages"]) == 31
|
||||
assert data["default"] == "en"
|
||||
# Verify each language has required fields
|
||||
for lang in data["languages"]:
|
||||
assert "code" in lang
|
||||
assert "name" in lang
|
||||
assert "native" in lang
|
||||
assert "flag" in lang
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_set_language(self, client: TestClient) -> None:
|
||||
"""POST /api/i18n/language should set language preference."""
|
||||
response = client.post(
|
||||
"/api/i18n/language",
|
||||
json={"language": "de"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["language"] == "de"
|
||||
# Verify cookie was set
|
||||
assert "docuelevate_lang" in response.cookies
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_set_language_invalid_falls_back_to_default(self, client: TestClient) -> None:
|
||||
"""Invalid language code should fall back to default."""
|
||||
response = client.post(
|
||||
"/api/i18n/language",
|
||||
json={"language": "invalid"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["language"] == "en"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_set_language_persists_in_cookie(self, client: TestClient) -> None:
|
||||
"""Language setting should be persisted in a cookie."""
|
||||
client.post("/api/i18n/language", json={"language": "fr"})
|
||||
# Subsequent requests should detect the language from cookie
|
||||
response = client.get("/api/i18n/languages")
|
||||
data = response.json()
|
||||
assert data["current"] == "fr"
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_base_html_uses_current_locale(self, client: TestClient) -> None:
|
||||
"""The base template should set lang attribute to current locale."""
|
||||
# Set language to German
|
||||
client.post("/api/i18n/language", json={"language": "de"})
|
||||
# Load homepage
|
||||
response = client.get("/", follow_redirects=True)
|
||||
assert response.status_code == 200
|
||||
# The lang attribute should reflect the locale
|
||||
assert 'lang="de"' in response.text or 'lang="en"' in response.text
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_language_selector_in_nav(self, client: TestClient) -> None:
|
||||
"""The navigation should contain the language selector globe icon."""
|
||||
response = client.get("/", follow_redirects=True)
|
||||
if response.status_code == 200:
|
||||
assert "fa-globe" in response.text
|
||||
assert "setLanguage" in response.text
|
||||
@@ -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