gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into perf/async-url-upload-7099409897484162483

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-03-16 12:12:21 +01:00
committed by GitHub
parent 8ce41d723e 66fdb11e39
commit b8a1ac52b3
46 changed files with 2356 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_advanced_filters.py
+140
View File
@@ -269,6 +269,53 @@ class TestSavedSearchesCRUD:
response2 = client.post("/api/saved-searches", json=payload)
assert response2.status_code == 409
def test_create_saved_search_db_error(self, client: TestClient, monkeypatch):
"""POST /api/saved-searches returns 500 on DB exception."""
# Mock db.add or db.commit to raise an exception
# We can monkeypatch the route's dependency or the models
# It's easier to mock the SavedSearch model's __init__ or db's add
# Since we use db: DbSession, it's an instance of sqlalchemy.orm.Session
from sqlalchemy.orm import Session
original_commit = Session.commit
def mock_commit(*args, **kwargs):
raise Exception("Simulated DB error")
monkeypatch.setattr(Session, "commit", mock_commit)
payload = {
"name": "DB Error Search",
"filters": {"status": "completed"},
}
response = client.post("/api/saved-searches", json=payload)
assert response.status_code == 500
assert "Failed to save search" in response.json()["detail"]
def test_create_saved_search_limit_reached(self, client: TestClient, monkeypatch):
"""POST /api/saved-searches returns 409 if max limit is reached."""
monkeypatch.setattr("app.api.saved_searches.MAX_SAVED_SEARCHES_PER_USER", 1)
# Create first one
payload1 = {"name": "Search 1", "filters": {"status": "completed"}}
response1 = client.post("/api/saved-searches", json=payload1)
assert response1.status_code == 201
# Creating second one should fail due to limit
payload2 = {"name": "Search 2", "filters": {"status": "pending"}}
response2 = client.post("/api/saved-searches", json=payload2)
assert response2.status_code == 409
assert "Maximum of 1 saved searches reached" in response2.json()["detail"]
def test_create_saved_search_invalid_name_type(self, client: TestClient):
"""POST /api/saved-searches with non-string name returns 422."""
payload = {
"name": 12345,
"filters": {"status": "completed"},
}
response = client.post("/api/saved-searches", json=payload)
assert response.status_code == 422
def test_update_saved_search(self, client: TestClient):
"""PUT /api/saved-searches/{id} updates the saved search."""
# Create
@@ -296,6 +343,83 @@ class TestSavedSearchesCRUD:
)
assert response.status_code == 404
def test_update_saved_search_duplicate_name(self, client: TestClient):
"""PUT /api/saved-searches/{id} with duplicate name returns 409."""
# Create first search
client.post(
"/api/saved-searches",
json={"name": "First Search", "filters": {"status": "pending"}},
)
# Create second search
create_resp2 = client.post(
"/api/saved-searches",
json={"name": "Second Search", "filters": {"status": "completed"}},
)
search_id2 = create_resp2.json()["id"]
# Try to rename second search to "First Search"
update_resp = client.put(
f"/api/saved-searches/{search_id2}",
json={"name": "First Search", "filters": {"status": "completed"}},
)
assert update_resp.status_code == 409
def test_update_saved_search_name_too_long(self, client: TestClient):
"""PUT /api/saved-searches/{id} with name > 100 chars returns 422."""
create_resp = client.post(
"/api/saved-searches",
json={"name": "Valid Name", "filters": {"status": "pending"}},
)
search_id = create_resp.json()["id"]
update_resp = client.put(
f"/api/saved-searches/{search_id}",
json={"name": "x" * 101, "filters": {"status": "completed"}},
)
assert update_resp.status_code == 422
def test_update_saved_search_empty_name(self, client: TestClient):
"""PUT /api/saved-searches/{id} with empty name returns 422."""
create_resp = client.post(
"/api/saved-searches",
json={"name": "Valid Name", "filters": {"status": "pending"}},
)
search_id = create_resp.json()["id"]
update_resp = client.put(
f"/api/saved-searches/{search_id}",
json={"name": "", "filters": {"status": "completed"}},
)
assert update_resp.status_code == 422
def test_update_saved_search_empty_filters(self, client: TestClient):
"""PUT /api/saved-searches/{id} with empty filters returns 422."""
create_resp = client.post(
"/api/saved-searches",
json={"name": "Valid Name", "filters": {"status": "pending"}},
)
search_id = create_resp.json()["id"]
update_resp = client.put(
f"/api/saved-searches/{search_id}",
json={"name": "Valid Name", "filters": {}},
)
assert update_resp.status_code == 422
def test_update_saved_search_invalid_filters(self, client: TestClient):
"""PUT /api/saved-searches/{id} with only invalid filters returns 422."""
create_resp = client.post(
"/api/saved-searches",
json={"name": "Valid Name", "filters": {"status": "pending"}},
)
search_id = create_resp.json()["id"]
update_resp = client.put(
f"/api/saved-searches/{search_id}",
json={"name": "Valid Name", "filters": {"invalid_key": "value"}},
)
assert update_resp.status_code == 422
def test_delete_saved_search(self, client: TestClient):
"""DELETE /api/saved-searches/{id} removes the saved search."""
# Create
@@ -318,6 +442,22 @@ class TestSavedSearchesCRUD:
response = client.delete("/api/saved-searches/999")
assert response.status_code == 404
def test_delete_saved_search_db_error(self, client: TestClient):
"""DELETE /api/saved-searches/{id} handles database errors (500)."""
from unittest.mock import patch
# Create
create_resp = client.post(
"/api/saved-searches",
json={"name": "To Delete DB Error", "filters": {"status": "failed"}},
)
search_id = create_resp.json()["id"]
with patch("sqlalchemy.orm.Session.delete", side_effect=Exception("DB Delete Error")):
response = client.delete(f"/api/saved-searches/{search_id}")
assert response.status_code == 500
assert response.json()["detail"] == "Failed to delete saved search"
def test_create_name_too_long(self, client: TestClient):
"""POST /api/saved-searches with name > 100 chars returns 422."""
payload = {
tests/test_api_dropbox.py
+11 -7
View File
@@ -7,7 +7,6 @@ Covers Dropbox OAuth endpoints, settings management, and token testing.
from unittest.mock import Mock, patch
import pytest
import requests
@pytest.mark.unit
@@ -137,7 +136,7 @@ class TestTestDropboxToken:
assert data["status"] == "error"
assert "not fully configured" in data["message"]
@patch("app.api.dropbox.requests.post")
@patch("app.api.dropbox.httpx.AsyncClient.post")
@patch("app.api.dropbox.settings")
def test_valid_token(self, mock_settings, mock_post, client):
"""Test successful token validation."""
@@ -162,7 +161,7 @@ class TestTestDropboxToken:
assert data["account"] == "user@example.com"
assert data["account_name"] == "Test User"
@patch("app.api.dropbox.requests.post")
@patch("app.api.dropbox.httpx.AsyncClient.post")
@patch("app.api.dropbox.settings")
def test_expired_token_refreshed(self, mock_settings, mock_post, client):
"""Test that expired token triggers refresh and retry."""
@@ -194,7 +193,7 @@ class TestTestDropboxToken:
data = response.json()
assert data["status"] == "success"
@patch("app.api.dropbox.requests.post")
@patch("app.api.dropbox.httpx.AsyncClient.post")
@patch("app.api.dropbox.settings")
def test_refresh_token_expired(self, mock_settings, mock_post, client):
"""Test handling when refresh token itself is expired."""
@@ -220,7 +219,7 @@ class TestTestDropboxToken:
assert data["status"] == "error"
assert data["needs_reauth"] is True
@patch("app.api.dropbox.requests.post")
@patch("app.api.dropbox.httpx.AsyncClient.post")
@patch("app.api.dropbox.settings")
def test_token_validation_failure(self, mock_settings, mock_post, client):
"""Test handling non-401, non-200 response."""
@@ -240,16 +239,21 @@ class TestTestDropboxToken:
data = response.json()
assert data["status"] == "error"
@patch("app.api.dropbox.requests.post")
@patch("app.api.dropbox.httpx.AsyncClient.post")
@patch("app.api.dropbox.settings")
def test_connection_error(self, mock_settings, mock_post, client):
"""Test handling of connection exceptions."""
import httpx
mock_settings.dropbox_refresh_token = "token"
mock_settings.dropbox_app_key = "app-key"
mock_settings.dropbox_app_secret = "app-secret"
mock_settings.http_request_timeout = 30
mock_post.side_effect = requests.exceptions.ConnectionError("Connection refused")
mock_post.side_effect = httpx.RequestError(
"Connection refused",
request=httpx.Request("POST", "https://api.dropboxapi.com/2/users/get_current_account"),
)
response = client.get("/api/dropbox/test-token")
tests/test_api_dropbox_extended.py
+23 -19
View File
@@ -84,8 +84,8 @@ class TestUpdateDropboxSettings:
class TestTestDropboxToken:
"""Tests for GET /dropbox/test-token endpoint."""
@patch("app.api.dropbox.requests.post")
def test_test_token_success(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_success(self, mock_client_cls):
"""Test successful token validation."""
from app.config import settings
@@ -95,7 +95,7 @@ class TestTestDropboxToken:
"email": "test@example.com",
"name": {"display_name": "Test User"},
}
mock_post.return_value = mock_response
mock_client_cls.return_value.__aenter__.return_value.post.return_value = mock_response
with patch.object(settings, "dropbox_refresh_token", "token"):
with patch.object(settings, "dropbox_app_key", "key"):
@@ -104,8 +104,8 @@ class TestTestDropboxToken:
# Should include account email and name
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_not_configured(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_not_configured(self, mock_client_cls):
"""Test when credentials are not configured."""
from app.config import settings
@@ -113,8 +113,8 @@ class TestTestDropboxToken:
# Should return error indicating not configured
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_partial_config(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_partial_config(self, mock_client_cls):
"""Test with partial configuration (missing some credentials)."""
from app.config import settings
@@ -123,8 +123,8 @@ class TestTestDropboxToken:
# Should return error
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_expired_requires_refresh(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_expired_requires_refresh(self, mock_client_cls):
"""Test when access token is expired and needs refresh."""
from app.config import settings
@@ -145,7 +145,9 @@ class TestTestDropboxToken:
"name": {"display_name": "Test User"},
}
mock_post.side_effect = [mock_response_401, mock_refresh_response, mock_success_response]
mock_client = MagicMock()
mock_client.post.side_effect = [mock_response_401, mock_refresh_response, mock_success_response]
mock_client_cls.return_value.__aenter__.return_value = mock_client
with patch.object(settings, "dropbox_refresh_token", "token"):
with patch.object(settings, "dropbox_app_key", "key"):
@@ -153,8 +155,8 @@ class TestTestDropboxToken:
# Should refresh and succeed
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_refresh_failed(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_refresh_failed(self, mock_client_cls):
"""Test when refresh token is invalid."""
from app.config import settings
@@ -167,7 +169,9 @@ class TestTestDropboxToken:
mock_refresh_response.status_code = 400
mock_refresh_response.text = "Invalid refresh token"
mock_post.side_effect = [mock_response_401, mock_refresh_response]
mock_client = MagicMock()
mock_client.post.side_effect = [mock_response_401, mock_refresh_response]
mock_client_cls.return_value.__aenter__.return_value = mock_client
with patch.object(settings, "dropbox_refresh_token", "token"):
with patch.object(settings, "dropbox_app_key", "key"):
@@ -175,8 +179,8 @@ class TestTestDropboxToken:
# Should return error with needs_reauth: True
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_perpetual_token_info(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_perpetual_token_info(self, mock_client_cls):
"""Test that perpetual token info is returned."""
from app.config import settings
@@ -186,7 +190,7 @@ class TestTestDropboxToken:
"email": "test@example.com",
"name": {"display_name": "Test User"},
}
mock_post.return_value = mock_response
mock_client_cls.return_value.__aenter__.return_value.post.return_value = mock_response
with patch.object(settings, "dropbox_refresh_token", "token"):
with patch.object(settings, "dropbox_app_key", "key"):
@@ -194,12 +198,12 @@ class TestTestDropboxToken:
# token_info should indicate never expires
pass
@patch("app.api.dropbox.requests.post")
def test_test_token_exception_handling(self, mock_post):
@patch("app.api.dropbox.httpx.AsyncClient")
def test_test_token_exception_handling(self, mock_client_cls):
"""Test handling of exceptions."""
from app.config import settings
mock_post.side_effect = Exception("Network error")
mock_client_cls.return_value.__aenter__.return_value.post.side_effect = Exception("Network error")
with patch.object(settings, "dropbox_refresh_token", "token"):
with patch.object(settings, "dropbox_app_key", "key"):
tests/test_api_google_drive_final.py
+8 -8
View File
@@ -7,9 +7,9 @@ Targets the remaining uncovered branches from the 97.03% baseline:
- 214 : test_google_drive_token — generic connection error (not token-related)
- 302->306: get_google_drive_token_info — credentials already valid (no refresh)
- 307->318: get_google_drive_token_info — credentials have no expiry
- 395->397: save_dropbox_settings — refresh_token falsy inside use_oauth block
- 449->451: save_dropbox_settings — refresh_token falsy in in-memory update
- 468->470: save_dropbox_settings — folder_id falsy in db-persist block
- 395->397: save_google_drive_settings — refresh_token falsy inside use_oauth block
- 449->451: save_google_drive_settings — refresh_token falsy in in-memory update
- 468->470: save_google_drive_settings — folder_id falsy in db-persist block
"""
from datetime import datetime, timedelta
@@ -152,10 +152,10 @@ class TestGetTokenInfoCredentialsBranches:
@pytest.mark.unit
class TestSaveGoogleDriveSettingsFalsyFields:
"""Cover branches 395->397, 449->451, 468->470 in save_dropbox_settings.
"""Cover branches 395->397, 449->451, 468->470 in save_google_drive_settings.
Note: the Google Drive save endpoint is named save_dropbox_settings in the
source (app/api/google_drive.py) due to an existing naming inconsistency.
Note: the Google Drive save endpoint is named save_google_drive_settings in the
source (app/api/google_drive.py).
"""
@patch("app.api.google_drive.settings")
@@ -167,7 +167,7 @@ class TestSaveGoogleDriveSettingsFalsyFields:
from starlette.requests import Request as StarletteRequest
from app.api.google_drive import save_dropbox_settings
from app.api.google_drive import save_google_drive_settings
mock_request = MagicMock(spec=StarletteRequest)
mock_request.session = {}
@@ -175,7 +175,7 @@ class TestSaveGoogleDriveSettingsFalsyFields:
with patch("app.api.google_drive.save_setting_to_db"):
with patch("app.api.google_drive.notify_settings_updated"):
result = await save_dropbox_settings(
result = await save_google_drive_settings(
request=mock_request,
refresh_token="", # falsy → branches 395->397 and 449->451
client_id="cid",
tests/test_api_onedrive_coverage.py
+7 -7
View File
@@ -75,8 +75,8 @@ class TestTestTokenRotation:
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("app.utils.env_utils.os.path.join", return_value=str(env_file)),
patch("app.utils.env_utils.os.path.exists", return_value=True),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
@@ -117,7 +117,7 @@ class TestTestTokenRotation:
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=False),
patch("app.utils.env_utils.os.path.exists", return_value=False),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
@@ -157,7 +157,7 @@ class TestTestTokenRotation:
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("app.utils.env_utils.os.path.exists", return_value=True),
patch("builtins.open", side_effect=PermissionError("Permission denied")),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
@@ -198,7 +198,7 @@ class TestTestTokenRotation:
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=False),
patch("app.utils.env_utils.os.path.exists", return_value=False),
patch("app.database.SessionLocal", side_effect=Exception("DB error")),
):
response = client.get("/api/onedrive/test-token")
@@ -277,8 +277,8 @@ class TestTokenRotationEnvAppendLine:
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("app.utils.env_utils.os.path.join", return_value=str(env_file)),
patch("app.utils.env_utils.os.path.exists", return_value=True),
patch("app.database.SessionLocal") as mock_sl,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
tests/test_api_saved_searches.py
+191
View File
@@ -0,0 +1,191 @@
"""Tests for the saved searches API (app/api/saved_searches.py)."""
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 SavedSearch
# ---------------------------------------------------------------------------
# Test data constants
# ---------------------------------------------------------------------------
_OWNER = "test_user@example.com"
_OTHER_OWNER = "other_user@example.com"
# ---------------------------------------------------------------------------
# Shared fixture helpers
# ---------------------------------------------------------------------------
@pytest.fixture()
def int_engine():
"""In-memory SQLite engine for integration 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 int_session(int_engine):
"""DB session scoped to one test."""
Session = sessionmaker(bind=int_engine)
session = Session()
yield session
session.close()
def _make_client(int_engine, owner_id: str = _OWNER):
"""Return a TestClient with *owner_id* injected as the authenticated user."""
from unittest.mock import patch
from app.main import app
def override_db():
Session = sessionmaker(bind=int_engine)
session = Session()
try:
yield session
finally:
session.close()
app.dependency_overrides[get_db] = override_db
with patch("app.api.saved_searches._get_user_id", return_value=owner_id):
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
yield client
app.dependency_overrides.clear()
@pytest.fixture()
def int_client(int_engine):
"""TestClient authenticated as _OWNER."""
yield from _make_client(int_engine, _OWNER)
# ---------------------------------------------------------------------------
# CRUD tests
# ---------------------------------------------------------------------------
@pytest.mark.integration
class TestSavedSearchesAPI:
"""Tests for Saved Searches endpoints."""
def test_list_saved_searches_empty(self, int_client):
"""No saved searches returns empty list."""
resp = int_client.get("/api/saved-searches")
assert resp.status_code == 200
assert resp.json() == []
def test_create_saved_search(self, int_client):
"""Create a saved search and verify the response."""
payload = {"name": "My Invoices", "filters": {"tags": "invoice", "document_type": "Invoice"}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "My Invoices"
assert data["filters"] == {"tags": "invoice", "document_type": "Invoice"}
assert "id" in data
def test_create_saved_search_invalid_filters(self, int_client):
"""Creating with invalid filters returns 422."""
# Missing filters parameter (or empty after sanitization)
payload = {"name": "My Invoices", "filters": {}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 422
# Invalid filters format
payload2 = {"name": "My Invoices", "filters": "not_a_dict"}
resp2 = int_client.post("/api/saved-searches", json=payload2)
assert resp2.status_code == 422
def test_create_saved_search_duplicate(self, int_client):
"""Creating a duplicate named search returns 409."""
payload = {"name": "Duplicate", "filters": {"q": "test"}}
int_client.post("/api/saved-searches", json=payload)
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 409
def test_create_saved_search_limit(self, int_client, int_session):
"""Exceeding MAX_SAVED_SEARCHES_PER_USER returns 409."""
# Create 50 searches using the API to ensure they are visible
for i in range(50):
resp = int_client.post("/api/saved-searches", json={"name": f"Search LIMIT {i}", "filters": {"q": "test"}})
assert resp.status_code == 201
payload = {"name": "One too many", "filters": {"q": "test"}}
resp = int_client.post("/api/saved-searches", json=payload)
assert resp.status_code == 409
def test_update_saved_search(self, int_client):
"""Update an existing saved search."""
payload = {"name": "Original Name", "filters": {"q": "test"}}
created = int_client.post("/api/saved-searches", json=payload).json()
search_id = created["id"]
update_payload = {"name": "Updated Name", "filters": {"tags": "new"}}
resp = int_client.put(f"/api/saved-searches/{search_id}", json=update_payload)
assert resp.status_code == 200
data = resp.json()
assert data["name"] == "Updated Name"
assert data["filters"] == {"tags": "new"}
def test_update_saved_search_not_found(self, int_client):
"""Updating a non-existent search returns 404."""
update_payload = {"name": "Updated Name"}
resp = int_client.put("/api/saved-searches/999", json=update_payload)
assert resp.status_code == 404
def test_update_saved_search_duplicate_name(self, int_client):
"""Updating name to an existing search name returns 409."""
payload1 = {"name": "Search 1", "filters": {"q": "a"}}
payload2 = {"name": "Search 2", "filters": {"q": "b"}}
int_client.post("/api/saved-searches", json=payload1)
created2 = int_client.post("/api/saved-searches", json=payload2).json()
search2_id = created2["id"]
update_payload = {"name": "Search 1"}
resp = int_client.put(f"/api/saved-searches/{search2_id}", json=update_payload)
assert resp.status_code == 409
def test_delete_saved_search(self, int_client, int_session):
"""Delete an existing search."""
payload = {"name": "To be deleted", "filters": {"q": "test"}}
created = int_client.post("/api/saved-searches", json=payload).json()
search_id = created["id"]
resp = int_client.delete(f"/api/saved-searches/{search_id}")
assert resp.status_code == 204
assert int_session.query(SavedSearch).filter(SavedSearch.id == search_id).first() is None
def test_delete_saved_search_not_found(self, int_client):
"""Deleting a non-existent search returns 404."""
resp = int_client.delete("/api/saved-searches/999")
assert resp.status_code == 404
def test_other_users_searches_isolated(self, int_engine, int_session):
"""Users only see and can only modify their own saved searches."""
int_session.add(SavedSearch(user_id=_OTHER_OWNER, name="Other Search", filters='{"q": "test"}'))
int_session.commit()
client = next(_make_client(int_engine, _OWNER))
resp = client.get("/api/saved-searches")
assert resp.status_code == 200
assert len(resp.json()) == 0
other_search = int_session.query(SavedSearch).first()
resp = client.put(f"/api/saved-searches/{other_search.id}", json={"name": "Hacked"})
assert resp.status_code == 404
resp = client.delete(f"/api/saved-searches/{other_search.id}")
assert resp.status_code == 404
tests/test_api_tokens.py
+129 -6
View File
@@ -101,6 +101,43 @@ def _cleanup(app):
app.dependency_overrides.clear()
# ---------------------------------------------------------------------------
# Tests – Auth Helper
# ---------------------------------------------------------------------------
class TestGetOwnerId:
"""Tests for the _get_owner_id dependency helper."""
@pytest.mark.unit
def test_get_owner_id_unauthenticated(self):
"""_get_owner_id should raise a 401 if the user is not authenticated."""
from unittest.mock import MagicMock, patch
from fastapi import HTTPException
from app.api.api_tokens import _get_owner_id
mock_request = MagicMock()
with patch("app.api.api_tokens.get_current_owner_id", return_value=None):
with pytest.raises(HTTPException) as exc_info:
_get_owner_id(mock_request)
assert exc_info.value.status_code == 401
assert exc_info.value.detail == "Not authenticated"
@pytest.mark.unit
def test_get_owner_id_authenticated(self):
"""_get_owner_id should return owner_id if user is authenticated."""
from unittest.mock import MagicMock, patch
from app.api.api_tokens import _get_owner_id
mock_request = MagicMock()
with patch("app.api.api_tokens.get_current_owner_id", return_value="owner123"):
owner_id = _get_owner_id(mock_request)
assert owner_id == "owner123"
# ---------------------------------------------------------------------------
# Tests – Token CRUD
# ---------------------------------------------------------------------------
@@ -157,6 +194,46 @@ class TestTokenCreate:
finally:
_cleanup(app)
@pytest.mark.unit
def test_create_token_database_error(self, tok_engine, tok_session):
"""Creating a token should rollback and raise 500 if database commit fails."""
from unittest.mock import patch
from sqlalchemy.orm import Session as SASession
from app.main import app
client = _make_client(tok_engine)
try:
# Wrap commit: flush first so changes are staged in the transaction,
# then raise to simulate a commit failure after data has been written.
def _fail_after_flush(self):
self.flush() # stage changes inside the open transaction
raise Exception("DB Failure")
# Spy on rollback so we can assert it is called.
rollback_called = False
real_rollback = SASession.rollback
def _spy_rollback(self):
nonlocal rollback_called
rollback_called = True
real_rollback(self)
with patch.object(SASession, "commit", _fail_after_flush):
with patch.object(SASession, "rollback", _spy_rollback):
resp = client.post("/api/api-tokens/", json={"name": "DB Error Create Test"})
assert resp.status_code == 500
# rollback() must have been called to undo the flushed changes.
assert rollback_called, "db.rollback() was not called after commit failure in create_token"
# After rollback the token must not exist in the database.
db_token = tok_session.query(ApiToken).filter(ApiToken.name == "DB Error Create Test").first()
assert db_token is None
finally:
_cleanup(app)
class TestTokenList:
"""Tests for GET /api/api-tokens/."""
@@ -326,12 +403,10 @@ class TestTokenRevoke:
rollback_called = True
real_rollback(self)
with (
patch.object(SASession, "commit", _fail_after_flush),
patch.object(SASession, "rollback", _spy_rollback),
):
resp = client.delete(f"/api/api-tokens/{token_id}")
assert resp.status_code == 500
with patch.object(SASession, "commit", _fail_after_flush):
with patch.object(SASession, "rollback", _spy_rollback):
resp = client.delete(f"/api/api-tokens/{token_id}")
assert resp.status_code == 500
# rollback() must have been called to undo the flushed changes.
assert rollback_called, "db.rollback() was not called after commit failure"
@@ -534,6 +609,44 @@ class TestTokenUtils:
tokens = {generate_api_token() for _ in range(100)}
assert len(tokens) == 100
@pytest.mark.unit
def test_generate_api_token_length(self):
"""Generated tokens should have the exact expected length based on TOKEN_BYTES."""
import math
from app.api.api_tokens import TOKEN_BYTES, TOKEN_PREFIX, generate_api_token
# base64url encoding of N bytes without padding: ceil(N * 4 / 3) characters
expected_b64_len = math.ceil(TOKEN_BYTES * 4 / 3)
expected_total_len = len(TOKEN_PREFIX) + expected_b64_len
token = generate_api_token()
assert len(token) == expected_total_len
@pytest.mark.unit
def test_generate_api_token_charset(self):
"""Generated tokens should only contain URL-safe base64 characters and the prefix."""
import re
from app.api.api_tokens import TOKEN_PREFIX, generate_api_token
token = generate_api_token()
# Check it starts with prefix and the rest is base64url chars ([A-Za-z0-9_-])
pattern = f"^{re.escape(TOKEN_PREFIX)}[A-Za-z0-9_\\-]+$"
assert re.match(pattern, token) is not None
@pytest.mark.unit
def test_generate_api_token_uses_secrets(self):
"""Generated tokens should use secrets.token_urlsafe with the correct number of bytes."""
from unittest.mock import patch
from app.api.api_tokens import TOKEN_BYTES, TOKEN_PREFIX, generate_api_token
with patch("app.api.api_tokens.secrets.token_urlsafe", return_value="mocked_token") as mock_secrets:
token = generate_api_token()
mock_secrets.assert_called_once_with(TOKEN_BYTES)
assert token == f"{TOKEN_PREFIX}mocked_token"
@pytest.mark.unit
def test_hash_token_deterministic(self):
"""Hashing the same token should always produce the same result."""
@@ -554,3 +667,13 @@ class TestTokenUtils:
# All characters should be valid lowercase hex digits.
int(h, 16)
assert h == h.lower()
@pytest.mark.unit
def test_hash_token_known_value(self):
"""hash_token should return the exact expected PBKDF2 digest for a known input."""
from app.api.api_tokens import hash_token
# PBKDF2-HMAC-SHA256 with 100,000 iterations and salt b"api-token-v1"
token = "de_test_token_value"
expected_hash = "9b89d9adf2f390c75bf2fd0ff2bb5622ef5a9dce438354cce6e39f2f5401129e"
assert hash_token(token) == expected_hash
tests/test_auth.py
+35
View File
@@ -2,6 +2,7 @@
import asyncio
import hashlib
import logging
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -38,6 +39,40 @@ class TestGetCurrentUser:
result = get_current_user(mock_request)
assert result is None
def test_logs_debug_when_session_user_found(self, caplog):
"""Test that get_current_user emits a DEBUG log when session user is found."""
mock_request = MagicMock(spec=Request)
mock_request.session = {"user": {"id": "u1", "preferred_username": "alice"}}
mock_request.state = MagicMock(spec=[]) # no api_token_user attribute
with caplog.at_level(logging.DEBUG, logger="app.auth"):
get_current_user(mock_request)
assert any("[AUTH] get_current_user: resolved from session" in m for m in caplog.messages)
def test_logs_debug_when_no_user(self, caplog):
"""Test that get_current_user emits a DEBUG log when no user is present."""
mock_request = MagicMock(spec=Request)
mock_request.session = {}
mock_request.state = MagicMock(spec=[])
with caplog.at_level(logging.DEBUG, logger="app.auth"):
get_current_user(mock_request)
assert any("[AUTH] get_current_user: no user in session or API token" in m for m in caplog.messages)
def test_logs_debug_when_api_token_user(self, caplog):
"""Test that get_current_user emits a DEBUG log when resolved from API token."""
mock_request = MagicMock(spec=Request)
mock_request.state.api_token_user = {"id": "tok_user"}
mock_request.session = {}
with caplog.at_level(logging.DEBUG, logger="app.auth"):
result = get_current_user(mock_request)
assert result == {"id": "tok_user"}
assert any("[AUTH] get_current_user: resolved from API token" in m for m in caplog.messages)
@pytest.mark.unit
class TestGetGravatarUrl:
tests/test_coverage_remaining_gaps.py
+149 -2
View File
@@ -704,6 +704,18 @@ def _set_minimal_provider_settings(mock_settings):
class TestSettingsSyncAdditional:
"""Additional tests for settings_sync covering reload failure branch."""
def test_notify_settings_updated_redis_failure_logs_warning(self):
"""Test that a Redis failure is logged, not raised (lines 58-59)."""
from app.utils.settings_sync import notify_settings_updated
with patch("app.utils.settings_sync.redis") as mock_redis_module:
mock_redis_module.from_url.side_effect = Exception("Redis connection failed")
with patch("app.utils.settings_sync.logger") as mock_logger:
notify_settings_updated()
mock_logger.warning.assert_any_call(
"Could not publish settings update to Redis: Redis connection failed"
)
def test_reload_failure_is_logged_not_raised(self):
"""Test that a reload failure is logged, not raised (lines 71-72)."""
from app.utils.settings_sync import notify_settings_updated
@@ -711,8 +723,66 @@ class TestSettingsSyncAdditional:
with patch("app.utils.settings_sync.redis") as mock_redis_module:
mock_redis_module.from_url.return_value = MagicMock() # Redis OK
with patch("app.utils.config_loader.reload_settings_from_db", side_effect=Exception("reload failed")):
# Should not raise despite reload failure
notify_settings_updated()
with patch("app.utils.settings_sync.logger") as mock_logger:
# Should not raise despite reload failure
notify_settings_updated()
mock_logger.warning.assert_any_call("Could not reload in-process settings: reload failed")
def test_notify_settings_updated_ocr_failure_logs_warning(self):
"""Test that OCR check failure is logged, not raised (lines 82-83)."""
from app.utils.settings_sync import notify_settings_updated
with patch("app.utils.settings_sync.redis") as mock_redis_module:
mock_redis_module.from_url.return_value = MagicMock() # Redis OK
with patch("app.utils.config_loader.reload_settings_from_db"):
with patch(
"app.utils.ocr_language_manager.ensure_ocr_languages_async", side_effect=Exception("OCR failed")
):
with patch("app.utils.settings_sync.logger") as mock_logger:
notify_settings_updated()
mock_logger.warning.assert_any_call("Could not schedule OCR language check: OCR failed")
def test_signal_handler_ocr_check_failure_logs_warning(self):
"""Test that signal handler logs warning if OCR language check fails on worker (lines 115-116)."""
from app.utils.settings_sync import register_settings_reload_signal
handler_fn = None
def capture_connect(fn=None, weak=None, **kwargs):
nonlocal handler_fn
if fn is not None:
handler_fn = fn
return fn
def decorator(func):
nonlocal handler_fn
handler_fn = func
return func
return decorator
with patch("app.utils.settings_sync.task_prerun") as mock_signal:
mock_signal.connect = capture_connect
register_settings_reload_signal()
assert handler_fn is not None
mock_redis = MagicMock()
mock_redis.get.return_value = b"1234567890.0"
with patch("app.utils.settings_sync.redis") as mock_redis_mod:
mock_redis_mod.from_url.return_value = mock_redis
with patch("app.utils.config_loader.reload_settings_from_db"):
with patch("app.utils.settings_sync._last_seen_version", ""):
with patch(
"app.utils.ocr_language_manager.ensure_ocr_languages_async",
side_effect=Exception("Worker OCR fail"),
):
with patch("app.utils.settings_sync.logger") as mock_logger:
handler_fn(sender=None)
mock_logger.warning.assert_any_call(
"Could not schedule OCR language check on worker: Worker OCR fail"
)
def test_signal_handler_reloads_on_version_change(self):
"""Test the task_prerun signal handler reloads settings when version changes (lines 95-98)."""
@@ -820,6 +890,83 @@ class TestSettingsSyncAdditional:
# Should not raise
handler_fn(sender=None)
def test_signal_handler_no_version_returned(self):
"""Test that handler does nothing if Redis returns None for version."""
from app.utils.settings_sync import register_settings_reload_signal
handler_fn = None
def capture_connect(fn=None, weak=None, **kwargs):
nonlocal handler_fn
if fn is not None:
handler_fn = fn
return fn
def decorator(func):
nonlocal handler_fn
handler_fn = func
return func
return decorator
with patch("app.utils.settings_sync.task_prerun") as mock_signal:
mock_signal.connect = capture_connect
register_settings_reload_signal()
assert handler_fn is not None
mock_redis = MagicMock()
mock_redis.get.return_value = None # Return None for version
with patch("app.utils.settings_sync.redis") as mock_redis_mod:
mock_redis_mod.from_url.return_value = mock_redis
with patch("app.utils.config_loader.reload_settings_from_db") as mock_reload:
handler_fn(sender=None)
mock_reload.assert_not_called()
def test_signal_handler_ocr_language_manager_exception(self):
"""Test that OCR language check exception inside handler is caught and logged."""
from app.utils.settings_sync import register_settings_reload_signal
handler_fn = None
def capture_connect(fn=None, weak=None, **kwargs):
nonlocal handler_fn
if fn is not None:
handler_fn = fn
return fn
def decorator(func):
nonlocal handler_fn
handler_fn = func
return func
return decorator
with patch("app.utils.settings_sync.task_prerun") as mock_signal:
mock_signal.connect = capture_connect
register_settings_reload_signal()
assert handler_fn is not None
mock_redis = MagicMock()
mock_redis.get.return_value = b"9999999.0" # New version
with patch("app.utils.settings_sync.redis") as mock_redis_mod:
mock_redis_mod.from_url.return_value = mock_redis
with patch("app.utils.config_loader.reload_settings_from_db") as mock_reload:
with patch("app.utils.settings_sync._last_seen_version", "111.0"):
with patch("app.utils.settings_sync.logger") as mock_logger:
with patch(
"app.utils.ocr_language_manager.ensure_ocr_languages_async",
side_effect=Exception("OCR failed"),
):
handler_fn(sender=None)
mock_reload.assert_called_once()
mock_logger.warning.assert_called_with(
"Could not schedule OCR language check on worker: OCR failed"
)
# ===========================================================================
# app/api/logs.py – additional branches
tests/test_file_upload.py
+1 -1
View File
@@ -345,7 +345,7 @@ class TestUploadErrorHandling:
def test_upload_disk_write_failure(self, client: TestClient):
"""Test handling of disk write failures."""
with patch("builtins.open", side_effect=IOError("Disk full")):
with patch("aiofiles.open", side_effect=IOError("Disk full")):
pdf_content = b"%PDF-1.4\n%EOF"
response = client.post(
tests/test_logging_config.py
+285
View File
@@ -0,0 +1,285 @@
"""Tests for application logging configuration.
Validates that the LOG_LEVEL and DEBUG settings correctly control the
Python root-logger level and that the standard precedence rules are respected:
1. Explicit LOG_LEVEL always wins.
2. DEBUG=True without LOG_LEVEL → effective DEBUG.
3. Neither set → default INFO.
"""
import logging
import os
from unittest.mock import patch
import pytest
from app.config import Settings
@pytest.mark.unit
class TestLogLevelSetting:
"""Tests for the log_level config field."""
_BASE_KWARGS = {
"database_url": "sqlite:///test.db",
"redis_url": "redis://localhost:6379",
"openai_api_key": "test",
"azure_ai_key": "test",
"azure_region": "test",
"azure_endpoint": "https://test.example.com",
"gotenberg_url": "http://localhost:3000",
"workdir": "/tmp",
"auth_enabled": False,
"session_secret": None,
}
def test_log_level_default_is_info(self):
"""Test that log_level defaults to INFO."""
config = Settings(**self._BASE_KWARGS)
assert config.log_level.upper() == "INFO"
def test_log_level_accepts_debug(self):
"""Test that log_level accepts DEBUG."""
config = Settings(**self._BASE_KWARGS, log_level="DEBUG")
assert config.log_level.upper() == "DEBUG"
def test_log_level_accepts_warning(self):
"""Test that log_level accepts WARNING."""
config = Settings(**self._BASE_KWARGS, log_level="WARNING")
assert config.log_level.upper() == "WARNING"
def test_log_level_accepts_error(self):
"""Test that log_level accepts ERROR."""
config = Settings(**self._BASE_KWARGS, log_level="ERROR")
assert config.log_level.upper() == "ERROR"
def test_log_level_case_insensitive(self):
"""Test that log_level is case-insensitive in usage."""
config = Settings(**self._BASE_KWARGS, log_level="debug")
assert config.log_level.upper() == "DEBUG"
def test_debug_flag_defaults_to_false(self):
"""Test that debug defaults to False."""
config = Settings(**self._BASE_KWARGS)
assert config.debug is False
@pytest.mark.unit
class TestEffectiveLogLevel:
"""Tests for the effective log-level resolution logic in main.py."""
def test_debug_true_without_log_level_gives_debug(self):
"""When DEBUG=True and LOG_LEVEL is not set, effective level is DEBUG."""
with patch.dict(os.environ, {"DEBUG": "true"}, clear=False):
# Remove LOG_LEVEL from env if present
env = os.environ.copy()
env.pop("LOG_LEVEL", None)
with patch.dict(os.environ, env, clear=True):
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
debug=True,
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "DEBUG"
def test_explicit_log_level_overrides_debug(self):
"""When LOG_LEVEL is explicitly set, it takes precedence over DEBUG=True."""
with patch.dict(os.environ, {"LOG_LEVEL": "WARNING", "DEBUG": "true"}, clear=False):
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
debug=True,
log_level="WARNING",
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "WARNING"
def test_default_no_flags_gives_info(self):
"""When neither DEBUG nor LOG_LEVEL is set, effective level is INFO."""
env = os.environ.copy()
env.pop("LOG_LEVEL", None)
env.pop("DEBUG", None)
with patch.dict(os.environ, env, clear=True):
s = Settings(
database_url="sqlite:///test.db",
redis_url="redis://localhost:6379",
openai_api_key="test",
azure_ai_key="test",
azure_region="test",
azure_endpoint="https://test.example.com",
gotenberg_url="http://localhost:3000",
workdir="/tmp",
auth_enabled=False,
session_secret=None,
)
explicit = os.environ.get("LOG_LEVEL")
if s.debug and explicit is None:
effective = "DEBUG"
else:
effective = s.log_level.upper()
assert effective == "INFO"
def test_effective_level_maps_to_logging_constant(self):
"""The effective level string maps to a valid logging constant."""
for level_name in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
assert getattr(logging, level_name) is not None
@pytest.mark.unit
class TestLoggingConfiguredAtStartup:
"""Tests that the main module configures the root logger on import."""
def test_root_logger_has_handler(self):
"""Root logger should have at least one handler after app import."""
root = logging.getLogger()
assert len(root.handlers) > 0, "Root logger has no handlers after app startup"
def test_root_logger_level_is_not_warning_default(self):
"""Root logger should not be at the unconfigured WARNING default.
Our basicConfig(force=True) should have set it to at least INFO.
"""
root = logging.getLogger()
# The test env doesn't set DEBUG=True, so the level should be INFO (20)
assert root.level <= logging.INFO
@pytest.mark.unit
class TestJsonFormatter:
"""Tests for the _JsonFormatter used when LOG_FORMAT=json."""
def _make_formatter(self):
"""Lazily import the JSON formatter from main module."""
from app.main import _JsonFormatter
return _JsonFormatter()
def test_output_is_valid_json(self):
"""JSON formatter output should be parseable JSON."""
import json
fmt = self._make_formatter()
record = logging.LogRecord(
name="test.logger",
level=logging.INFO,
pathname="test.py",
lineno=42,
msg="Hello %s",
args=("world",),
exc_info=None,
)
result = fmt.format(record)
parsed = json.loads(result)
assert parsed["level"] == "INFO"
assert parsed["logger"] == "test.logger"
assert parsed["message"] == "Hello world"
assert parsed["lineno"] == 42
def test_includes_timestamp_iso8601(self):
"""JSON output should contain an ISO 8601 timestamp."""
import json
fmt = self._make_formatter()
record = logging.LogRecord(
name="x",
level=logging.DEBUG,
pathname="x.py",
lineno=1,
msg="test",
args=(),
exc_info=None,
)
parsed = json.loads(fmt.format(record))
assert "timestamp" in parsed
# ISO 8601 timestamps contain "T" and "+00:00" (UTC)
assert "T" in parsed["timestamp"]
def test_includes_exc_info_when_present(self):
"""JSON output should include exc_info when an exception is logged."""
import json
fmt = self._make_formatter()
try:
raise ValueError("boom") # noqa: TRY301
except ValueError:
import sys
record = logging.LogRecord(
name="x",
level=logging.ERROR,
pathname="x.py",
lineno=1,
msg="error",
args=(),
exc_info=sys.exc_info(),
)
parsed = json.loads(fmt.format(record))
assert "exc_info" in parsed
assert "ValueError" in parsed["exc_info"]
@pytest.mark.unit
class TestLogFormatSetting:
"""Tests for the log_format and log_syslog_* config fields."""
_BASE_KWARGS = {
"database_url": "sqlite:///test.db",
"redis_url": "redis://localhost:6379",
"openai_api_key": "test",
"azure_ai_key": "test",
"azure_region": "test",
"azure_endpoint": "https://test.example.com",
"gotenberg_url": "http://localhost:3000",
"workdir": "/tmp",
"auth_enabled": False,
"session_secret": None,
}
def test_log_format_default_is_text(self):
"""Test that log_format defaults to 'text'."""
config = Settings(**self._BASE_KWARGS)
assert config.log_format == "text"
def test_log_format_accepts_json(self):
"""Test that log_format accepts 'json'."""
config = Settings(**self._BASE_KWARGS, log_format="json")
assert config.log_format == "json"
def test_log_syslog_defaults(self):
"""Test syslog forwarding defaults."""
config = Settings(**self._BASE_KWARGS)
assert config.log_syslog_enabled is False
assert config.log_syslog_host == "localhost"
assert config.log_syslog_port == 514
assert config.log_syslog_protocol == "udp"
def test_log_syslog_can_be_enabled(self):
"""Test that syslog forwarding can be enabled."""
config = Settings(**self._BASE_KWARGS, log_syslog_enabled=True, log_syslog_host="syslog.example.com")
assert config.log_syslog_enabled is True
assert config.log_syslog_host == "syslog.example.com"
tests/test_notifications_api.py
+55
View File
@@ -830,3 +830,58 @@ class TestUserNotificationService:
result = _send_email_notification({"smtp_host": "smtp.example.com"}, "Title", "Body")
assert result is False
class TestBenchmark:
@pytest.mark.unit
def test_update_preferences_benchmark(self, notif_engine, notif_session):
import statistics
import time
from app.main import app
target = UserNotificationTarget(
owner_id=_OWNER,
channel_type="webhook",
name="My Webhook",
config=json.dumps({"url": "https://x.com"}),
)
notif_session.add(target)
notif_session.commit()
notif_session.refresh(target)
client = _make_client(notif_engine, _OWNER)
try:
items_count = 100
preferences = []
for i in range(items_count):
preferences.append(
{
"event_type": f"event.type.{i}",
"channel_type": "webhook",
"is_enabled": True,
"target_id": target.id,
}
)
payload = {"preferences": preferences}
# Warm up
client.put("/api/user-notifications/preferences", json=payload)
times = []
for _ in range(5):
# Alter the values a bit so it's a real update
for p in payload["preferences"]:
p["is_enabled"] = not p["is_enabled"]
start = time.time()
resp = client.put("/api/user-notifications/preferences", json=payload)
end = time.time()
assert resp.status_code == 200
times.append(end - start)
print(f"\nAverage time: {statistics.mean(times):.4f}s")
finally:
_cleanup(app)
tests/test_settings_audit_log.py
+31 -1
View File
@@ -289,7 +289,8 @@ class TestNotifySettingsUpdated:
call_args = mock_redis_instance.set.call_args[0]
assert call_args[0] == SETTINGS_VERSION_KEY
def test_does_not_raise_on_redis_failure(self):
@patch("app.utils.settings_sync.logger")
def test_does_not_raise_on_redis_failure(self, mock_logger):
"""notify_settings_updated must not propagate Redis errors."""
from app.utils.settings_sync import notify_settings_updated
@@ -297,6 +298,35 @@ class TestNotifySettingsUpdated:
mock_redis_module.from_url.side_effect = Exception("Redis down")
# Should not raise
notify_settings_updated()
mock_logger.warning.assert_any_call("Could not publish settings update to Redis: Redis down")
@patch("app.utils.settings_sync.logger")
def test_does_not_raise_on_reload_failure(self, mock_logger):
"""notify_settings_updated must not propagate settings reload errors."""
from app.utils.settings_sync import notify_settings_updated
with patch("app.utils.config_loader.reload_settings_from_db") as mock_reload:
mock_reload.side_effect = Exception("Reload error")
# We mock redis so that we skip over the redis block, and mock ensure_ocr_languages_async to prevent its side effects.
with patch("app.utils.settings_sync.redis"):
with patch("app.utils.ocr_language_manager.ensure_ocr_languages_async"):
# Should not raise
notify_settings_updated()
mock_logger.warning.assert_any_call("Could not reload in-process settings: Reload error")
@patch("app.utils.settings_sync.logger")
def test_does_not_raise_on_ocr_language_check_failure(self, mock_logger):
"""notify_settings_updated must not propagate OCR language check errors."""
from app.utils.settings_sync import notify_settings_updated
with patch("app.utils.ocr_language_manager.ensure_ocr_languages_async") as mock_ensure:
mock_ensure.side_effect = Exception("OCR error")
# We mock redis and reload_settings_from_db so we only test the OCR block failure.
with patch("app.utils.settings_sync.redis"):
with patch("app.utils.config_loader.reload_settings_from_db"):
# Should not raise
notify_settings_updated()
mock_logger.warning.assert_any_call("Could not schedule OCR language check: OCR error")
@pytest.mark.unit
tests/test_settings_sync.py
+211
View File
@@ -0,0 +1,211 @@
from unittest.mock import MagicMock, patch
import pytest
import app.utils.settings_sync
from app.utils.settings_sync import (
SETTINGS_VERSION_KEY,
notify_settings_updated,
register_settings_reload_signal,
)
@pytest.fixture
def reset_last_seen_version():
"""Reset the global variable before and after tests."""
app.utils.settings_sync._last_seen_version = ""
yield
app.utils.settings_sync._last_seen_version = ""
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
@patch("app.utils.settings_sync.time.time", return_value=12345.0)
def test_notify_settings_updated_success(mock_time, mock_ensure_ocr, mock_reload, mock_redis):
# Setup mock redis instance
mock_redis_instance = MagicMock()
mock_redis.return_value = mock_redis_instance
notify_settings_updated()
# Verify redis calls
mock_redis.assert_called_once()
mock_redis_instance.set.assert_called_once_with(SETTINGS_VERSION_KEY, "12345.0")
# Verify other calls
mock_reload.assert_called_once()
mock_ensure_ocr.assert_called_once()
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_notify_settings_updated_redis_failure(mock_ensure_ocr, mock_reload, mock_redis, caplog):
# Setup mock redis to fail
mock_redis.side_effect = Exception("Redis connection failed")
notify_settings_updated()
# Verification: should continue and call reload and ocr despite redis failure
mock_reload.assert_called_once()
mock_ensure_ocr.assert_called_once()
assert "Could not publish settings update to Redis: Redis connection failed" in caplog.text
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_notify_settings_updated_reload_failure(mock_ensure_ocr, mock_reload, mock_redis, caplog):
# Setup reload to fail
mock_reload.side_effect = Exception("Reload failed")
mock_redis_instance = MagicMock()
mock_redis.return_value = mock_redis_instance
notify_settings_updated()
# Verification: redis should be called, reload fails, ocr should still be called
mock_redis_instance.set.assert_called_once()
mock_ensure_ocr.assert_called_once()
assert "Could not reload in-process settings: Reload failed" in caplog.text
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_notify_settings_updated_ocr_failure(mock_ensure_ocr, mock_reload, mock_redis, caplog):
# Setup ocr check to fail
mock_ensure_ocr.side_effect = Exception("OCR check failed")
mock_redis_instance = MagicMock()
mock_redis.return_value = mock_redis_instance
notify_settings_updated()
# Verification: all should be called, ocr failure logged
mock_redis_instance.set.assert_called_once()
mock_reload.assert_called_once()
assert "Could not schedule OCR language check: OCR check failed" in caplog.text
@patch("app.utils.settings_sync.task_prerun.connect")
def test_register_settings_reload_signal(mock_connect):
register_settings_reload_signal()
# It should register a signal with task_prerun
mock_connect.assert_called_once_with(weak=False)
@patch("app.utils.settings_sync.task_prerun.connect")
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_reload_if_stale_new_version(mock_ensure_ocr, mock_reload, mock_redis, mock_connect, reset_last_seen_version):
# Capture the registered callback
mock_decorator = MagicMock()
mock_connect.return_value = mock_decorator
register_settings_reload_signal()
mock_connect.assert_called_once_with(weak=False)
# Get the callback function
callback = mock_decorator.call_args[0][0]
# Setup redis to return a new version
mock_redis_instance = MagicMock()
mock_redis_instance.get.return_value = b"new_version"
mock_redis.return_value = mock_redis_instance
# Initial state check
assert app.utils.settings_sync._last_seen_version == ""
# Call the callback
callback(sender="test")
# Verification
mock_redis_instance.get.assert_called_once_with(SETTINGS_VERSION_KEY)
mock_reload.assert_called_once()
mock_ensure_ocr.assert_called_once()
assert app.utils.settings_sync._last_seen_version == "new_version"
@patch("app.utils.settings_sync.task_prerun.connect")
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_reload_if_stale_same_version(mock_ensure_ocr, mock_reload, mock_redis, mock_connect, reset_last_seen_version):
# Set initial state
app.utils.settings_sync._last_seen_version = "existing_version"
mock_decorator = MagicMock()
mock_connect.return_value = mock_decorator
register_settings_reload_signal()
callback = mock_decorator.call_args[0][0]
# Setup redis to return the SAME version
mock_redis_instance = MagicMock()
mock_redis_instance.get.return_value = b"existing_version"
mock_redis.return_value = mock_redis_instance
# Call the callback
callback(sender="test")
# Verification
mock_redis_instance.get.assert_called_once_with(SETTINGS_VERSION_KEY)
# Should NOT reload or check OCR
mock_reload.assert_not_called()
mock_ensure_ocr.assert_not_called()
assert app.utils.settings_sync._last_seen_version == "existing_version"
@patch("app.utils.settings_sync.task_prerun.connect")
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
def test_reload_if_stale_redis_error(mock_reload, mock_redis, mock_connect, reset_last_seen_version, caplog):
import logging
caplog.set_level(logging.DEBUG)
mock_decorator = MagicMock()
mock_connect.return_value = mock_decorator
register_settings_reload_signal()
callback = mock_decorator.call_args[0][0]
# Setup redis to fail
mock_redis.side_effect = Exception("Redis error")
# Call the callback
callback(sender="test")
# Verification
mock_reload.assert_not_called()
assert "Settings version check skipped: Redis error" in caplog.text
@patch("app.utils.settings_sync.task_prerun.connect")
@patch("app.utils.settings_sync.redis.from_url")
@patch("app.utils.config_loader.reload_settings_from_db")
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_async")
def test_reload_if_stale_ocr_error(
mock_ensure_ocr, mock_reload, mock_redis, mock_connect, reset_last_seen_version, caplog
):
mock_decorator = MagicMock()
mock_connect.return_value = mock_decorator
register_settings_reload_signal()
callback = mock_decorator.call_args[0][0]
# Setup redis to return a new version
mock_redis_instance = MagicMock()
mock_redis_instance.get.return_value = b"new_version"
mock_redis.return_value = mock_redis_instance
# Setup OCR check to fail
mock_ensure_ocr.side_effect = Exception("OCR error")
# Call the callback
callback(sender="test")
# Verification
mock_reload.assert_called_once()
mock_ensure_ocr.assert_called_once()
assert "Could not schedule OCR language check on worker: OCR error" in caplog.text
assert app.utils.settings_sync._last_seen_version == "new_version"
tests/test_upload_email.py
+28
View File
@@ -121,6 +121,34 @@ class TestExtractMetadataFromFile:
assert result == {}
def test_extract_metadata_from_pdf(self, tmp_path):
"""Test extracting metadata from a PDF file using pypdf when JSON is missing."""
import pypdf
file_path = tmp_path / "test.pdf"
# Create a test PDF with metadata
writer = pypdf.PdfWriter()
writer.add_blank_page(width=100, height=100)
writer.add_metadata(
{
"/Title": "Test Title",
"/Author": "Test Author",
"/Subject": "Test Document",
"/Keywords": "test, metadata, pypdf",
}
)
with open(file_path, "wb") as f:
writer.write(f)
result = extract_metadata_from_file(str(file_path))
# Check that the leading slash is stripped and keys/values match
assert result.get("Title") == "Test Title"
assert result.get("Author") == "Test Author"
assert result.get("Subject") == "Test Document"
assert result.get("Keywords") == "test, metadata, pypdf"
@pytest.mark.unit
class TestAttachLogo: