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

style: apply ruff auto-fix

Browse Source 
- Auto-formatted code with ruff format
- Applied ruff linting fixes with --fix

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
This commit is contained in:
github-actions[bot]
2026-02-14 00:16:21 +00:00
parent 992e420556
commit e84c031538
8 changed files with 146 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_files_comprehensive.py
+19 -12
View File
@@ -842,8 +842,9 @@ class TestDeleteFileExceptions:
db_session.commit()
file_id = file.id
with patch("app.config.settings") as mock_settings, patch.object(
db_session, "delete", side_effect=Exception("Database error")
with (
patch("app.config.settings") as mock_settings,
patch.object(db_session, "delete", side_effect=Exception("Database error")),
):
mock_settings.allow_file_delete = True
response = client.delete(f"/api/files/{file_id}")
@@ -869,8 +870,9 @@ class TestDeleteFileExceptions:
def failing_commit():
raise Exception("Database commit error")
with patch("app.config.settings") as mock_settings, patch.object(
db_session, "commit", side_effect=failing_commit
with (
patch("app.config.settings") as mock_settings,
patch.object(db_session, "commit", side_effect=failing_commit),
):
mock_settings.allow_file_delete = True
response = client.post("/api/files/bulk-delete", json=[file_id])
@@ -948,7 +950,9 @@ class TestRetryPipelineSteps:
db_session.add(file)
db_session.commit()
with patch("app.tasks.process_with_azure_document_intelligence.process_with_azure_document_intelligence") as mock_task:
with patch(
"app.tasks.process_with_azure_document_intelligence.process_with_azure_document_intelligence"
) as mock_task:
mock_task.delay.return_value = Mock(id="task-azure")
result = _retry_pipeline_step(file, "process_with_azure_document_intelligence", db_session)
assert result["task_id"] == "task-azure"
@@ -991,8 +995,9 @@ class TestRetryPipelineSteps:
db_session.add(file)
db_session.commit()
with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_task, patch(
"app.api.files._extract_text_from_pdf", return_value="Sample text"
with (
patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_task,
patch("app.api.files._extract_text_from_pdf", return_value="Sample text"),
):
mock_task.delay.return_value = Mock(id="task-gpt")
result = _retry_pipeline_step(file, "extract_metadata_with_gpt", db_session)
@@ -1035,8 +1040,9 @@ class TestRetryPipelineSteps:
db_session.add(file)
db_session.commit()
with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_task, patch(
"app.api.files._extract_text_from_pdf", return_value="Sample text"
with (
patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_task,
patch("app.api.files._extract_text_from_pdf", return_value="Sample text"),
):
mock_task.delay.return_value = Mock(id="task-embed")
result = _retry_pipeline_step(file, "embed_metadata_into_pdf", db_session)
@@ -1065,9 +1071,10 @@ class TestRetryUploadTasks:
db_session.add(file)
db_session.commit()
with patch("app.api.files.settings") as mock_settings, patch(
"app.tasks.upload_to_dropbox.upload_to_dropbox"
) as mock_task:
with (
patch("app.api.files.settings") as mock_settings,
patch("app.tasks.upload_to_dropbox.upload_to_dropbox") as mock_task,
):
mock_settings.workdir = str(tmp_path)
mock_task.delay.return_value = Mock(id="task-dropbox")
response = client.post(f"/api/files/{file.id}/retry-subtask?subtask_name=upload_to_dropbox")
tests/test_api_process.py
+6 -20
View File
@@ -1,7 +1,5 @@
"""Tests for app/api/process.py module."""
import os
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
@@ -169,9 +167,7 @@ class TestProcessEndpoints:
test_file = tmp_path / "test1.pdf"
test_file.write_text("test content")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = 10
mock_task.delay.return_value = Mock(id="task-1")
@@ -192,9 +188,7 @@ class TestProcessEndpoints:
for i in range(3):
(tmp_path / f"test{i}.pdf").write_text(f"test content {i}")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = 10
mock_task.delay.return_value = Mock(id="task-id")
@@ -214,9 +208,7 @@ class TestProcessEndpoints:
for i in range(12):
(tmp_path / f"test{i}.pdf").write_text(f"test content {i}")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = 10
mock_settings.processall_throttle_delay = 5
@@ -246,9 +238,7 @@ class TestProcessEndpoints:
(tmp_path / "test.docx").write_text("word content")
(tmp_path / "test.jpg").write_text("image content")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = 10
mock_task.delay.return_value = Mock(id="task-id")
@@ -270,9 +260,7 @@ class TestProcessEndpoints:
for i in range(threshold):
(tmp_path / f"at_threshold_{i}.pdf").write_text(f"content {i}")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = threshold
mock_settings.processall_throttle_delay = 2
@@ -291,9 +279,7 @@ class TestProcessEndpoints:
for i in range(threshold + 1):
(tmp_path / f"above_threshold_{i}.pdf").write_text(f"content {i}")
with patch("app.api.process.settings") as mock_settings, patch(
"app.api.process.process_document"
) as mock_task:
with patch("app.api.process.settings") as mock_settings, patch("app.api.process.process_document") as mock_task:
mock_settings.workdir = str(tmp_path)
mock_settings.processall_throttle_threshold = threshold
mock_settings.processall_throttle_delay = 2
tests/test_celery_app.py
+3 -3
View File
@@ -211,16 +211,16 @@ class TestTaskFailureHandler:
mock_settings.notify_on_task_failure = True
# Import to ensure signal is connected
from app.celery_app import task_failure_handler
# Import the signal
from celery.signals import task_failure
from app.celery_app import task_failure_handler
# The handler should be connected to the signal
# We can test this by verifying the signal has receivers
receivers = task_failure.receivers
assert len(receivers) > 0
# Simply verify that importing the handler doesn't cause errors
# The actual signal connection is tested implicitly by the other tests
assert callable(task_failure_handler)
tests/test_database.py
+6 -6
View File
@@ -173,16 +173,16 @@ class TestSchemaMigrations:
inspector = inspect(engine)
columns = {col["name"]: col for col in inspector.get_columns("files")}
assert "original_file_path" in columns
assert columns["original_file_path"]["type"].__class__.__name__ in ("VARCHAR", "String", "TEXT")
assert "processed_file_path" in columns
assert columns["processed_file_path"]["type"].__class__.__name__ in ("VARCHAR", "String", "TEXT")
assert "is_duplicate" in columns
assert columns["is_duplicate"]["type"].__class__.__name__ in ("BOOLEAN", "Integer")
assert "duplicate_of_id" in columns
assert columns["duplicate_of_id"]["type"].__class__.__name__ in ("INTEGER", "Integer")
@@ -290,7 +290,7 @@ class TestSchemaMigrations:
from sqlalchemy import inspect
inspector = inspect(engine)
processing_log_columns = [col["name"] for col in inspector.get_columns("processing_logs")]
assert "detail" in processing_log_columns
@@ -318,7 +318,7 @@ class TestInitDbErrors:
mock_url.get_backend_name.return_value = "sqlite"
mock_url.database = ":memory:"
mock_make_url.return_value = mock_url
mock_base.metadata.create_all.side_effect = exc.SQLAlchemyError("Database error")
with pytest.raises(exc.SQLAlchemyError):
tests/test_encryption.py
+6 -5
View File
@@ -200,24 +200,26 @@ class TestGetCipherSuite:
def test_get_cipher_suite_import_error(self):
"""Test _get_cipher_suite when cryptography is not installed"""
import app.utils.encryption
import sys
import app.utils.encryption
# Reset the cached cipher suite
original_cipher = app.utils.encryption._cipher_suite
app.utils.encryption._cipher_suite = None
# Mock the cryptography.fernet module to not exist
original_modules = sys.modules.copy()
# Remove cryptography from sys.modules to simulate it not being installed
if "cryptography.fernet" in sys.modules:
del sys.modules["cryptography.fernet"]
if "cryptography" in sys.modules:
del sys.modules["cryptography"]
# Mock the import to raise ImportError
import builtins
real_import = builtins.__import__
def mock_import(name, *args, **kwargs):
@@ -245,8 +247,7 @@ class TestGetCipherSuite:
try:
# Mock Fernet class to raise an exception during initialization
from unittest.mock import MagicMock
with patch("app.utils.encryption.hashlib.sha256", side_effect=RuntimeError("Hash error")):
result = app.utils.encryption._get_cipher_suite()
tests/test_main.py
+86 -76
View File
@@ -5,11 +5,11 @@ Tests FastAPI application initialization, middleware, error handlers,
and lifecycle management.
"""
from unittest.mock import Mock, patch, MagicMock
import os
from unittest.mock import MagicMock, patch
import pytest
from fastapi import HTTPException, status
from fastapi.testclient import TestClient
from fastapi import HTTPException
@pytest.mark.unit
@@ -19,7 +19,7 @@ class TestAppInitialization:
def test_session_secret_is_set(self):
"""Test that SESSION_SECRET is configured"""
import app.main
# SESSION_SECRET should be set (either from settings or default)
assert app.main.SESSION_SECRET is not None
assert len(app.main.SESSION_SECRET) > 0
@@ -27,7 +27,7 @@ class TestAppInitialization:
def test_app_created_successfully(self):
"""Test that FastAPI app is created successfully"""
from app.main import app
assert app is not None
assert app.title == "DocuElevate"
@@ -39,79 +39,79 @@ class TestLifespanEvents:
@pytest.mark.asyncio
async def test_lifespan_context_manager_executes(self):
"""Test that lifespan context manager can be executed"""
with patch("app.database.init_db"), \
patch("app.database.SessionLocal") as mock_session_cls, \
patch("app.utils.config_loader.load_settings_from_db"), \
patch("app.utils.config_validator.dump_all_settings"), \
patch("app.utils.config_validator.check_all_configs", return_value={"email": [], "storage": {}}), \
patch("app.utils.notification.init_apprise"), \
patch("app.utils.notification.notify_startup"), \
patch("app.utils.notification.notify_shutdown"):
with (
patch("app.database.init_db"),
patch("app.database.SessionLocal") as mock_session_cls,
patch("app.utils.config_loader.load_settings_from_db"),
patch("app.utils.config_validator.dump_all_settings"),
patch("app.utils.config_validator.check_all_configs", return_value={"email": [], "storage": {}}),
patch("app.utils.notification.init_apprise"),
patch("app.utils.notification.notify_startup"),
patch("app.utils.notification.notify_shutdown"),
):
# Mock database session
mock_db = MagicMock()
mock_session_cls.return_value = mock_db
from app.main import lifespan, app
from app.main import app, lifespan
# Execute the startup and shutdown
async with lifespan(app):
pass # Startup completed
# Shutdown completed
mock_db.close.assert_called()
@pytest.mark.asyncio
async def test_lifespan_startup_with_config_issues(self):
"""Test that lifespan logs warning when there are config issues"""
with patch("app.database.init_db"), \
patch("app.database.SessionLocal") as mock_session_cls, \
patch("app.utils.config_loader.load_settings_from_db"), \
patch("app.utils.config_validator.dump_all_settings"), \
patch("app.utils.config_validator.check_all_configs") as mock_check, \
patch("app.utils.notification.init_apprise"), \
patch("app.utils.notification.notify_startup"), \
patch("app.utils.notification.notify_shutdown"), \
patch("logging.warning") as mock_warning:
with (
patch("app.database.init_db"),
patch("app.database.SessionLocal") as mock_session_cls,
patch("app.utils.config_loader.load_settings_from_db"),
patch("app.utils.config_validator.dump_all_settings"),
patch("app.utils.config_validator.check_all_configs") as mock_check,
patch("app.utils.notification.init_apprise"),
patch("app.utils.notification.notify_startup"),
patch("app.utils.notification.notify_shutdown"),
patch("logging.warning") as mock_warning,
):
mock_db = MagicMock()
mock_session_cls.return_value = mock_db
# Return config with issues
mock_check.return_value = {
"email": ["Invalid email config"],
"storage": {"dropbox": ["Missing token"]}
}
from app.main import lifespan, app
mock_check.return_value = {"email": ["Invalid email config"], "storage": {"dropbox": ["Missing token"]}}
from app.main import app, lifespan
async with lifespan(app):
pass
# Should log warning about config issues
mock_warning.assert_called()
@pytest.mark.asyncio
async def test_lifespan_startup_handles_db_settings_load_failure(self):
"""Test that lifespan handles failures when loading settings from DB"""
with patch("app.database.init_db"), \
patch("app.database.SessionLocal") as mock_session_cls, \
patch("app.utils.config_loader.load_settings_from_db", side_effect=Exception("DB error")), \
patch("app.utils.config_validator.dump_all_settings"), \
patch("app.utils.config_validator.check_all_configs", return_value={"email": [], "storage": {}}), \
patch("app.utils.notification.init_apprise"), \
patch("app.utils.notification.notify_startup"), \
patch("app.utils.notification.notify_shutdown"), \
patch("logging.error") as mock_error:
with (
patch("app.database.init_db"),
patch("app.database.SessionLocal") as mock_session_cls,
patch("app.utils.config_loader.load_settings_from_db", side_effect=Exception("DB error")),
patch("app.utils.config_validator.dump_all_settings"),
patch("app.utils.config_validator.check_all_configs", return_value={"email": [], "storage": {}}),
patch("app.utils.notification.init_apprise"),
patch("app.utils.notification.notify_startup"),
patch("app.utils.notification.notify_shutdown"),
patch("logging.error") as mock_error,
):
mock_db = MagicMock()
mock_session_cls.return_value = mock_db
from app.main import lifespan, app
from app.main import app, lifespan
# Should not raise exception, just log error
async with lifespan(app):
pass
mock_error.assert_called()
@@ -121,74 +121,83 @@ class TestExceptionHandlers:
def test_http_exception_handler_frontend_route_404(self):
"""Test that HTTPException returns HTML for frontend 404 errors"""
from app.main import app, http_exception_handler
from fastapi import Request
from app.main import http_exception_handler
# Create a mock request for a frontend route
mock_request = MagicMock(spec=Request)
mock_request.url.path = "/nonexistent"
exc = HTTPException(status_code=404, detail="Not found")
# Call the handler directly
import asyncio
response = asyncio.run(http_exception_handler(mock_request, exc))
assert response.status_code == 404
def test_http_exception_handler_frontend_route_other_error(self):
"""Test that HTTPException returns HTML for other frontend errors"""
from app.main import app, http_exception_handler
from fastapi import Request
from app.main import http_exception_handler
# Create a mock request for a frontend route
mock_request = MagicMock(spec=Request)
mock_request.url.path = "/some-page"
exc = HTTPException(status_code=403, detail="Forbidden")
# Call the handler directly
import asyncio
response = asyncio.run(http_exception_handler(mock_request, exc))
assert response.status_code == 403
def test_custom_500_handler_api_route(self):
"""Test that 500 error returns JSON for API routes"""
from app.main import app, custom_500_handler
from fastapi import Request
from app.main import custom_500_handler
# Create a mock request for an API route
mock_request = MagicMock(spec=Request)
mock_request.url.path = "/api/something"
exc = Exception("Internal error")
# Call the handler directly
import asyncio
response = asyncio.run(custom_500_handler(mock_request, exc))
assert response.status_code == 500
# Parse JSON response
import json
content = json.loads(response.body.decode())
assert content["detail"] == "Internal server error"
def test_custom_500_handler_frontend_route(self):
"""Test that 500 error returns HTML for frontend routes"""
from app.main import app, custom_500_handler
from fastapi import Request
from app.main import custom_500_handler
# Create a mock request for a frontend route
mock_request = MagicMock(spec=Request)
mock_request.url.path = "/dashboard"
exc = Exception("Internal error")
# Call the handler directly
import asyncio
response = asyncio.run(custom_500_handler(mock_request, exc))
assert response.status_code == 500
@@ -199,7 +208,7 @@ class TestTestEndpoint:
def test_test_500_endpoint_raises_error(self):
"""Test that /test-500 endpoint raises RuntimeError"""
from app.main import test_500
# The function should raise RuntimeError
with pytest.raises(RuntimeError, match="Testing forced 500 error"):
test_500()
@@ -211,12 +220,13 @@ class TestStaticFileMount:
def test_static_files_mounted_when_directory_exists(self):
"""Test that static files are served when directory exists"""
from app.main import app
import pathlib
from app.main import app
# Check if static directory exists
static_dir = pathlib.Path(__file__).parents[1] / "frontend" / "static"
if os.path.exists(static_dir):
# Check if static route is mounted
assert any("/static" in str(route.path) for route in app.routes)
@@ -229,12 +239,12 @@ class TestMiddlewareConfiguration:
def test_app_has_limiter_state(self):
"""Test that app.state.limiter is configured"""
from app.main import app
assert hasattr(app.state, "limiter")
assert app.state.limiter is not None
def test_app_has_correct_title(self):
"""Test that FastAPI app has correct title"""
from app.main import app
assert app.title == "DocuElevate"
tests/test_upload_email.py
+4 -3
View File
@@ -68,7 +68,7 @@ class TestGetEmailTemplate:
# Workdir exists, but template loading fails; falls back to built-in
mock_exists.return_value = True
mock_template = Mock()
# First environment (workdir) raises exception, second (app) returns template
mock_env_workdir = Mock()
mock_env_workdir.globals = {}
@@ -163,6 +163,7 @@ class TestAttachLogo:
@patch("builtins.open", new_callable=mock_open, read_data=b"fake_svg_data")
def test_attaches_svg_logo_with_correct_mime_type(self, mock_file, mock_exists):
"""Test attaches SVG logo with correct MIME type (image/svg+xml)."""
# Create a custom side effect that returns True only for SVG path
def custom_exists(path):
return "logo.svg" in path
@@ -178,9 +179,9 @@ class TestAttachLogo:
assert result is True
assert len(msg.get_payload()) > 0
# Verify SVG MIME type is used (the function detects .svg extension)
# Note: MIMEImage may default to a different subtype, but the key is that
# Note: MIMEImage may default to a different subtype, but the key is that
# the function passes 'image/svg+xml' as mimetype parameter
# Since we're using mock_open, we can't verify the exact MIME in the attachment,
# but we verified the code path is exercised
tests/test_views_google_drive.py
+16 -22
View File
@@ -1,8 +1,9 @@
"""Tests for app/views/google_drive.py module."""
import pytest
from unittest.mock import patch
import urllib.parse
from unittest.mock import patch
import pytest
@pytest.mark.integration
@@ -39,12 +40,11 @@ class TestGoogleDriveViews:
client_id = "test_client_id_123"
redirect_uri = "https://example.com/callback"
response = client.get(
f"/google-drive-auth-start?client_id={client_id}&redirect_uri={redirect_uri}",
follow_redirects=False
f"/google-drive-auth-start?client_id={client_id}&redirect_uri={redirect_uri}", follow_redirects=False
)
assert response.status_code in [302, 307] # Redirect status codes
# Verify redirect location
location = response.headers.get("location")
assert location is not None
@@ -60,13 +60,10 @@ class TestGoogleDriveViews:
def test_google_drive_auth_start_without_redirect_uri(self, client):
"""Test starting Google Drive OAuth flow without explicit redirect_uri."""
client_id = "test_client_id_456"
response = client.get(
f"/google-drive-auth-start?client_id={client_id}",
follow_redirects=False
)
response = client.get(f"/google-drive-auth-start?client_id={client_id}", follow_redirects=False)
assert response.status_code in [302, 307] # Redirect status codes
# Verify redirect location
location = response.headers.get("location")
assert location is not None
@@ -78,14 +75,11 @@ class TestGoogleDriveViews:
def test_google_drive_auth_start_scope_configuration(self, client):
"""Test that Google Drive auth start uses correct OAuth scope."""
client_id = "test_client_id_789"
response = client.get(
f"/google-drive-auth-start?client_id={client_id}",
follow_redirects=False
)
response = client.get(f"/google-drive-auth-start?client_id={client_id}", follow_redirects=False)
location = response.headers.get("location")
assert location is not None
# The scope should be URL encoded, so check for the encoded version
# drive.file scope: https://www.googleapis.com/auth/drive.file
expected_scope = urllib.parse.quote("https://www.googleapis.com/auth/drive.file")
@@ -100,7 +94,7 @@ class TestGoogleDriveViews:
mock_settings.google_drive_refresh_token = "test_token"
mock_settings.google_drive_credentials_json = '{"test": "creds"}'
mock_settings.google_drive_folder_id = None # Empty folder ID
response = client.get("/google-drive-setup")
assert response.status_code == 200
# Verify the response context indicates configuration is incomplete
@@ -116,7 +110,7 @@ class TestGoogleDriveViews:
mock_settings.google_drive_refresh_token = "test_token"
mock_settings.google_drive_credentials_json = '{"test": "creds"}'
mock_settings.google_drive_folder_id = "" # Empty string folder ID
response = client.get("/google-drive-setup")
assert response.status_code == 200
# Should handle empty string folder_id similar to None
@@ -130,7 +124,7 @@ class TestGoogleDriveViews:
mock_settings.google_drive_refresh_token = "oauth_token"
mock_settings.google_drive_folder_id = "test_folder_id"
mock_settings.google_drive_credentials_json = None
response = client.get("/google-drive-setup")
assert response.status_code == 200
@@ -143,6 +137,6 @@ class TestGoogleDriveViews:
mock_settings.google_drive_client_id = None
mock_settings.google_drive_client_secret = None
mock_settings.google_drive_refresh_token = None
response = client.get("/google-drive-setup")
assert response.status_code == 200