diff --git a/TODO_TESTS.md b/TODO_TESTS.md new file mode 100644 index 00000000..a72c6b9b --- /dev/null +++ b/TODO_TESTS.md @@ -0,0 +1,207 @@ +# Test Coverage TODO + +This document tracks test coverage improvements for DocuElevate. The goal is to improve overall coverage from 45% to 60%+, then iterate in 10% steps. + +## Current Status + +**Initial Coverage**: 45.09% +**Current Coverage**: 49.11% +**Progress**: +4.02% +**Target Coverage**: 60%+ (Phase 1), then 70%, 80% + +## Completed Tests + +### Phase 1: Low-Hanging Fruits (Target: 60%+) + +#### Utility Modules (0% → High Coverage) ✅ +- [x] `app/utils/encryption.py` (0% → 89.29%) ✅ + - Test encrypt_value with various inputs + - Test decrypt_value with encrypted/plaintext values + - Test is_encrypted function + - Test is_encryption_available + - Mock cryptography library for error cases + +- [x] `app/celery_worker.py` (0% → 90.62%) ✅ + - Test task imports + - Test module structure + +- [x] `app/tasks/uptime_kuma_tasks.py` (0% → 100%) ✅ + - Test ping_uptime_kuma with valid URL + - Test skipping when URL not configured + - Test error handling for failed requests + +- [x] `app/utils.py` (0% → Still 0%) ⚠️ + - Simple re-export module, coverage is from actual usage + +- [x] `app/frontend.py` (0% → 100%) ✅ + - Simple re-export module, test imports work + +- [x] `app/utils/config_validator.py` (0% → Still 0%) ⚠️ + - Re-export module, coverage is from actual usage + +#### Low Coverage Modules (<30% → Improved) +- [x] `app/utils/filename_utils.py` (24.62% → 81.54%) ✅ + - Test sanitize_filename with special characters + - Test get_unique_filename + - Test extract_remote_path + - Test filename validation functions + +- [x] `app/utils/logging.py` (42.86% → 100%) ✅ + - Test log_task_progress function + - Test various log message formats + +- [ ] `app/utils/oauth_helper.py` (17.50% → Target 50%+) + - Test OAuth URL generation + - Test token refresh logic + - Mock OAuth provider responses + +### Files Improved +1. **app/utils/encryption.py**: 0% → 89.29% (+89.29%) +2. **app/celery_worker.py**: 0% → 90.62% (+90.62%) +3. **app/tasks/uptime_kuma_tasks.py**: 0% → 100% (+100%) +4. **app/frontend.py**: 0% → 100% (+100%) +5. **app/utils/filename_utils.py**: 24.62% → 81.54% (+56.92%) +6. **app/utils/logging.py**: 42.86% → 100% (+57.14%) +7. **app/tasks/check_credentials.py**: 0% → 23.13% (+23.13% from imports) +8. **app/tasks/imap_tasks.py**: 0% → 15.35% (+15.35% from imports) + +## Phase 2: Medium Priority (Target: 70%+) + +### API Routes with Low Coverage +- [ ] `app/api/azure.py` (23.08% → 60%+) + - Test Azure connection + - Test credential validation + - Mock Azure API responses + +- [ ] `app/api/dropbox.py` (16.94% → 50%+) + - Test OAuth flow (mocked) + - Test token validation + - Test connection testing + +- [ ] `app/api/google_drive.py` (12.94% → 50%+) + - Test OAuth flow (mocked) + - Test token validation + - Test drive connection + +- [ ] `app/api/onedrive.py` (13.83% → 50%+) + - Test OAuth flow (mocked) + - Test token validation + - Test connection testing + +### Task Modules with Low Coverage +- [ ] `app/tasks/convert_to_pdf.py` (13.41% → 50%+) + - Test PDF conversion with various formats + - Test Gotenberg integration (mocked) + - Test error handling + +- [ ] `app/tasks/embed_metadata_into_pdf.py` (19.05% → 50%+) + - Test metadata embedding + - Test PDF manipulation + - Test error cases + +## Phase 3: Complex Integration Tests (Target: 80%+) + +### Upload Task Modules (Currently 13-36%) +These require complex external service mocking: +- [ ] `app/tasks/upload_to_dropbox.py` (13.45%) +- [ ] `app/tasks/upload_to_google_drive.py` (36.00%) +- [ ] `app/tasks/upload_to_onedrive.py` (26.32%) +- [ ] `app/tasks/upload_to_nextcloud.py` (15.19%) +- [ ] `app/tasks/upload_to_paperless.py` (18.60%) +- [ ] `app/tasks/upload_to_email.py` (36.08%) + +### Complex Background Tasks (0-36%) +- [ ] `app/tasks/check_credentials.py` (0%) + - Requires mocking multiple external services + - Test credential validation for each provider + - Test failure state management + - Test notification system + +- [ ] `app/tasks/imap_tasks.py` (0%) + - Requires IMAP server mocking + - Test email fetching + - Test email parsing + - Test lock management with Redis + +- [ ] `app/tasks/upload_with_rclone.py` (0%) + - Test rclone command execution + - Test configuration management + - Test error handling + +- [ ] `app/tasks/extract_metadata_with_gpt.py` (28.79%) + - Test GPT metadata extraction + - Mock OpenAI API responses + - Test various document types + +### View Routes (25-61%) +- [ ] `app/views/status.py` (25.00%) +- [ ] `app/views/wizard.py` (38.98%) +- [ ] `app/views/settings.py` (42.86%) +- [ ] `app/views/google_drive.py` (42.42%) + +## Testing Strategy + +### For Low-Hanging Fruits (Phase 1) +1. Focus on pure functions with minimal dependencies +2. Mock external services (OpenAI, Azure, cloud storage) +3. Test error paths and edge cases +4. Use pytest fixtures for common setup + +### For Integration Tests (Phases 2-3) +1. Create comprehensive mocks for external services +2. Use pytest-mock for patching +3. Test async functions with pytest-asyncio +4. Use TestClient for API endpoint tests +5. Mock Redis, database, and Celery for task tests + +## Coverage Goals by Phase + +| Phase | Target Coverage | Status | +|-------|----------------|--------| +| Phase 1: Low-Hanging Fruits | 60% | In Progress | +| Phase 2: Medium Priority | 70% | Not Started | +| Phase 3: Complex Integration | 80% | Not Started | + +## Notes + +- Files with 100% coverage: Keep them at 100% +- Files with 90%+ coverage: Low priority for improvement +- Focus on business logic, not simple re-exports +- Mock external dependencies to avoid flaky tests +- All tests must pass CI/CD pipeline +- Maintain test execution time under 2 minutes for fast feedback + +## Files Excluded from Coverage + +These files are infrastructure/configuration and don't require high coverage: +- `migrations/*` - Database migrations (excluded in pytest.ini) +- `app/__init__.py` - Empty init files +- `app/*/__init__.py` - Package init files + +## Running Tests + +```bash +# Run all tests with coverage +pytest --cov=app --cov-report=term-missing + +# Run tests for specific module +pytest tests/test_encryption.py -v + +# Run tests with coverage report +pytest --cov=app --cov-report=html +open htmlcov/index.html + +# Run only unit tests (fast) +pytest -m unit + +# Run integration tests +pytest -m integration +``` + +## Contributing + +When adding new code: +1. Write tests for new functionality +2. Aim for 80%+ coverage on new files +3. Update this TODO when completing test coverage work +4. Run coverage report before submitting PR diff --git a/tests/test_celery_worker.py b/tests/test_celery_worker.py new file mode 100644 index 00000000..c9dd6655 --- /dev/null +++ b/tests/test_celery_worker.py @@ -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) diff --git a/tests/test_encryption.py b/tests/test_encryption.py new file mode 100644 index 00000000..686d8ed8 --- /dev/null +++ b/tests/test_encryption.py @@ -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 diff --git a/tests/test_filename_utils.py b/tests/test_filename_utils.py new file mode 100644 index 00000000..3fb9df7d --- /dev/null +++ b/tests/test_filename_utils.py @@ -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 diff --git a/tests/test_logging_utils.py b/tests/test_logging_utils.py new file mode 100644 index 00000000..81dfbcfb --- /dev/null +++ b/tests/test_logging_utils.py @@ -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() diff --git a/tests/test_reexports.py b/tests/test_reexports.py new file mode 100644 index 00000000..cd118b12 --- /dev/null +++ b/tests/test_reexports.py @@ -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) diff --git a/tests/test_uptime_kuma.py b/tests/test_uptime_kuma.py new file mode 100644 index 00000000..8dee281e --- /dev/null +++ b/tests/test_uptime_kuma.py @@ -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