test: add tests for encryption, uptime_kuma, filename_utils, logging, and celery_worker
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -0,0 +1,114 @@
|
||||
"""
|
||||
Tests for app/celery_worker.py
|
||||
|
||||
Tests Celery worker configuration and task registration.
|
||||
Note: These tests use pytest.mark.requires_redis since they depend on Celery configuration.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestCeleryWorkerConfiguration:
|
||||
"""Test Celery worker configuration"""
|
||||
|
||||
def test_celery_worker_module_imports(self):
|
||||
"""Test that celery_worker module can be imported"""
|
||||
import app.celery_worker
|
||||
|
||||
assert hasattr(app.celery_worker, "celery")
|
||||
assert hasattr(app.celery_worker, "test_task")
|
||||
|
||||
def test_test_task_defined(self):
|
||||
"""Test that test_task is defined"""
|
||||
from app.celery_worker import test_task
|
||||
|
||||
# Task should be callable
|
||||
assert callable(test_task)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestTaskImports:
|
||||
"""Test that all tasks can be imported correctly"""
|
||||
|
||||
def test_process_document_imported(self):
|
||||
"""Test process_document task import"""
|
||||
from app.celery_worker import process_document
|
||||
|
||||
assert callable(process_document)
|
||||
|
||||
def test_convert_to_pdf_imported(self):
|
||||
"""Test convert_to_pdf task import"""
|
||||
from app.celery_worker import convert_to_pdf
|
||||
|
||||
assert callable(convert_to_pdf)
|
||||
|
||||
def test_embed_metadata_into_pdf_imported(self):
|
||||
"""Test embed_metadata_into_pdf task import"""
|
||||
from app.celery_worker import embed_metadata_into_pdf
|
||||
|
||||
assert callable(embed_metadata_into_pdf)
|
||||
|
||||
def test_extract_metadata_with_gpt_imported(self):
|
||||
"""Test extract_metadata_with_gpt task import"""
|
||||
from app.celery_worker import extract_metadata_with_gpt
|
||||
|
||||
assert callable(extract_metadata_with_gpt)
|
||||
|
||||
def test_send_to_all_destinations_imported(self):
|
||||
"""Test send_to_all_destinations task import"""
|
||||
from app.celery_worker import send_to_all_destinations
|
||||
|
||||
assert callable(send_to_all_destinations)
|
||||
|
||||
def test_upload_tasks_imported(self):
|
||||
"""Test that upload tasks are imported"""
|
||||
from app.celery_worker import (
|
||||
upload_to_dropbox,
|
||||
upload_to_email,
|
||||
upload_to_ftp,
|
||||
upload_to_google_drive,
|
||||
upload_to_nextcloud,
|
||||
upload_to_onedrive,
|
||||
upload_to_paperless,
|
||||
upload_to_s3,
|
||||
upload_to_sftp,
|
||||
upload_to_webdav,
|
||||
)
|
||||
|
||||
# All should be callable
|
||||
assert callable(upload_to_dropbox)
|
||||
assert callable(upload_to_email)
|
||||
assert callable(upload_to_ftp)
|
||||
assert callable(upload_to_google_drive)
|
||||
assert callable(upload_to_nextcloud)
|
||||
assert callable(upload_to_onedrive)
|
||||
assert callable(upload_to_paperless)
|
||||
assert callable(upload_to_s3)
|
||||
assert callable(upload_to_sftp)
|
||||
assert callable(upload_to_webdav)
|
||||
|
||||
def test_utility_tasks_imported(self):
|
||||
"""Test utility tasks are imported"""
|
||||
from app.celery_worker import (
|
||||
pull_all_inboxes,
|
||||
ping_uptime_kuma,
|
||||
check_credentials,
|
||||
)
|
||||
|
||||
assert callable(pull_all_inboxes)
|
||||
assert callable(ping_uptime_kuma)
|
||||
assert callable(check_credentials)
|
||||
|
||||
def test_processing_tasks_imported(self):
|
||||
"""Test processing tasks are imported"""
|
||||
from app.celery_worker import (
|
||||
process_with_azure_document_intelligence,
|
||||
refine_text_with_gpt,
|
||||
rotate_pdf_pages,
|
||||
)
|
||||
|
||||
assert callable(process_with_azure_document_intelligence)
|
||||
assert callable(refine_text_with_gpt)
|
||||
assert callable(rotate_pdf_pages)
|
||||
@@ -0,0 +1,231 @@
|
||||
"""
|
||||
Tests for app/utils/encryption.py
|
||||
|
||||
Tests encryption/decryption functionality for sensitive settings.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestEncryption:
|
||||
"""Test encryption utility functions"""
|
||||
|
||||
def test_encrypt_value_with_none(self):
|
||||
"""Test that None values are returned as-is"""
|
||||
from app.utils.encryption import encrypt_value
|
||||
|
||||
result = encrypt_value(None)
|
||||
assert result is None
|
||||
|
||||
def test_encrypt_value_with_empty_string(self):
|
||||
"""Test that empty strings are returned as-is"""
|
||||
from app.utils.encryption import encrypt_value
|
||||
|
||||
result = encrypt_value("")
|
||||
assert result == ""
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_encrypt_value_when_encryption_unavailable(self, mock_cipher):
|
||||
"""Test that plaintext is returned when encryption is unavailable"""
|
||||
from app.utils.encryption import encrypt_value
|
||||
|
||||
mock_cipher.return_value = None
|
||||
result = encrypt_value("secret_value")
|
||||
|
||||
# Should return plaintext with warning logged
|
||||
assert result == "secret_value"
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_encrypt_value_success(self, mock_cipher):
|
||||
"""Test successful encryption"""
|
||||
from app.utils.encryption import encrypt_value
|
||||
|
||||
# Mock cipher that returns encrypted bytes
|
||||
mock_fernet = Mock()
|
||||
mock_fernet.encrypt.return_value = b"encrypted_data"
|
||||
mock_cipher.return_value = mock_fernet
|
||||
|
||||
result = encrypt_value("secret_value")
|
||||
|
||||
# Should have "enc:" prefix
|
||||
assert result.startswith("enc:")
|
||||
assert "encrypted_data" in result
|
||||
mock_fernet.encrypt.assert_called_once()
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_encrypt_value_encryption_failure(self, mock_cipher):
|
||||
"""Test that encryption failures fall back to plaintext"""
|
||||
from app.utils.encryption import encrypt_value
|
||||
|
||||
# Mock cipher that raises exception
|
||||
mock_fernet = Mock()
|
||||
mock_fernet.encrypt.side_effect = Exception("Encryption error")
|
||||
mock_cipher.return_value = mock_fernet
|
||||
|
||||
result = encrypt_value("secret_value")
|
||||
|
||||
# Should fall back to plaintext
|
||||
assert result == "secret_value"
|
||||
|
||||
def test_decrypt_value_with_none(self):
|
||||
"""Test that None values are returned as-is"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
result = decrypt_value(None)
|
||||
assert result is None
|
||||
|
||||
def test_decrypt_value_with_empty_string(self):
|
||||
"""Test that empty strings are returned as-is"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
result = decrypt_value("")
|
||||
assert result == ""
|
||||
|
||||
def test_decrypt_value_plaintext(self):
|
||||
"""Test that plaintext values without enc: prefix are returned as-is"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
result = decrypt_value("plain_value")
|
||||
assert result == "plain_value"
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_decrypt_value_when_encryption_unavailable(self, mock_cipher):
|
||||
"""Test decryption when cipher is unavailable"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
mock_cipher.return_value = None
|
||||
result = decrypt_value("enc:encrypted_data")
|
||||
|
||||
# Should return error message
|
||||
assert result == "[ENCRYPTED - Cannot decrypt]"
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_decrypt_value_success(self, mock_cipher):
|
||||
"""Test successful decryption"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
# Mock cipher that returns decrypted bytes
|
||||
mock_fernet = Mock()
|
||||
mock_fernet.decrypt.return_value = b"decrypted_value"
|
||||
mock_cipher.return_value = mock_fernet
|
||||
|
||||
result = decrypt_value("enc:encrypted_data")
|
||||
|
||||
assert result == "decrypted_value"
|
||||
mock_fernet.decrypt.assert_called_once()
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_decrypt_value_decryption_failure(self, mock_cipher):
|
||||
"""Test that decryption failures return error message"""
|
||||
from app.utils.encryption import decrypt_value
|
||||
|
||||
# Mock cipher that raises exception
|
||||
mock_fernet = Mock()
|
||||
mock_fernet.decrypt.side_effect = Exception("Decryption error")
|
||||
mock_cipher.return_value = mock_fernet
|
||||
|
||||
result = decrypt_value("enc:bad_data")
|
||||
|
||||
# Should return error message
|
||||
assert result == "[DECRYPTION FAILED]"
|
||||
|
||||
def test_is_encrypted_with_encrypted_value(self):
|
||||
"""Test is_encrypted returns True for encrypted values"""
|
||||
from app.utils.encryption import is_encrypted
|
||||
|
||||
assert is_encrypted("enc:some_encrypted_data") is True
|
||||
|
||||
def test_is_encrypted_with_plaintext(self):
|
||||
"""Test is_encrypted returns False for plaintext"""
|
||||
from app.utils.encryption import is_encrypted
|
||||
|
||||
assert is_encrypted("plain_value") is False
|
||||
|
||||
def test_is_encrypted_with_none(self):
|
||||
"""Test is_encrypted returns False for None"""
|
||||
from app.utils.encryption import is_encrypted
|
||||
|
||||
assert is_encrypted(None) is False
|
||||
|
||||
def test_is_encrypted_with_empty_string(self):
|
||||
"""Test is_encrypted returns False for empty string"""
|
||||
from app.utils.encryption import is_encrypted
|
||||
|
||||
assert is_encrypted("") is False
|
||||
|
||||
def test_is_encrypted_with_non_string(self):
|
||||
"""Test is_encrypted returns False for non-string types"""
|
||||
from app.utils.encryption import is_encrypted
|
||||
|
||||
assert is_encrypted(123) is False
|
||||
assert is_encrypted([]) is False
|
||||
assert is_encrypted({}) is False
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_is_encryption_available_true(self, mock_cipher):
|
||||
"""Test is_encryption_available when cryptography is available"""
|
||||
from app.utils.encryption import is_encryption_available
|
||||
|
||||
mock_cipher.return_value = Mock() # Non-None cipher
|
||||
assert is_encryption_available() is True
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_is_encryption_available_false(self, mock_cipher):
|
||||
"""Test is_encryption_available when cryptography is not available"""
|
||||
from app.utils.encryption import is_encryption_available
|
||||
|
||||
mock_cipher.return_value = None
|
||||
assert is_encryption_available() is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetCipherSuite:
|
||||
"""Test the _get_cipher_suite internal function"""
|
||||
|
||||
def test_cipher_suite_caching(self):
|
||||
"""Test that cipher suite is cached after first call"""
|
||||
import app.utils.encryption
|
||||
|
||||
# First call
|
||||
result1 = app.utils.encryption._get_cipher_suite()
|
||||
|
||||
# Second call should return same instance (cached)
|
||||
result2 = app.utils.encryption._get_cipher_suite()
|
||||
|
||||
# Both calls should return the same object (cached)
|
||||
assert result1 is result2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestEncryptionIntegration:
|
||||
"""Integration tests for encrypt/decrypt cycle"""
|
||||
|
||||
@patch("app.utils.encryption._get_cipher_suite")
|
||||
def test_encrypt_decrypt_cycle(self, mock_cipher):
|
||||
"""Test that encrypting and then decrypting returns original value"""
|
||||
from app.utils.encryption import encrypt_value, decrypt_value
|
||||
|
||||
# Mock a simple reversible encryption
|
||||
mock_fernet = Mock()
|
||||
|
||||
# Simulate encryption: just add a prefix
|
||||
def mock_encrypt(data):
|
||||
return b"ENCRYPTED_" + data
|
||||
|
||||
# Simulate decryption: remove the prefix
|
||||
def mock_decrypt(data):
|
||||
return data.replace(b"ENCRYPTED_", b"")
|
||||
|
||||
mock_fernet.encrypt = mock_encrypt
|
||||
mock_fernet.decrypt = mock_decrypt
|
||||
mock_cipher.return_value = mock_fernet
|
||||
|
||||
original = "my_secret_password"
|
||||
encrypted = encrypt_value(original)
|
||||
decrypted = decrypt_value(encrypted)
|
||||
|
||||
assert encrypted != original
|
||||
assert encrypted.startswith("enc:")
|
||||
assert decrypted == original
|
||||
@@ -0,0 +1,259 @@
|
||||
"""
|
||||
Tests for app/utils/filename_utils.py
|
||||
|
||||
Tests filename sanitization and manipulation functions.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
import os
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestFilenameSanitization:
|
||||
"""Test filename sanitization functions"""
|
||||
|
||||
def test_sanitize_filename_basic(self):
|
||||
"""Test basic filename sanitization"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
# Basic valid filename
|
||||
result = sanitize_filename("document.pdf")
|
||||
assert result == "document.pdf"
|
||||
|
||||
def test_sanitize_filename_with_spaces(self):
|
||||
"""Test sanitization of filenames with spaces"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("my document file.pdf")
|
||||
# Spaces should be preserved
|
||||
assert "my" in result
|
||||
assert "document" in result
|
||||
assert "file.pdf" in result
|
||||
|
||||
def test_sanitize_filename_with_special_characters(self):
|
||||
"""Test sanitization removes or replaces special characters"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("file:with*special?chars.pdf")
|
||||
# Special characters should be replaced with underscores
|
||||
assert ":" not in result
|
||||
assert "*" not in result
|
||||
assert "?" not in result
|
||||
assert "_" in result
|
||||
|
||||
def test_sanitize_filename_with_path_separators(self):
|
||||
"""Test that path separators are handled"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("../../../etc/passwd")
|
||||
# Path traversal characters should be replaced
|
||||
assert ".." not in result or result.count("..") < 3
|
||||
|
||||
def test_sanitize_filename_empty_string(self):
|
||||
"""Test sanitization of empty string"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("")
|
||||
# Should return a valid string (default name with timestamp)
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
assert "document" in result
|
||||
|
||||
def test_sanitize_filename_only_periods(self):
|
||||
"""Test sanitization of only periods"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("...")
|
||||
# Should return a default name
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
assert "document" in result
|
||||
|
||||
def test_sanitize_filename_leading_trailing_spaces(self):
|
||||
"""Test sanitization trims leading/trailing spaces"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename(" filename.pdf ")
|
||||
assert result == "filename.pdf"
|
||||
|
||||
def test_sanitize_filename_multiple_underscores(self):
|
||||
"""Test sanitization collapses multiple underscores"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("file____name.pdf")
|
||||
assert "____" not in result
|
||||
assert result == "file_name.pdf"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUniqueFilenameGeneration:
|
||||
"""Test unique filename generation"""
|
||||
|
||||
def test_get_unique_filename_no_collision(self):
|
||||
"""Test that original filename is returned when no collision"""
|
||||
from app.utils.filename_utils import get_unique_filename
|
||||
|
||||
# Mock check_exists_func to return False (file doesn't exist)
|
||||
check_func = Mock(return_value=False)
|
||||
|
||||
result = get_unique_filename("/tmp/document.pdf", check_exists_func=check_func)
|
||||
assert result == "/tmp/document.pdf"
|
||||
check_func.assert_called_once_with("/tmp/document.pdf")
|
||||
|
||||
def test_get_unique_filename_with_collision(self):
|
||||
"""Test that unique filename is generated on collision"""
|
||||
from app.utils.filename_utils import get_unique_filename
|
||||
|
||||
# Mock check_exists_func to return True for original, False for timestamped
|
||||
def check_func(path):
|
||||
return path == "/tmp/document.pdf"
|
||||
|
||||
result = get_unique_filename("/tmp/document.pdf", check_exists_func=check_func)
|
||||
assert result != "/tmp/document.pdf"
|
||||
assert "document" in result
|
||||
assert ".pdf" in result
|
||||
|
||||
def test_get_unique_filename_uses_timestamp(self):
|
||||
"""Test that timestamp is added on collision"""
|
||||
from app.utils.filename_utils import get_unique_filename
|
||||
|
||||
# First file exists
|
||||
check_func = Mock(side_effect=[True, False])
|
||||
|
||||
result = get_unique_filename("/tmp/test.pdf", check_exists_func=check_func)
|
||||
assert result != "/tmp/test.pdf"
|
||||
assert "test_" in result
|
||||
assert ".pdf" in result
|
||||
|
||||
def test_get_unique_filename_falls_back_to_uuid(self):
|
||||
"""Test UUID fallback when timestamp collision occurs"""
|
||||
from app.utils.filename_utils import get_unique_filename
|
||||
|
||||
# Original and timestamp both exist
|
||||
check_func = Mock(side_effect=[True, True, False])
|
||||
|
||||
result = get_unique_filename("/tmp/test.pdf", check_exists_func=check_func)
|
||||
assert result != "/tmp/test.pdf"
|
||||
assert "test_" in result
|
||||
assert ".pdf" in result
|
||||
|
||||
def test_get_unique_filename_default_check_function(self):
|
||||
"""Test that os.path.exists is used by default"""
|
||||
from app.utils.filename_utils import get_unique_filename
|
||||
|
||||
# Use actual filesystem check
|
||||
result = get_unique_filename("/tmp/nonexistent_file_12345.pdf")
|
||||
# Should return original since file doesn't exist
|
||||
assert result == "/tmp/nonexistent_file_12345.pdf"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExtractRemotePath:
|
||||
"""Test remote path extraction"""
|
||||
|
||||
def test_extract_remote_path_basic(self):
|
||||
"""Test basic remote path extraction"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/home/user/docs/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
remote_base = "Documents"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, remote_base)
|
||||
assert result == "Documents/docs/file.pdf"
|
||||
|
||||
def test_extract_remote_path_without_remote_base(self):
|
||||
"""Test remote path extraction without remote base"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/home/user/docs/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, "")
|
||||
assert result == "docs/file.pdf"
|
||||
|
||||
def test_extract_remote_path_skips_processed_dir(self):
|
||||
"""Test that 'processed' directory is skipped"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/home/user/processed/docs/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, "")
|
||||
assert "processed" not in result
|
||||
assert result == "docs/file.pdf"
|
||||
|
||||
def test_extract_remote_path_with_absolute_remote_base(self):
|
||||
"""Test remote path extraction with absolute remote base"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/home/user/docs/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
remote_base = "/Documents"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, remote_base)
|
||||
# Leading slash should be stripped
|
||||
assert result == "Documents/docs/file.pdf"
|
||||
|
||||
def test_extract_remote_path_file_outside_base(self):
|
||||
"""Test handling of file outside base directory"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/other/path/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, "")
|
||||
# Should just use filename
|
||||
assert result == "file.pdf"
|
||||
|
||||
def test_extract_remote_path_uses_forward_slashes(self):
|
||||
"""Test that result uses forward slashes"""
|
||||
from app.utils.filename_utils import extract_remote_path
|
||||
|
||||
file_path = "/home/user/docs/subfolder/file.pdf"
|
||||
base_dir = "/home/user"
|
||||
|
||||
result = extract_remote_path(file_path, base_dir, "")
|
||||
# Should use forward slashes for cloud service compatibility
|
||||
assert "/" in result
|
||||
assert "\\" not in result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestFilenameUtilsEdgeCases:
|
||||
"""Test edge cases in filename utilities"""
|
||||
|
||||
def test_very_long_filename(self):
|
||||
"""Test handling of very long filenames"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
long_name = "a" * 300 + ".pdf"
|
||||
result = sanitize_filename(long_name)
|
||||
# Should handle long filenames
|
||||
assert isinstance(result, str)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_filename_with_multiple_dots(self):
|
||||
"""Test filename with multiple dots"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename("my.document.file.name.pdf")
|
||||
assert isinstance(result, str)
|
||||
assert ".pdf" in result
|
||||
assert result == "my.document.file.name.pdf"
|
||||
|
||||
def test_sanitize_filename_windows_reserved_chars(self):
|
||||
"""Test sanitization of Windows reserved characters"""
|
||||
from app.utils.filename_utils import sanitize_filename
|
||||
|
||||
result = sanitize_filename('file<>:"|?*.pdf')
|
||||
# All reserved chars should be replaced
|
||||
assert "<" not in result
|
||||
assert ">" not in result
|
||||
assert ":" not in result
|
||||
assert '"' not in result
|
||||
assert "|" not in result
|
||||
assert "?" not in result
|
||||
@@ -0,0 +1,176 @@
|
||||
"""
|
||||
Tests for app/utils/logging.py
|
||||
|
||||
Tests task progress logging functionality.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestTaskLogging:
|
||||
"""Test task progress logging"""
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_basic(self, mock_processing_log, mock_session_local):
|
||||
"""Test basic task progress logging"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
# Mock ProcessingLog model
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
# Call the function
|
||||
log_task_progress(
|
||||
task_id="task-123",
|
||||
step_name="processing",
|
||||
status="started",
|
||||
message="Processing document",
|
||||
file_id=456,
|
||||
)
|
||||
|
||||
# Verify ProcessingLog was created with correct parameters
|
||||
mock_processing_log.assert_called_once_with(
|
||||
task_id="task-123",
|
||||
step_name="processing",
|
||||
status="started",
|
||||
message="Processing document",
|
||||
file_id=456,
|
||||
)
|
||||
|
||||
# Verify database operations
|
||||
mock_db.add.assert_called_once_with(mock_log_entry)
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_without_message(self, mock_processing_log, mock_session_local):
|
||||
"""Test logging without message"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
# Call without message
|
||||
log_task_progress(
|
||||
task_id="task-456", step_name="upload", status="completed", message=None, file_id=None
|
||||
)
|
||||
|
||||
# Verify called with None for optional parameters
|
||||
mock_processing_log.assert_called_once_with(
|
||||
task_id="task-456", step_name="upload", status="completed", message=None, file_id=None
|
||||
)
|
||||
|
||||
mock_db.add.assert_called_once()
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_without_file_id(self, mock_processing_log, mock_session_local):
|
||||
"""Test logging without file_id"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
# Call without file_id
|
||||
log_task_progress(
|
||||
task_id="task-789", step_name="metadata", status="running", message="Extracting metadata"
|
||||
)
|
||||
|
||||
# file_id should default to None
|
||||
mock_processing_log.assert_called_once()
|
||||
call_args = mock_processing_log.call_args
|
||||
assert call_args[1]["task_id"] == "task-789"
|
||||
assert call_args[1]["step_name"] == "metadata"
|
||||
assert call_args[1]["status"] == "running"
|
||||
|
||||
mock_db.add.assert_called_once()
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_all_parameters(self, mock_processing_log, mock_session_local):
|
||||
"""Test logging with all parameters"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
# Call with all parameters
|
||||
log_task_progress(
|
||||
task_id="task-complete",
|
||||
step_name="finalization",
|
||||
status="success",
|
||||
message="Document processed successfully",
|
||||
file_id=999,
|
||||
)
|
||||
|
||||
mock_processing_log.assert_called_once_with(
|
||||
task_id="task-complete",
|
||||
step_name="finalization",
|
||||
status="success",
|
||||
message="Document processed successfully",
|
||||
file_id=999,
|
||||
)
|
||||
|
||||
mock_db.add.assert_called_once()
|
||||
mock_db.commit.assert_called_once()
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_session_context_manager(self, mock_processing_log, mock_session_local):
|
||||
"""Test that database session is properly managed with context manager"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
mock_session_context = MagicMock()
|
||||
mock_session_local.return_value = mock_session_context
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_context.__enter__.return_value = mock_db
|
||||
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
log_task_progress(task_id="test", step_name="test", status="test")
|
||||
|
||||
# Verify context manager was used
|
||||
mock_session_context.__enter__.assert_called_once()
|
||||
mock_session_context.__exit__.assert_called_once()
|
||||
|
||||
@patch("app.utils.logging.SessionLocal")
|
||||
@patch("app.utils.logging.ProcessingLog")
|
||||
def test_log_task_progress_with_different_statuses(self, mock_processing_log, mock_session_local):
|
||||
"""Test logging with various status values"""
|
||||
from app.utils.logging import log_task_progress
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
mock_log_entry = Mock()
|
||||
mock_processing_log.return_value = mock_log_entry
|
||||
|
||||
statuses = ["pending", "processing", "completed", "failed", "error"]
|
||||
|
||||
for status in statuses:
|
||||
log_task_progress(task_id=f"task-{status}", step_name="test", status=status)
|
||||
|
||||
# Should be called for each status
|
||||
assert mock_processing_log.called
|
||||
mock_db.add.assert_called()
|
||||
mock_db.commit.assert_called()
|
||||
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Tests for simple re-export modules (app/utils.py, app/frontend.py, app/utils/config_validator.py)
|
||||
|
||||
These modules are simple re-exports of functions from other modules.
|
||||
We test that imports work correctly.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUtilsReexports:
|
||||
"""Test that app/utils.py re-exports work correctly"""
|
||||
|
||||
def test_hash_file_import(self):
|
||||
"""Test that hash_file can be imported from app.utils"""
|
||||
from app.utils import hash_file
|
||||
|
||||
# Function should exist and be callable
|
||||
assert callable(hash_file)
|
||||
|
||||
def test_log_task_progress_import(self):
|
||||
"""Test that log_task_progress can be imported from app.utils"""
|
||||
from app.utils import log_task_progress
|
||||
|
||||
# Function should exist and be callable
|
||||
assert callable(log_task_progress)
|
||||
|
||||
def test_utils_module_is_backward_compatible(self):
|
||||
"""Test that utils module maintains backward compatibility"""
|
||||
# The module comment says it's deprecated but maintains compatibility
|
||||
import app.utils
|
||||
|
||||
# Module should exist and have expected attributes
|
||||
assert hasattr(app.utils, "hash_file")
|
||||
assert hasattr(app.utils, "log_task_progress")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestFrontendReexports:
|
||||
"""Test that app/frontend.py re-exports work correctly"""
|
||||
|
||||
def test_router_import(self):
|
||||
"""Test that router can be imported from app.frontend"""
|
||||
from app.frontend import router
|
||||
|
||||
# Router should exist
|
||||
assert router is not None
|
||||
|
||||
def test_frontend_module_imports(self):
|
||||
"""Test that frontend module can be imported"""
|
||||
import app.frontend
|
||||
|
||||
# Module should exist and have router
|
||||
assert hasattr(app.frontend, "router")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestConfigValidatorReexports:
|
||||
"""Test that app/utils/config_validator.py re-exports work correctly"""
|
||||
|
||||
def test_validate_email_config_import(self):
|
||||
"""Test validate_email_config import"""
|
||||
from app.utils.config_validator import validate_email_config
|
||||
|
||||
assert callable(validate_email_config)
|
||||
|
||||
def test_validate_storage_configs_import(self):
|
||||
"""Test validate_storage_configs import"""
|
||||
from app.utils.config_validator import validate_storage_configs
|
||||
|
||||
assert callable(validate_storage_configs)
|
||||
|
||||
def test_validate_notification_config_import(self):
|
||||
"""Test validate_notification_config import"""
|
||||
from app.utils.config_validator import validate_notification_config
|
||||
|
||||
assert callable(validate_notification_config)
|
||||
|
||||
def test_mask_sensitive_value_import(self):
|
||||
"""Test mask_sensitive_value import"""
|
||||
from app.utils.config_validator import mask_sensitive_value
|
||||
|
||||
assert callable(mask_sensitive_value)
|
||||
|
||||
def test_get_provider_status_import(self):
|
||||
"""Test get_provider_status import"""
|
||||
from app.utils.config_validator import get_provider_status
|
||||
|
||||
assert callable(get_provider_status)
|
||||
|
||||
def test_get_settings_for_display_import(self):
|
||||
"""Test get_settings_for_display import"""
|
||||
from app.utils.config_validator import get_settings_for_display
|
||||
|
||||
assert callable(get_settings_for_display)
|
||||
|
||||
def test_dump_all_settings_import(self):
|
||||
"""Test dump_all_settings import"""
|
||||
from app.utils.config_validator import dump_all_settings
|
||||
|
||||
assert callable(dump_all_settings)
|
||||
|
||||
def test_check_all_configs_import(self):
|
||||
"""Test check_all_configs import"""
|
||||
from app.utils.config_validator import check_all_configs
|
||||
|
||||
assert callable(check_all_configs)
|
||||
|
||||
def test_config_validator_all_exports(self):
|
||||
"""Test that __all__ contains expected exports"""
|
||||
from app.utils import config_validator
|
||||
|
||||
expected_exports = [
|
||||
"validate_email_config",
|
||||
"validate_storage_configs",
|
||||
"validate_notification_config",
|
||||
"mask_sensitive_value",
|
||||
"get_provider_status",
|
||||
"get_settings_for_display",
|
||||
"dump_all_settings",
|
||||
"check_all_configs",
|
||||
]
|
||||
|
||||
# Check that __all__ is defined and contains expected items
|
||||
if hasattr(config_validator, "__all__"):
|
||||
for export in expected_exports:
|
||||
assert export in config_validator.__all__
|
||||
|
||||
# Also check direct imports work
|
||||
for export in expected_exports:
|
||||
assert hasattr(config_validator, export)
|
||||
@@ -0,0 +1,163 @@
|
||||
"""
|
||||
Tests for app/tasks/uptime_kuma_tasks.py
|
||||
|
||||
Tests Uptime Kuma health check ping functionality.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
import requests
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUptimeKumaTasks:
|
||||
"""Test Uptime Kuma ping task"""
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_no_url_configured(self, mock_settings):
|
||||
"""Test that task does nothing when URL is not configured"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = None
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return None and not make any requests
|
||||
assert result is None
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_success(self, mock_settings, mock_get):
|
||||
"""Test successful ping to Uptime Kuma"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Mock successful response
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return True on success
|
||||
assert result is True
|
||||
mock_get.assert_called_once_with(
|
||||
"https://uptime.example.com/ping/123", timeout=10
|
||||
)
|
||||
mock_response.raise_for_status.assert_called_once()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_connection_error(self, mock_settings, mock_get):
|
||||
"""Test handling of connection errors"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Mock connection error
|
||||
mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused")
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return False on error
|
||||
assert result is False
|
||||
mock_get.assert_called_once()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_timeout(self, mock_settings, mock_get):
|
||||
"""Test handling of request timeout"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Mock timeout error
|
||||
mock_get.side_effect = requests.exceptions.Timeout("Request timed out")
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return False on timeout
|
||||
assert result is False
|
||||
mock_get.assert_called_once()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_http_error(self, mock_settings, mock_get):
|
||||
"""Test handling of HTTP errors (4xx, 5xx)"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Mock HTTP error
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 500
|
||||
mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError(
|
||||
"500 Server Error"
|
||||
)
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return False on HTTP error
|
||||
assert result is False
|
||||
mock_get.assert_called_once()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_generic_request_exception(self, mock_settings, mock_get):
|
||||
"""Test handling of generic request exceptions"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Mock generic request exception
|
||||
mock_get.side_effect = requests.exceptions.RequestException("Generic error")
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return False on any request exception
|
||||
assert result is False
|
||||
mock_get.assert_called_once()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_empty_url(self, mock_settings, mock_get):
|
||||
"""Test handling of empty URL string"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = ""
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
|
||||
# Should return None and not make any requests
|
||||
assert result is None
|
||||
mock_get.assert_not_called()
|
||||
|
||||
@patch("app.tasks.uptime_kuma_tasks.requests.get")
|
||||
@patch("app.tasks.uptime_kuma_tasks.settings")
|
||||
def test_ping_uptime_kuma_with_various_status_codes(self, mock_settings, mock_get):
|
||||
"""Test successful ping with various 2xx status codes"""
|
||||
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma
|
||||
|
||||
mock_settings.uptime_kuma_url = "https://uptime.example.com/ping/123"
|
||||
|
||||
# Test with 200 OK
|
||||
mock_response = Mock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.raise_for_status = Mock()
|
||||
mock_get.return_value = mock_response
|
||||
|
||||
result = ping_uptime_kuma()
|
||||
assert result is True
|
||||
|
||||
# Test with 204 No Content
|
||||
mock_response.status_code = 204
|
||||
result = ping_uptime_kuma()
|
||||
assert result is True
|
||||
|
||||
# Test with 202 Accepted
|
||||
mock_response.status_code = 202
|
||||
result = ping_uptime_kuma()
|
||||
assert result is True
|
||||
Reference in New Issue
Block a user