feat: add comprehensive unit tests for tasks and auth modules
- Add test_extract_metadata_gpt.py with 16 tests (7 passing) - Add test_convert_pdf.py with 13 tests achieving 90.56% coverage - Add test_embed_pdf_metadata.py with 10 tests achieving 46.81% coverage - Add test_finalize_storage.py with 9 tests - Add test_auth_module.py with 14 tests achieving 43.48% coverage Coverage improvements: - convert_to_pdf: 10% → 90.56% - embed_metadata_into_pdf: 14.89% → 46.81% - extract_metadata_with_gpt: 21.84% → 36.78% - auth.py: 27.83% → 43.48% 43 tests passing, working on fixing remaining tests for full coverage Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
+1
-1
File diff suppressed because one or more lines are too long
@@ -0,0 +1,511 @@
|
||||
"""Comprehensive unit tests for app/auth.py module."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import Request, status
|
||||
from starlette.responses import RedirectResponse
|
||||
|
||||
from app.auth import get_current_user, get_gravatar_url, require_login
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetCurrentUser:
|
||||
"""Tests for get_current_user function."""
|
||||
|
||||
def test_returns_user_from_session(self):
|
||||
"""Test returns user data from request session."""
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {"user": {"id": "123", "name": "John Doe", "email": "john@example.com"}}
|
||||
|
||||
result = get_current_user(mock_request)
|
||||
|
||||
assert result == {"id": "123", "name": "John Doe", "email": "john@example.com"}
|
||||
|
||||
def test_returns_none_when_no_user_in_session(self):
|
||||
"""Test returns None when no user in session."""
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
|
||||
result = get_current_user(mock_request)
|
||||
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetGravatarUrl:
|
||||
"""Tests for get_gravatar_url function."""
|
||||
|
||||
def test_generates_gravatar_url(self):
|
||||
"""Test generates correct Gravatar URL."""
|
||||
email = "test@example.com"
|
||||
result = get_gravatar_url(email)
|
||||
|
||||
assert result.startswith("https://www.gravatar.com/avatar/")
|
||||
assert "?d=identicon" in result
|
||||
|
||||
def test_handles_uppercase_email(self):
|
||||
"""Test handles uppercase email correctly."""
|
||||
email1 = "Test@Example.COM"
|
||||
email2 = "test@example.com"
|
||||
|
||||
result1 = get_gravatar_url(email1)
|
||||
result2 = get_gravatar_url(email2)
|
||||
|
||||
# Should produce the same hash for case-insensitive emails
|
||||
assert result1 == result2
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
"""Test strips whitespace from email."""
|
||||
email1 = " test@example.com "
|
||||
email2 = "test@example.com"
|
||||
|
||||
result1 = get_gravatar_url(email1)
|
||||
result2 = get_gravatar_url(email2)
|
||||
|
||||
assert result1 == result2
|
||||
|
||||
def test_hash_is_md5(self):
|
||||
"""Test that the hash is MD5 (32 hexadecimal characters)."""
|
||||
email = "test@example.com"
|
||||
result = get_gravatar_url(email)
|
||||
|
||||
# Extract hash from URL
|
||||
hash_part = result.split("/avatar/")[1].split("?")[0]
|
||||
assert len(hash_part) == 32
|
||||
assert all(c in "0123456789abcdef" for c in hash_part)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRequireLogin:
|
||||
"""Tests for require_login decorator."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allows_access_when_auth_disabled(self):
|
||||
"""Test allows access when AUTH_ENABLED is False."""
|
||||
with patch("app.auth.AUTH_ENABLED", False):
|
||||
|
||||
@require_login
|
||||
async def test_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
|
||||
result = await test_endpoint(mock_request)
|
||||
|
||||
assert result == {"message": "success"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_allows_access_when_user_logged_in(self):
|
||||
"""Test allows access when user is logged in."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
|
||||
@require_login
|
||||
async def test_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {"user": {"id": "123", "name": "John"}}
|
||||
mock_request.url = "http://localhost/test"
|
||||
|
||||
result = await test_endpoint(mock_request)
|
||||
|
||||
assert result == {"message": "success"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redirects_to_login_when_not_authenticated(self):
|
||||
"""Test redirects to login when user is not authenticated."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
|
||||
@require_login
|
||||
async def test_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_request.url = "http://localhost/protected"
|
||||
|
||||
result = await test_endpoint(mock_request)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.status_code == status.HTTP_302_FOUND
|
||||
assert "/login" in result.headers["location"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_saves_redirect_url_in_session(self):
|
||||
"""Test saves redirect URL in session before redirecting to login."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
|
||||
@require_login
|
||||
async def test_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_request.url = "http://localhost/protected/page"
|
||||
|
||||
result = await test_endpoint(mock_request)
|
||||
|
||||
assert "redirect_after_login" in mock_request.session
|
||||
assert mock_request.session["redirect_after_login"] == "http://localhost/protected/page"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_works_with_sync_functions(self):
|
||||
"""Test decorator works with synchronous functions."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
|
||||
@require_login
|
||||
def test_sync_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {"user": {"id": "123"}}
|
||||
mock_request.url = "http://localhost/test"
|
||||
|
||||
result = test_sync_endpoint(mock_request)
|
||||
|
||||
assert result == {"message": "success"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redirects_sync_function_when_not_authenticated(self):
|
||||
"""Test redirects synchronous functions when not authenticated."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
|
||||
@require_login
|
||||
def test_sync_endpoint(request: Request):
|
||||
return {"message": "success"}
|
||||
|
||||
mock_request = MagicMock(spec=Request)
|
||||
mock_request.session = {}
|
||||
mock_request.url = "http://localhost/protected"
|
||||
|
||||
result = test_sync_endpoint(mock_request)
|
||||
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.status_code == status.HTTP_302_FOUND
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOAuthConfiguration:
|
||||
"""Tests for OAuth configuration."""
|
||||
|
||||
def test_oauth_not_configured_without_credentials(self):
|
||||
"""Test OAuth is not configured when credentials are missing."""
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.auth_enabled = True
|
||||
mock_settings.authentik_client_id = None
|
||||
mock_settings.authentik_client_secret = None
|
||||
|
||||
# Re-import to trigger configuration logic
|
||||
import importlib
|
||||
|
||||
import app.auth
|
||||
|
||||
importlib.reload(app.auth)
|
||||
|
||||
from app.auth import OAUTH_CONFIGURED
|
||||
|
||||
assert OAUTH_CONFIGURED is False
|
||||
|
||||
def test_oauth_configured_with_credentials(self):
|
||||
"""Test OAuth is configured when credentials are provided."""
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
mock_settings.auth_enabled = True
|
||||
mock_settings.authentik_client_id = "test_client_id"
|
||||
mock_settings.authentik_client_secret = "test_secret"
|
||||
mock_settings.authentik_config_url = "https://auth.example.com/.well-known/openid-configuration"
|
||||
mock_settings.oauth_provider_name = "Test SSO"
|
||||
|
||||
# Re-import to trigger configuration logic
|
||||
import importlib
|
||||
|
||||
import app.auth
|
||||
|
||||
importlib.reload(app.auth)
|
||||
|
||||
from app.auth import OAUTH_CONFIGURED, OAUTH_PROVIDER_NAME
|
||||
|
||||
assert OAUTH_CONFIGURED is True
|
||||
assert OAUTH_PROVIDER_NAME == "Test SSO"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestLoginEndpoint:
|
||||
"""Tests for login endpoint (when AUTH_ENABLED)."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login_page_shows_oauth_when_configured(self):
|
||||
"""Test login page shows OAuth option when configured."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.OAUTH_CONFIGURED", True):
|
||||
with patch("app.auth.OAUTH_PROVIDER_NAME", "Test SSO"):
|
||||
with patch("app.auth.templates") as mock_templates:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.version = "1.0.0"
|
||||
|
||||
from app.auth import login
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.query_params.get.return_value = None
|
||||
|
||||
await login(mock_request)
|
||||
|
||||
# Verify template was rendered with OAuth enabled
|
||||
mock_templates.TemplateResponse.assert_called_once()
|
||||
call_args = mock_templates.TemplateResponse.call_args
|
||||
context = call_args[0][1]
|
||||
assert context["show_oauth"] is True
|
||||
assert context["oauth_provider_name"] == "Test SSO"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAuthEndpoint:
|
||||
"""Tests for local authentication endpoint."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_successful_authentication(self):
|
||||
"""Test successful local authentication."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_form_data = {"username": "admin", "password": "secret123"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request)
|
||||
|
||||
# Verify redirect to upload page
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert result.status_code == 302
|
||||
|
||||
# Verify user was added to session
|
||||
assert "user" in mock_request.session
|
||||
assert mock_request.session["user"]["id"] == "admin"
|
||||
assert mock_request.session["user"]["is_admin"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_authentication(self):
|
||||
"""Test failed authentication with wrong credentials."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_form_data = {"username": "admin", "password": "wrong_password"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {}
|
||||
|
||||
result = await auth(mock_request)
|
||||
|
||||
# Verify redirect to login with error
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "error=Invalid" in result.headers["location"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_authentication_with_redirect_after_login(self):
|
||||
"""Test authentication redirects to saved URL after login."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "secret123"
|
||||
|
||||
from app.auth import auth
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_form_data = {"username": "admin", "password": "secret123"}
|
||||
mock_request.form = AsyncMock(return_value=mock_form_data)
|
||||
mock_request.session = {"redirect_after_login": "/protected/page"}
|
||||
|
||||
result = await auth(mock_request)
|
||||
|
||||
# Verify redirect to saved URL
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "/protected/page" in result.headers["location"]
|
||||
|
||||
# Verify redirect_after_login was removed from session
|
||||
assert "redirect_after_login" not in mock_request.session
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOAuthCallback:
|
||||
"""Tests for OAuth callback endpoint."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_oauth_callback_not_available_when_not_configured(self):
|
||||
"""Test OAuth callback returns error when OAuth not configured."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.OAUTH_CONFIGURED", False):
|
||||
from app.auth import oauth_callback, oauth_login
|
||||
|
||||
mock_request = MagicMock()
|
||||
|
||||
result = await oauth_login(mock_request)
|
||||
|
||||
# Should redirect to login with error
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "error=OAuth+not+configured" in result.headers["location"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_oauth_callback_successful_authentication(self):
|
||||
"""Test OAuth callback with successful authentication."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.OAUTH_CONFIGURED", True):
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
# Mock OAuth token response
|
||||
mock_token = {
|
||||
"userinfo": {
|
||||
"sub": "user123",
|
||||
"name": "John Doe",
|
||||
"email": "john@example.com",
|
||||
"preferred_username": "johndoe",
|
||||
"groups": ["admin", "users"],
|
||||
}
|
||||
}
|
||||
mock_oauth.authentik.authorize_access_token = AsyncMock(return_value=mock_token)
|
||||
|
||||
from app.auth import oauth_callback
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
|
||||
# Verify user was added to session
|
||||
assert "user" in mock_request.session
|
||||
assert mock_request.session["user"]["email"] == "john@example.com"
|
||||
assert mock_request.session["user"]["is_admin"] is True
|
||||
|
||||
# Verify redirect
|
||||
assert isinstance(result, RedirectResponse)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_oauth_callback_non_admin_user(self):
|
||||
"""Test OAuth callback for non-admin user."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.OAUTH_CONFIGURED", True):
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
# Mock OAuth token response without admin group
|
||||
mock_token = {
|
||||
"userinfo": {
|
||||
"sub": "user456",
|
||||
"name": "Jane Doe",
|
||||
"email": "jane@example.com",
|
||||
"preferred_username": "janedoe",
|
||||
"groups": ["users"], # Not admin
|
||||
}
|
||||
}
|
||||
mock_oauth.authentik.authorize_access_token = AsyncMock(return_value=mock_token)
|
||||
|
||||
from app.auth import oauth_callback
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
|
||||
# Verify user is not admin
|
||||
assert mock_request.session["user"]["is_admin"] is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_oauth_callback_adds_gravatar_when_no_picture(self):
|
||||
"""Test OAuth callback adds Gravatar when no picture provided."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
with patch("app.auth.OAUTH_CONFIGURED", True):
|
||||
with patch("app.auth.oauth") as mock_oauth:
|
||||
with patch("app.auth.settings") as mock_settings:
|
||||
mock_settings.admin_group_name = "admin"
|
||||
|
||||
# Mock OAuth token response without picture
|
||||
mock_token = {
|
||||
"userinfo": {
|
||||
"sub": "user789",
|
||||
"name": "Bob Smith",
|
||||
"email": "bob@example.com",
|
||||
"preferred_username": "bobsmith",
|
||||
# No picture field
|
||||
}
|
||||
}
|
||||
mock_oauth.authentik.authorize_access_token = AsyncMock(return_value=mock_token)
|
||||
|
||||
from app.auth import oauth_callback
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
|
||||
result = await oauth_callback(mock_request)
|
||||
|
||||
# Verify Gravatar was added
|
||||
assert "picture" in mock_request.session["user"]
|
||||
assert "gravatar.com/avatar/" in mock_request.session["user"]["picture"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestLogoutEndpoint:
|
||||
"""Tests for logout endpoint."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_logout_clears_session(self):
|
||||
"""Test logout clears user from session."""
|
||||
with patch("app.auth.AUTH_ENABLED", True):
|
||||
from app.auth import logout
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {"user": {"id": "123", "name": "John"}}
|
||||
|
||||
result = await logout(mock_request)
|
||||
|
||||
# Verify user was removed from session
|
||||
assert "user" not in mock_request.session
|
||||
|
||||
# Verify redirect to login with message
|
||||
assert isinstance(result, RedirectResponse)
|
||||
assert "message=You+have+been+logged+out" in result.headers["location"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestWhoAmIEndpoint:
|
||||
"""Tests for whoami API endpoint."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_whoami_returns_user_when_authenticated(self):
|
||||
"""Test whoami returns user data when authenticated."""
|
||||
from app.auth import whoami
|
||||
|
||||
mock_request = MagicMock()
|
||||
user_data = {"id": "123", "name": "John Doe", "email": "john@example.com"}
|
||||
mock_request.session = {"user": user_data}
|
||||
|
||||
# Since require_login is applied, we need to bypass it for this test
|
||||
with patch("app.auth.AUTH_ENABLED", False):
|
||||
result = await whoami(mock_request)
|
||||
|
||||
assert result == user_data
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_whoami_returns_error_when_not_authenticated(self):
|
||||
"""Test whoami returns error when not authenticated."""
|
||||
from app.auth import whoami
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
|
||||
with patch("app.auth.AUTH_ENABLED", False):
|
||||
result = await whoami(mock_request)
|
||||
|
||||
assert result == {"error": "Not authenticated"}
|
||||
@@ -0,0 +1,417 @@
|
||||
"""Comprehensive unit tests for app/tasks/convert_to_pdf.py module."""
|
||||
|
||||
from unittest.mock import MagicMock, Mock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.convert_to_pdf import (
|
||||
_build_filename,
|
||||
_detect_extension,
|
||||
_detect_mime_type,
|
||||
_detect_mime_type_from_magic,
|
||||
convert_to_pdf,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDetectMimeTypeFromMagic:
|
||||
"""Tests for _detect_mime_type_from_magic function."""
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.puremagic.from_file")
|
||||
def test_detects_mime_from_puremagic(self, mock_puremagic):
|
||||
"""Test MIME type detection using puremagic."""
|
||||
mock_match = MagicMock()
|
||||
mock_match.mime_type = "application/pdf"
|
||||
mock_puremagic.return_value = [mock_match]
|
||||
|
||||
result = _detect_mime_type_from_magic("/test/file.pdf")
|
||||
assert result == "application/pdf"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.filetype.guess")
|
||||
@patch("app.tasks.convert_to_pdf.puremagic.from_file")
|
||||
def test_falls_back_to_filetype(self, mock_puremagic, mock_filetype):
|
||||
"""Test fallback to filetype when puremagic fails."""
|
||||
from app.tasks.convert_to_pdf import puremagic
|
||||
|
||||
mock_puremagic.side_effect = puremagic.PureError("Cannot detect")
|
||||
mock_guess = MagicMock()
|
||||
mock_guess.mime = "image/jpeg"
|
||||
mock_filetype.return_value = mock_guess
|
||||
|
||||
result = _detect_mime_type_from_magic("/test/image.jpg")
|
||||
assert result == "image/jpeg"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.filetype.guess")
|
||||
@patch("app.tasks.convert_to_pdf.puremagic.from_file")
|
||||
def test_returns_none_when_detection_fails(self, mock_puremagic, mock_filetype):
|
||||
"""Test returns None when all detection methods fail."""
|
||||
from app.tasks.convert_to_pdf import puremagic
|
||||
|
||||
mock_puremagic.side_effect = puremagic.PureError("Cannot detect")
|
||||
mock_filetype.return_value = None
|
||||
|
||||
result = _detect_mime_type_from_magic("/test/unknown")
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDetectMimeType:
|
||||
"""Tests for _detect_mime_type function."""
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_type")
|
||||
def test_detects_from_file_path(self, mock_guess_type):
|
||||
"""Test MIME type detection from file path extension."""
|
||||
mock_guess_type.return_value = ("application/pdf", None)
|
||||
|
||||
mime_type, encoding = _detect_mime_type("/test/file.pdf", None)
|
||||
assert mime_type == "application/pdf"
|
||||
assert encoding is None
|
||||
|
||||
@patch("app.tasks.convert_to_pdf._detect_mime_type_from_magic")
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_type")
|
||||
def test_uses_original_filename_when_provided(self, mock_guess_type, mock_magic):
|
||||
"""Test uses original filename for detection when provided."""
|
||||
mock_guess_type.side_effect = [(None, None), ("application/vnd.ms-excel", None)]
|
||||
|
||||
mime_type, encoding = _detect_mime_type("/tmp/uuid.bin", "report.xls")
|
||||
assert mime_type == "application/vnd.ms-excel"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf._detect_mime_type_from_magic")
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_type")
|
||||
def test_falls_back_to_magic_detection(self, mock_guess_type, mock_magic):
|
||||
"""Test fallback to magic byte detection."""
|
||||
mock_guess_type.return_value = (None, None)
|
||||
mock_magic.return_value = "image/png"
|
||||
|
||||
mime_type, encoding = _detect_mime_type("/test/file", None)
|
||||
assert mime_type == "image/png"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDetectExtension:
|
||||
"""Tests for _detect_extension function."""
|
||||
|
||||
def test_detects_extension_from_file_path(self):
|
||||
"""Test extension detection from file path."""
|
||||
result = _detect_extension("/test/file.PDF", None, None)
|
||||
assert result == ".pdf"
|
||||
|
||||
def test_uses_original_filename_extension(self):
|
||||
"""Test uses original filename when file path has no extension."""
|
||||
result = _detect_extension("/tmp/uuid", "document.docx", None)
|
||||
assert result == ".docx"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_extension")
|
||||
def test_guesses_from_mime_type(self, mock_guess_ext):
|
||||
"""Test extension guessing from MIME type."""
|
||||
mock_guess_ext.return_value = ".jpg"
|
||||
|
||||
result = _detect_extension("/test/file", None, "image/jpeg")
|
||||
assert result == ".jpg"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.filetype.guess")
|
||||
@patch("app.tasks.convert_to_pdf.puremagic.from_file")
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_extension")
|
||||
def test_uses_puremagic_fallback(self, mock_guess_ext, mock_puremagic, mock_filetype):
|
||||
"""Test uses puremagic as fallback for extension detection."""
|
||||
mock_guess_ext.return_value = None
|
||||
mock_match = MagicMock()
|
||||
mock_match.extension = ".png"
|
||||
mock_puremagic.return_value = [mock_match]
|
||||
|
||||
result = _detect_extension("/test/file", None, None)
|
||||
assert result == ".png"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.filetype.guess")
|
||||
@patch("app.tasks.convert_to_pdf.puremagic.from_file")
|
||||
@patch("app.tasks.convert_to_pdf.mimetypes.guess_extension")
|
||||
def test_returns_empty_when_all_fail(self, mock_guess_ext, mock_puremagic, mock_filetype):
|
||||
"""Test returns empty string when all detection fails."""
|
||||
from app.tasks.convert_to_pdf import puremagic
|
||||
|
||||
mock_guess_ext.return_value = None
|
||||
mock_puremagic.side_effect = puremagic.PureError("Cannot detect")
|
||||
mock_filetype.return_value = None
|
||||
|
||||
result = _detect_extension("/test/file", None, None)
|
||||
assert result == ""
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestBuildFilename:
|
||||
"""Tests for _build_filename function."""
|
||||
|
||||
def test_uses_original_filename_with_extension(self):
|
||||
"""Test uses original filename when it has an extension."""
|
||||
result = _build_filename("/tmp/uuid", "document.pdf", ".pdf")
|
||||
assert result == "document.pdf"
|
||||
|
||||
def test_appends_extension_to_basename(self):
|
||||
"""Test appends extension when needed."""
|
||||
result = _build_filename("/tmp/file", None, ".pdf")
|
||||
assert result == "file.pdf"
|
||||
|
||||
def test_does_not_duplicate_extension(self):
|
||||
"""Test does not duplicate extension."""
|
||||
result = _build_filename("/tmp/file.pdf", None, ".pdf")
|
||||
assert result == "file.pdf"
|
||||
|
||||
def test_returns_basename_when_no_extension(self):
|
||||
"""Test returns basename when no extension provided."""
|
||||
result = _build_filename("/tmp/file", None, "")
|
||||
assert result == "file"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestConvertToPdf:
|
||||
"""Tests for convert_to_pdf Celery task."""
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.process_document")
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"file content")
|
||||
def test_converts_office_document_successfully(self, mock_file, mock_log_progress, mock_post, mock_process):
|
||||
"""Test successful conversion of Office document."""
|
||||
# Mock successful Gotenberg response
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.content = b"%PDF-1.4 converted content"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
# Mock task context
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
# Mock file type detection
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = (
|
||||
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
||||
None,
|
||||
)
|
||||
mock_detect_ext.return_value = ".docx"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/test.docx", "document.docx")
|
||||
|
||||
# Verify Gotenberg was called
|
||||
mock_post.assert_called_once()
|
||||
call_args = mock_post.call_args
|
||||
assert "libreoffice/convert" in call_args[0][0]
|
||||
|
||||
# Verify PDF was written
|
||||
write_calls = [call for call in mock_file().write.call_args_list]
|
||||
assert len(write_calls) > 0
|
||||
|
||||
# Verify process_document was queued
|
||||
mock_process.delay.assert_called_once()
|
||||
|
||||
# Verify result
|
||||
assert result == "/tmp/test.pdf"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
def test_returns_none_when_gotenberg_url_not_configured(self, mock_log_progress):
|
||||
"""Test returns None when Gotenberg URL is not configured."""
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = None
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/test.docx")
|
||||
|
||||
assert result is None
|
||||
# Verify error was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) > 0
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
def test_returns_none_when_file_type_unknown(self, mock_log_progress):
|
||||
"""Test returns None when file type cannot be determined."""
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_detect_mime.return_value = (None, None)
|
||||
mock_detect_ext.return_value = ""
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/unknown_file")
|
||||
|
||||
assert result is None
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.process_document")
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"image content")
|
||||
def test_converts_image_file(self, mock_file, mock_log_progress, mock_post, mock_process):
|
||||
"""Test conversion of image file."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.content = b"%PDF-1.4 converted image"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("image/jpeg", None)
|
||||
mock_detect_ext.return_value = ".jpg"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/photo.jpg")
|
||||
|
||||
# Verify LibreOffice endpoint was used for images
|
||||
mock_post.assert_called_once()
|
||||
call_args = mock_post.call_args
|
||||
assert "libreoffice/convert" in call_args[0][0]
|
||||
assert result == "/tmp/photo.pdf"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.process_document")
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"<html><body>Test</body></html>")
|
||||
def test_converts_html_file(self, mock_file, mock_log_progress, mock_post, mock_process):
|
||||
"""Test conversion of HTML file."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.content = b"%PDF-1.4 converted html"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("text/html", None)
|
||||
mock_detect_ext.return_value = ".html"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/page.html")
|
||||
|
||||
# Verify Chromium endpoint was used
|
||||
mock_post.assert_called_once()
|
||||
call_args = mock_post.call_args
|
||||
assert "chromium/convert/html" in call_args[0][0]
|
||||
assert result == "/tmp/page.pdf"
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"# Markdown\n\nTest content")
|
||||
def test_converts_markdown_file(self, mock_file, mock_log_progress, mock_post):
|
||||
"""Test conversion of Markdown file."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.content = b"%PDF-1.4 converted markdown"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
with patch("app.tasks.convert_to_pdf.os.path.exists") as mock_exists:
|
||||
with patch("app.tasks.convert_to_pdf.os.path.dirname") as mock_dirname:
|
||||
with patch("app.tasks.convert_to_pdf.os.remove") as mock_remove:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("text/markdown", None)
|
||||
mock_detect_ext.return_value = ".md"
|
||||
mock_exists.return_value = True
|
||||
mock_dirname.return_value = "/tmp"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/readme.md")
|
||||
|
||||
# Verify Chromium markdown endpoint was used
|
||||
mock_post.assert_called_once()
|
||||
call_args = mock_post.call_args
|
||||
assert "chromium/convert/markdown" in call_args[0][0]
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"file content")
|
||||
def test_handles_gotenberg_error(self, mock_file, mock_log_progress, mock_post):
|
||||
"""Test handling of Gotenberg API errors."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 500
|
||||
mock_response.text = "Internal Server Error"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("application/pdf", None)
|
||||
mock_detect_ext.return_value = ".pdf"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/test.pdf")
|
||||
|
||||
assert result is None
|
||||
# Verify error was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) >= 1
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"file content")
|
||||
def test_handles_network_exception(self, mock_file, mock_log_progress, mock_post):
|
||||
"""Test handling of network exceptions during conversion."""
|
||||
mock_post.side_effect = Exception("Connection timeout")
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("application/pdf", None)
|
||||
mock_detect_ext.return_value = ".pdf"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/test.pdf")
|
||||
|
||||
assert result is None
|
||||
|
||||
@patch("app.tasks.convert_to_pdf.process_document")
|
||||
@patch("app.tasks.convert_to_pdf.requests.post")
|
||||
@patch("app.tasks.convert_to_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"file content")
|
||||
def test_preserves_original_filename(self, mock_file, mock_log_progress, mock_post, mock_process):
|
||||
"""Test that original filename is preserved and passed to process_document."""
|
||||
mock_response = MagicMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.content = b"%PDF-1.4"
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
with patch("app.tasks.convert_to_pdf._detect_mime_type") as mock_detect_mime:
|
||||
with patch("app.tasks.convert_to_pdf._detect_extension") as mock_detect_ext:
|
||||
with patch("app.tasks.convert_to_pdf.settings") as mock_settings:
|
||||
mock_settings.gotenberg_url = "http://gotenberg:3000"
|
||||
mock_settings.http_request_timeout = 60
|
||||
mock_detect_mime.return_value = ("application/vnd.ms-excel", None)
|
||||
mock_detect_ext.return_value = ".xls"
|
||||
|
||||
result = convert_to_pdf.__wrapped__(mock_task, "/tmp/uuid.xls", "report.xls")
|
||||
|
||||
# Verify process_document was called with modified original filename
|
||||
mock_process.delay.assert_called_once()
|
||||
call_args = mock_process.delay.call_args
|
||||
assert call_args[1]["original_filename"] == "report.pdf"
|
||||
@@ -0,0 +1,470 @@
|
||||
"""Comprehensive unit tests for app/tasks/embed_metadata_into_pdf.py module."""
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, Mock, mock_open, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf, persist_metadata
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestPersistMetadata:
|
||||
"""Tests for persist_metadata function."""
|
||||
|
||||
@patch("builtins.open", new_callable=mock_open)
|
||||
def test_saves_metadata_to_json_file(self, mock_file):
|
||||
"""Test metadata is saved to JSON file with correct path."""
|
||||
metadata = {
|
||||
"filename": "test_document.pdf",
|
||||
"document_type": "Invoice",
|
||||
"tags": ["test", "invoice"],
|
||||
}
|
||||
|
||||
result = persist_metadata(metadata, "/workdir/processed/MyFile.pdf")
|
||||
|
||||
assert result == "/workdir/processed/MyFile.json"
|
||||
mock_file.assert_called_once_with("/workdir/processed/MyFile.json", "w", encoding="utf-8")
|
||||
|
||||
@patch("builtins.open", new_callable=mock_open)
|
||||
@patch("app.tasks.embed_metadata_into_pdf.json.dump")
|
||||
def test_augments_metadata_with_file_paths(self, mock_json_dump, mock_file):
|
||||
"""Test metadata is augmented with file path references."""
|
||||
metadata = {"filename": "test.pdf"}
|
||||
|
||||
persist_metadata(
|
||||
metadata,
|
||||
"/workdir/processed/test.pdf",
|
||||
original_file_path="/workdir/original/file.pdf",
|
||||
processed_file_path="/workdir/processed/test.pdf",
|
||||
)
|
||||
|
||||
# Verify json.dump was called with augmented metadata
|
||||
call_args = mock_json_dump.call_args
|
||||
augmented_metadata = call_args[0][0]
|
||||
assert augmented_metadata["original_file_path"] == "/workdir/original/file.pdf"
|
||||
assert augmented_metadata["processed_file_path"] == "/workdir/processed/test.pdf"
|
||||
assert augmented_metadata["filename"] == "test.pdf"
|
||||
|
||||
@patch("builtins.open", new_callable=mock_open)
|
||||
@patch("app.tasks.embed_metadata_into_pdf.json.dump")
|
||||
def test_handles_metadata_without_optional_paths(self, mock_json_dump, mock_file):
|
||||
"""Test metadata persistence works without optional file paths."""
|
||||
metadata = {"filename": "test.pdf"}
|
||||
|
||||
persist_metadata(metadata, "/workdir/processed/test.pdf")
|
||||
|
||||
call_args = mock_json_dump.call_args
|
||||
augmented_metadata = call_args[0][0]
|
||||
assert "original_file_path" not in augmented_metadata
|
||||
assert "processed_file_path" not in augmented_metadata
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestEmbedMetadataIntoPdf:
|
||||
"""Tests for embed_metadata_into_pdf Celery task."""
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.finalize_document_storage")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.persist_metadata")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.shutil.move")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.os.makedirs")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.get_unique_filepath_with_counter")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.sanitize_filename")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4 content")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfWriter")
|
||||
def test_successful_metadata_embedding(
|
||||
self,
|
||||
mock_pdf_writer_class,
|
||||
mock_pdf_reader_class,
|
||||
mock_file,
|
||||
mock_log_progress,
|
||||
mock_session_local,
|
||||
mock_sanitize,
|
||||
mock_unique_path,
|
||||
mock_makedirs,
|
||||
mock_move,
|
||||
mock_persist,
|
||||
mock_finalize,
|
||||
):
|
||||
"""Test successful embedding of metadata into PDF."""
|
||||
# Mock PDF reader/writer
|
||||
mock_reader = MagicMock()
|
||||
mock_reader.pages = [MagicMock(), MagicMock()]
|
||||
mock_pdf_reader_class.return_value = mock_reader
|
||||
|
||||
mock_writer = MagicMock()
|
||||
mock_pdf_writer_class.return_value = mock_writer
|
||||
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_file_record = MagicMock()
|
||||
mock_file_record.id = 123
|
||||
mock_file_record.original_file_path = "/workdir/original/file.pdf"
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_file_record
|
||||
|
||||
# Mock other dependencies
|
||||
mock_sanitize.return_value = "2024-01-15_Invoice"
|
||||
mock_unique_path.return_value = "/workdir/processed/2024-01-15_Invoice.pdf"
|
||||
mock_persist.return_value = "/workdir/processed/2024-01-15_Invoice.json"
|
||||
|
||||
# Mock tempfile
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile") as mock_tempfile:
|
||||
mock_temp = MagicMock()
|
||||
mock_temp.name = "/tmp/processed_123.pdf"
|
||||
mock_tempfile.return_value = mock_temp
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
# Mock task context
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
metadata = {
|
||||
"filename": "2024-01-15_Invoice.pdf",
|
||||
"absender": "Amazon",
|
||||
"document_type": "Invoice",
|
||||
"tags": ["invoice", "amazon"],
|
||||
}
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "Sample text", metadata, file_id=123
|
||||
)
|
||||
|
||||
# Verify PDF metadata was set
|
||||
mock_writer.add_metadata.assert_called_once()
|
||||
metadata_call = mock_writer.add_metadata.call_args[0][0]
|
||||
assert metadata_call["/Title"] == "2024-01-15_Invoice.pdf"
|
||||
assert metadata_call["/Author"] == "Amazon"
|
||||
assert metadata_call["/Subject"] == "Invoice"
|
||||
assert "invoice" in metadata_call["/Keywords"]
|
||||
assert "amazon" in metadata_call["/Keywords"]
|
||||
|
||||
# Verify file was moved
|
||||
mock_move.assert_called_once()
|
||||
|
||||
# Verify finalize task was queued
|
||||
mock_finalize.delay.assert_called_once()
|
||||
|
||||
# Verify result
|
||||
assert result["file"] == "/workdir/processed/2024-01-15_Invoice.pdf"
|
||||
assert result["metadata_file"] == "/workdir/processed/2024-01-15_Invoice.json"
|
||||
assert result["status"] == "Metadata embedded"
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
def test_handles_missing_file(self, mock_log_progress):
|
||||
"""Test handling of missing file."""
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=False):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/nonexistent/file.pdf", "text", {"filename": "test.pdf"}, file_id=123
|
||||
)
|
||||
|
||||
assert result == {"error": "File not found"}
|
||||
# Verify failure was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) > 0
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.finalize_document_storage")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.persist_metadata")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.shutil.move")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.os.makedirs")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.get_unique_filepath_with_counter")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.sanitize_filename")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
def test_retrieves_file_id_from_database(
|
||||
self,
|
||||
mock_log_progress,
|
||||
mock_session_local,
|
||||
mock_sanitize,
|
||||
mock_unique_path,
|
||||
mock_makedirs,
|
||||
mock_move,
|
||||
mock_persist,
|
||||
mock_finalize,
|
||||
):
|
||||
"""Test file_id retrieval from database when not provided."""
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_file_record = MagicMock()
|
||||
mock_file_record.id = 456
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_file_record
|
||||
|
||||
mock_sanitize.return_value = "test"
|
||||
mock_unique_path.return_value = "/workdir/processed/test.pdf"
|
||||
mock_persist.return_value = "/workdir/processed/test.json"
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
with patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfWriter"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile"):
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "text", {"filename": "test.pdf"}
|
||||
)
|
||||
|
||||
# Verify database was queried
|
||||
mock_db.query.assert_called()
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader")
|
||||
def test_handles_pdf_processing_exception(self, mock_pdf_reader, mock_file, mock_session_local, mock_log_progress):
|
||||
"""Test handling of PDF processing exceptions."""
|
||||
mock_pdf_reader.side_effect = Exception("Invalid PDF structure")
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile") as mock_tempfile:
|
||||
mock_temp = MagicMock()
|
||||
mock_temp.name = "/tmp/temp.pdf"
|
||||
mock_tempfile.return_value = mock_temp
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "text", {"filename": "test.pdf"}, file_id=789
|
||||
)
|
||||
|
||||
assert "error" in result
|
||||
# Verify failure was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) > 0
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.finalize_document_storage")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.persist_metadata")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.shutil.move")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.os.makedirs")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.get_unique_filepath_with_counter")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.sanitize_filename")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfWriter")
|
||||
def test_sanitizes_malicious_filename(
|
||||
self,
|
||||
mock_pdf_writer_class,
|
||||
mock_pdf_reader_class,
|
||||
mock_file,
|
||||
mock_log_progress,
|
||||
mock_session_local,
|
||||
mock_sanitize,
|
||||
mock_unique_path,
|
||||
mock_makedirs,
|
||||
mock_move,
|
||||
mock_persist,
|
||||
mock_finalize,
|
||||
):
|
||||
"""Test filename sanitization to prevent path traversal."""
|
||||
# Mock PDF reader/writer
|
||||
mock_reader = MagicMock()
|
||||
mock_reader.pages = [MagicMock()]
|
||||
mock_pdf_reader_class.return_value = mock_reader
|
||||
mock_pdf_writer_class.return_value = MagicMock()
|
||||
|
||||
# Mock database
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
||||
|
||||
# Sanitize should remove dangerous characters
|
||||
mock_sanitize.return_value = "safe_filename"
|
||||
mock_unique_path.return_value = "/workdir/processed/safe_filename.pdf"
|
||||
mock_persist.return_value = "/workdir/processed/safe_filename.json"
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile") as mock_tempfile:
|
||||
mock_temp = MagicMock()
|
||||
mock_temp.name = "/tmp/processed.pdf"
|
||||
mock_tempfile.return_value = mock_temp
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
# Try to embed metadata with malicious filename
|
||||
metadata = {"filename": "../../../etc/passwd"}
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "text", metadata, file_id=111
|
||||
)
|
||||
|
||||
# Verify sanitize_filename was called
|
||||
mock_sanitize.assert_called_once()
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.finalize_document_storage")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.persist_metadata")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.shutil.move")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.os.makedirs")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.get_unique_filepath_with_counter")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.sanitize_filename")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfWriter")
|
||||
def test_handles_missing_metadata_fields(
|
||||
self,
|
||||
mock_pdf_writer_class,
|
||||
mock_pdf_reader_class,
|
||||
mock_file,
|
||||
mock_log_progress,
|
||||
mock_session_local,
|
||||
mock_sanitize,
|
||||
mock_unique_path,
|
||||
mock_makedirs,
|
||||
mock_move,
|
||||
mock_persist,
|
||||
mock_finalize,
|
||||
):
|
||||
"""Test handling of metadata with missing fields."""
|
||||
mock_reader = MagicMock()
|
||||
mock_reader.pages = [MagicMock()]
|
||||
mock_pdf_reader_class.return_value = mock_reader
|
||||
|
||||
mock_writer = MagicMock()
|
||||
mock_pdf_writer_class.return_value = mock_writer
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
||||
|
||||
mock_sanitize.return_value = "test"
|
||||
mock_unique_path.return_value = "/workdir/processed/test.pdf"
|
||||
mock_persist.return_value = "/workdir/processed/test.json"
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile") as mock_tempfile:
|
||||
mock_temp = MagicMock()
|
||||
mock_temp.name = "/tmp/temp.pdf"
|
||||
mock_tempfile.return_value = mock_temp
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
# Metadata with missing fields
|
||||
metadata = {}
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "text", metadata, file_id=222
|
||||
)
|
||||
|
||||
# Verify PDF metadata was set with defaults
|
||||
mock_writer.add_metadata.assert_called_once()
|
||||
metadata_call = mock_writer.add_metadata.call_args[0][0]
|
||||
assert metadata_call["/Title"] == "Unknown Document"
|
||||
assert metadata_call["/Author"] == "Unknown"
|
||||
assert metadata_call["/Subject"] == "Unknown"
|
||||
assert metadata_call["/Keywords"] == ""
|
||||
|
||||
@patch("app.tasks.embed_metadata_into_pdf.finalize_document_storage")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.persist_metadata")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.shutil.move")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.os.makedirs")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.get_unique_filepath_with_counter")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.sanitize_filename")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.SessionLocal")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.log_task_progress")
|
||||
@patch("builtins.open", new_callable=mock_open, read_data=b"%PDF-1.4")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfReader")
|
||||
@patch("app.tasks.embed_metadata_into_pdf.pypdf.PdfWriter")
|
||||
def test_deletes_original_file_from_tmp(
|
||||
self,
|
||||
mock_pdf_writer_class,
|
||||
mock_pdf_reader_class,
|
||||
mock_file,
|
||||
mock_log_progress,
|
||||
mock_session_local,
|
||||
mock_sanitize,
|
||||
mock_unique_path,
|
||||
mock_makedirs,
|
||||
mock_move,
|
||||
mock_persist,
|
||||
mock_finalize,
|
||||
):
|
||||
"""Test that original file in tmp directory is deleted after processing."""
|
||||
mock_reader = MagicMock()
|
||||
mock_reader.pages = [MagicMock()]
|
||||
mock_pdf_reader_class.return_value = mock_reader
|
||||
mock_pdf_writer_class.return_value = MagicMock()
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
||||
|
||||
mock_sanitize.return_value = "test"
|
||||
mock_unique_path.return_value = "/workdir/processed/test.pdf"
|
||||
mock_persist.return_value = "/workdir/processed/test.json"
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.tempfile.NamedTemporaryFile") as mock_tempfile:
|
||||
mock_temp = MagicMock()
|
||||
mock_temp.name = "/tmp/temp.pdf"
|
||||
mock_tempfile.return_value = mock_temp
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.shutil.copy"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.Path") as mock_path_class:
|
||||
# Mock Path for deletion logic
|
||||
mock_original_path = MagicMock()
|
||||
mock_original_path.exists.return_value = True
|
||||
mock_original_path.is_relative_to.return_value = True
|
||||
mock_workdir_path = MagicMock()
|
||||
mock_path_class.side_effect = [mock_workdir_path, mock_original_path, mock_workdir_path]
|
||||
|
||||
with patch("app.tasks.embed_metadata_into_pdf.os.remove"):
|
||||
with patch("app.tasks.embed_metadata_into_pdf.settings") as mock_settings:
|
||||
mock_settings.workdir = "/workdir"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = embed_metadata_into_pdf.__wrapped__(
|
||||
mock_task, "/workdir/tmp/test.pdf", "text", {"filename": "test.pdf"}, file_id=333
|
||||
)
|
||||
|
||||
# Verify unlink (delete) was called
|
||||
mock_original_path.unlink.assert_called_once()
|
||||
@@ -0,0 +1,287 @@
|
||||
"""Comprehensive unit tests for app/tasks/extract_metadata_with_gpt.py module."""
|
||||
|
||||
import json
|
||||
from unittest.mock import MagicMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.extract_metadata_with_gpt import extract_json_from_text, extract_metadata_with_gpt
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExtractJsonFromText:
|
||||
"""Tests for extract_json_from_text function."""
|
||||
|
||||
def test_extracts_json_from_backticks_with_json_tag(self):
|
||||
"""Test extraction of JSON from triple-backtick block with json tag."""
|
||||
text = '```json\n{"key": "value", "num": 123}\n```'
|
||||
result = extract_json_from_text(text)
|
||||
assert result == '{"key": "value", "num": 123}'
|
||||
|
||||
def test_extracts_json_from_backticks_no_lang(self):
|
||||
"""Test extraction from backticks without language tag."""
|
||||
text = '```\n{"key": "value"}\n```'
|
||||
result = extract_json_from_text(text)
|
||||
assert result == '{"key": "value"}'
|
||||
|
||||
def test_extracts_json_from_raw_text(self):
|
||||
"""Test extraction from raw text with JSON."""
|
||||
text = 'Here is the result: {"key": "value", "nested": {"a": 1}} end.'
|
||||
result = extract_json_from_text(text)
|
||||
assert result == '{"key": "value", "nested": {"a": 1}}'
|
||||
|
||||
def test_returns_none_for_no_json(self):
|
||||
"""Test returns None when no JSON found."""
|
||||
text = "No JSON here at all, just plain text."
|
||||
result = extract_json_from_text(text)
|
||||
assert result is None
|
||||
|
||||
def test_returns_none_for_incomplete_json(self):
|
||||
"""Test returns None for incomplete JSON structures."""
|
||||
text = "Only opening brace: { but no closing"
|
||||
result = extract_json_from_text(text)
|
||||
assert result is None
|
||||
|
||||
def test_extracts_complex_nested_json(self):
|
||||
"""Test extraction of complex nested JSON."""
|
||||
text = '{"filename": "2024-01-01_Invoice", "tags": ["test", "invoice"], "metadata": {"amount": 100, "currency": "USD"}}'
|
||||
result = extract_json_from_text(text)
|
||||
parsed = json.loads(result)
|
||||
assert parsed["filename"] == "2024-01-01_Invoice"
|
||||
assert "tags" in parsed
|
||||
assert "metadata" in parsed
|
||||
assert parsed["metadata"]["amount"] == 100
|
||||
|
||||
def test_extracts_first_json_when_multiple_present(self):
|
||||
"""Test that extraction finds the outermost JSON object."""
|
||||
text = 'First: {"a": 1} and second: {"b": 2}'
|
||||
result = extract_json_from_text(text)
|
||||
# Should extract from first { to last }
|
||||
assert result is not None
|
||||
assert "{" in result and "}" in result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExtractMetadataWithGpt:
|
||||
"""Tests for extract_metadata_with_gpt Celery task."""
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_successful_metadata_extraction(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test successful metadata extraction with valid GPT response."""
|
||||
# Mock the OpenAI client response
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = json.dumps({
|
||||
"filename": "2024-01-15_Invoice_Amazon",
|
||||
"empfaenger": "John Doe",
|
||||
"absender": "Amazon",
|
||||
"correspondent": "Amazon",
|
||||
"kommunikationsart": "Rechnung",
|
||||
"kommunikationskategorie": "Finanz_und_Vertragsdokumente",
|
||||
"document_type": "Invoice",
|
||||
"tags": ["invoice", "amazon", "online-shopping"],
|
||||
"language": "de",
|
||||
"title": "Amazon Purchase Invoice",
|
||||
"confidence_score": 95,
|
||||
"reference_number": "INV-2024-001",
|
||||
"monetary_amounts": ["99.99 EUR"]
|
||||
})
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
# Mock the task context
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
# Call the underlying function directly (not through Celery)
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test_invoice.pdf", "Invoice from Amazon for 99.99 EUR", 123)
|
||||
|
||||
# Verify OpenAI was called
|
||||
mock_client.chat.completions.create.assert_called_once()
|
||||
call_args = mock_client.chat.completions.create.call_args
|
||||
assert call_args[1]["temperature"] == 0
|
||||
assert len(call_args[1]["messages"]) == 2
|
||||
|
||||
# Verify metadata was extracted correctly
|
||||
assert result["s3_file"] == "test_invoice.pdf"
|
||||
assert "metadata" in result
|
||||
assert result["metadata"]["document_type"] == "Invoice"
|
||||
assert result["metadata"]["correspondent"] == "Amazon"
|
||||
|
||||
# Verify embed task was queued
|
||||
mock_embed_task.delay.assert_called_once()
|
||||
|
||||
# Verify task progress was logged
|
||||
assert mock_log_progress.call_count >= 3
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_handles_json_in_backticks(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test extraction handles JSON wrapped in markdown code blocks."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = '```json\n{"filename": "test.pdf", "document_type": "Unknown"}\n```'
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 456)
|
||||
|
||||
assert result["metadata"]["filename"] == "test.pdf"
|
||||
assert result["metadata"]["document_type"] == "Unknown"
|
||||
mock_embed_task.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_handles_invalid_json_response(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test handling of invalid JSON in GPT response."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = "This is not valid JSON at all"
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 789)
|
||||
|
||||
assert result == {}
|
||||
mock_embed_task.delay.assert_not_called()
|
||||
# Verify failure was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) > 0
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_handles_openai_api_exception(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test handling of OpenAI API exceptions."""
|
||||
mock_client.chat.completions.create.side_effect = Exception("API Error: Rate limit exceeded")
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 101)
|
||||
|
||||
assert result == {}
|
||||
mock_embed_task.delay.assert_not_called()
|
||||
# Verify exception was logged
|
||||
failure_calls = [call for call in mock_log_progress.call_args_list if "failure" in str(call)]
|
||||
assert len(failure_calls) > 0
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.SessionLocal")
|
||||
def test_retrieves_file_id_from_database_when_not_provided(
|
||||
self, mock_session_local, mock_client, mock_log_progress, mock_embed_task
|
||||
):
|
||||
"""Test file_id retrieval from database when not provided."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = '{"filename": "test.pdf", "document_type": "Unknown"}'
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_file_record = MagicMock()
|
||||
mock_file_record.id = 999
|
||||
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_file_record
|
||||
|
||||
# Mock file existence
|
||||
with patch("app.tasks.extract_metadata_with_gpt.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.extract_metadata_with_gpt.settings.workdir", "/tmp"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(
|
||||
mock_task,
|
||||
filename="test.pdf",
|
||||
cleaned_text="Sample text",
|
||||
file_id=None # Not provided
|
||||
)
|
||||
|
||||
assert result["metadata"]["filename"] == "test.pdf"
|
||||
# Verify database was queried
|
||||
mock_db.query.assert_called_once()
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_validates_filename_security(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test filename validation to prevent path traversal."""
|
||||
mock_completion = MagicMock()
|
||||
# Try to inject a malicious filename
|
||||
mock_completion.choices[0].message.content = json.dumps({
|
||||
"filename": "../../../etc/passwd",
|
||||
"document_type": "Invoice"
|
||||
})
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 202)
|
||||
|
||||
# Filename should be sanitized (empty or safe)
|
||||
assert result["metadata"]["filename"] == ""
|
||||
mock_embed_task.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_validates_filename_with_dots(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test filename validation rejects '..' in filenames."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = json.dumps({
|
||||
"filename": "test..invoice.pdf",
|
||||
"document_type": "Invoice"
|
||||
})
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 303)
|
||||
|
||||
# Filename with .. should be rejected
|
||||
assert result["metadata"]["filename"] == ""
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_accepts_valid_filename(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test that valid filenames are accepted."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = json.dumps({
|
||||
"filename": "2024-01-15_Invoice_Amazon.pdf",
|
||||
"document_type": "Invoice"
|
||||
})
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 404)
|
||||
|
||||
# Valid filename should be preserved
|
||||
assert result["metadata"]["filename"] == "2024-01-15_Invoice_Amazon.pdf"
|
||||
|
||||
@patch("app.tasks.extract_metadata_with_gpt.embed_metadata_into_pdf")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.log_task_progress")
|
||||
@patch("app.tasks.extract_metadata_with_gpt.client")
|
||||
def test_handles_malformed_json_with_valid_structure(self, mock_client, mock_log_progress, mock_embed_task):
|
||||
"""Test handling of JSON that's parseable but missing expected fields."""
|
||||
mock_completion = MagicMock()
|
||||
mock_completion.choices[0].message.content = '{"unexpected_field": "value"}'
|
||||
mock_client.chat.completions.create.return_value = mock_completion
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = extract_metadata_with_gpt.__wrapped__(mock_task, "test.pdf", "Sample text", 505)
|
||||
|
||||
# Should still extract the JSON even if fields are unexpected
|
||||
assert "metadata" in result
|
||||
assert result["metadata"]["unexpected_field"] == "value"
|
||||
+376
-13
@@ -1,24 +1,387 @@
|
||||
"""Tests for app/tasks/finalize_document_storage.py module."""
|
||||
"""Comprehensive unit tests for app/tasks/finalize_document_storage.py module."""
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.finalize_document_storage import finalize_document_storage
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestFinalizeDocumentStorageHelpers:
|
||||
"""Tests for helper functions used in finalize_document_storage."""
|
||||
class TestFinalizeDocumentStorage:
|
||||
"""Tests for finalize_document_storage Celery task."""
|
||||
|
||||
def test_get_configured_services_from_validator(self):
|
||||
"""Test that get_configured_services_from_validator returns a dict."""
|
||||
from app.tasks.send_to_all import get_configured_services_from_validator
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_successful_finalization(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test successful document finalization with all services configured."""
|
||||
# Mock configured services
|
||||
mock_get_services.return_value = {
|
||||
"dropbox": True,
|
||||
"google_drive": True,
|
||||
"nextcloud": False,
|
||||
"s3": True,
|
||||
}
|
||||
|
||||
result = get_configured_services_from_validator()
|
||||
assert isinstance(result, dict)
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_file_record = MagicMock()
|
||||
mock_file_record.id = 123
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = mock_file_record
|
||||
|
||||
def test_module_imports(self):
|
||||
"""Test that the module can be imported without errors."""
|
||||
from app.tasks.finalize_document_storage import finalize_document_storage
|
||||
# Mock file existence and size
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=102400):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="test_document.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
assert callable(finalize_document_storage)
|
||||
metadata = {
|
||||
"filename": "test_document.pdf",
|
||||
"document_type": "Invoice",
|
||||
"tags": ["invoice", "test"],
|
||||
}
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/test_document.pdf",
|
||||
metadata=metadata,
|
||||
file_id=123,
|
||||
)
|
||||
|
||||
# Verify send_to_all_destinations was queued
|
||||
mock_send_all.delay.assert_called_once_with("/workdir/processed/test_document.pdf", True, 123)
|
||||
|
||||
# Verify notification was sent
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
assert notify_args["filename"] == "test_document.pdf"
|
||||
assert notify_args["file_size"] == 102400
|
||||
assert notify_args["metadata"] == metadata
|
||||
assert "Dropbox" in notify_args["destinations"]
|
||||
assert "Google Drive" in notify_args["destinations"]
|
||||
assert "S3" in notify_args["destinations"]
|
||||
|
||||
# Verify result
|
||||
assert result["status"] == "Completed"
|
||||
assert result["file"] == "/workdir/processed/test_document.pdf"
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_retrieves_file_id_from_database_when_not_provided(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test file_id retrieval from database when not provided."""
|
||||
mock_get_services.return_value = {"dropbox": True}
|
||||
|
||||
# Mock database session to return a file record
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_file_record = MagicMock()
|
||||
mock_file_record.id = 456
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = mock_file_record
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=50000):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="doc.pdf"):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.join", return_value="/tmp/tmp/original.pdf"):
|
||||
with patch("app.tasks.finalize_document_storage.settings") as mock_settings:
|
||||
mock_settings.workdir = "/tmp"
|
||||
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/doc.pdf",
|
||||
metadata={"filename": "doc.pdf"},
|
||||
file_id=None, # Not provided
|
||||
)
|
||||
|
||||
# Verify database was queried
|
||||
mock_db.query.assert_called_once()
|
||||
|
||||
# Verify send_to_all was called with retrieved file_id
|
||||
mock_send_all.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_handles_no_configured_services(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test handles case when no services are configured."""
|
||||
# No services configured
|
||||
mock_get_services.return_value = {
|
||||
"dropbox": False,
|
||||
"google_drive": False,
|
||||
"nextcloud": False,
|
||||
"s3": False,
|
||||
}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=1024):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="test.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/test.pdf",
|
||||
metadata={"filename": "test.pdf"},
|
||||
file_id=789,
|
||||
)
|
||||
|
||||
# Should still queue uploads (even if none configured)
|
||||
mock_send_all.delay.assert_called_once()
|
||||
|
||||
# Should still send notification
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
# Should have fallback destination text
|
||||
assert len(notify_args["destinations"]) > 0
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_handles_get_configured_services_exception(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test handles exception when getting configured services."""
|
||||
# Simulate exception
|
||||
mock_get_services.side_effect = Exception("Service validation failed")
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=2048):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="file.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/file.pdf",
|
||||
metadata={"filename": "file.pdf"},
|
||||
file_id=101,
|
||||
)
|
||||
|
||||
# Should still complete successfully
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
# Should use fallback destinations
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
assert "configured destinations" in notify_args["destinations"]
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_handles_notification_failure(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test handles notification failure gracefully."""
|
||||
mock_get_services.return_value = {"dropbox": True}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
# Simulate notification failure
|
||||
mock_notify.side_effect = Exception("Notification service unavailable")
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=4096):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="doc.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/doc.pdf",
|
||||
metadata={"filename": "doc.pdf"},
|
||||
file_id=202,
|
||||
)
|
||||
|
||||
# Should still complete successfully despite notification failure
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
# Should still queue uploads
|
||||
mock_send_all.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_handles_missing_file(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test handles case when processed file doesn't exist."""
|
||||
mock_get_services.return_value = {"dropbox": True}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
# File doesn't exist
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=False):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="missing.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/missing.pdf",
|
||||
metadata={"filename": "missing.pdf"},
|
||||
file_id=303,
|
||||
)
|
||||
|
||||
# Should still queue uploads (send_to_all handles missing files)
|
||||
mock_send_all.delay.assert_called_once()
|
||||
|
||||
# Notification should use file_size = 0
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
assert notify_args["file_size"] == 0
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_formats_service_names_for_display(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test that service names are formatted correctly for display."""
|
||||
# Mock services with underscores in names
|
||||
mock_get_services.return_value = {
|
||||
"google_drive": True,
|
||||
"one_drive": True,
|
||||
"next_cloud": False,
|
||||
}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=8192):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="test.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/test.pdf",
|
||||
metadata={"filename": "test.pdf"},
|
||||
file_id=404,
|
||||
)
|
||||
|
||||
# Verify service names are formatted with spaces and title case
|
||||
mock_notify.assert_called_once()
|
||||
notify_args = mock_notify.call_args[1]
|
||||
destinations = notify_args["destinations"]
|
||||
assert "Google Drive" in destinations
|
||||
assert "One Drive" in destinations
|
||||
assert "Next Cloud" not in destinations # Not configured
|
||||
|
||||
@patch("app.tasks.finalize_document_storage.notify_file_processed")
|
||||
@patch("app.tasks.finalize_document_storage.send_to_all_destinations")
|
||||
@patch("app.tasks.finalize_document_storage.get_configured_services_from_validator")
|
||||
@patch("app.tasks.finalize_document_storage.log_task_progress")
|
||||
@patch("app.tasks.finalize_document_storage.SessionLocal")
|
||||
def test_passes_delete_after_flag_to_send_all(
|
||||
self,
|
||||
mock_session_local,
|
||||
mock_log_progress,
|
||||
mock_get_services,
|
||||
mock_send_all,
|
||||
mock_notify,
|
||||
):
|
||||
"""Test that delete_after flag is correctly passed to send_to_all_destinations."""
|
||||
mock_get_services.return_value = {"s3": True}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
with patch("app.tasks.finalize_document_storage.os.path.exists", return_value=True):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.getsize", return_value=1024):
|
||||
with patch("app.tasks.finalize_document_storage.os.path.basename", return_value="file.pdf"):
|
||||
mock_task = MagicMock()
|
||||
mock_task.request.id = "test-task-id"
|
||||
|
||||
result = finalize_document_storage.__wrapped__(
|
||||
mock_task,
|
||||
original_file="/tmp/original.pdf",
|
||||
processed_file="/workdir/processed/file.pdf",
|
||||
metadata={"filename": "file.pdf"},
|
||||
file_id=505,
|
||||
)
|
||||
|
||||
# Verify send_to_all was called with delete_after=True
|
||||
mock_send_all.delay.assert_called_once_with("/workdir/processed/file.pdf", True, 505)
|
||||
|
||||
Reference in New Issue
Block a user