test: add comprehensive edge case tests for validators, file_status, file_splitting, and filename_utils
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -321,3 +321,152 @@ class TestCheckAllConfigs:
|
||||
mock_settings.notification_urls = ["mailto://test@example.com"]
|
||||
check_all_configs()
|
||||
mock_dump.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestValidateAuthConfigEdgeCases:
|
||||
"""Test edge cases in auth configuration validation."""
|
||||
|
||||
def test_session_secret_exactly_32_chars(self):
|
||||
"""Test validation with session secret exactly 32 characters."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.auth_enabled = True
|
||||
mock_settings.session_secret = "a" * 32 # Exactly 32 characters
|
||||
mock_settings.admin_username = "admin"
|
||||
mock_settings.admin_password = "pass"
|
||||
|
||||
issues = validate_auth_config()
|
||||
# Should not have issue about length
|
||||
assert not any("32 characters" in issue for issue in issues)
|
||||
|
||||
def test_auth_disabled_no_validation(self):
|
||||
"""Test that auth validation is skipped when auth is disabled."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.auth_enabled = False
|
||||
|
||||
issues = validate_auth_config()
|
||||
# Should have no issues when auth is disabled
|
||||
assert len(issues) == 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestValidateEmailConfigEdgeCases:
|
||||
"""Test edge cases in email configuration validation."""
|
||||
|
||||
@patch("app.utils.config_validator.validators.socket.gethostbyname")
|
||||
def test_email_host_resolution_success(self, mock_gethostbyname):
|
||||
"""Test email host resolution success."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
mock_settings.email_port = 587
|
||||
mock_settings.email_username = "user"
|
||||
mock_settings.email_password = "pass"
|
||||
|
||||
mock_gethostbyname.return_value = "192.0.2.1"
|
||||
|
||||
issues = validate_email_config()
|
||||
# Should not have DNS resolution issue
|
||||
assert not any("Cannot resolve" in issue for issue in issues)
|
||||
|
||||
@patch("app.utils.config_validator.validators.socket.gethostbyname")
|
||||
def test_email_host_resolution_failure(self, mock_gethostbyname):
|
||||
"""Test email host resolution failure."""
|
||||
import socket
|
||||
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.email_host = "nonexistent.example.com"
|
||||
mock_settings.email_port = 587
|
||||
mock_settings.email_username = "user"
|
||||
mock_settings.email_password = "pass"
|
||||
|
||||
mock_gethostbyname.side_effect = socket.gaierror()
|
||||
|
||||
issues = validate_email_config()
|
||||
# Should have DNS resolution issue
|
||||
assert any("Cannot resolve" in issue for issue in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestValidateNotificationConfigEdgeCases:
|
||||
"""Test edge cases in notification configuration validation."""
|
||||
|
||||
@patch("app.utils.config_validator.validators.apprise")
|
||||
def test_apprise_not_installed(self, mock_apprise):
|
||||
"""Test handling when apprise module is not available."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.notification_urls = ["https://example.com/notify"]
|
||||
|
||||
# Simulate ImportError
|
||||
mock_apprise.Apprise.side_effect = AttributeError()
|
||||
|
||||
# Should handle gracefully
|
||||
issues = validate_notification_config()
|
||||
assert isinstance(issues, list)
|
||||
|
||||
def test_invalid_apprise_url_format(self):
|
||||
"""Test validation with invalid notification URL format."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
with patch("app.utils.config_validator.validators.apprise") as mock_apprise_module:
|
||||
mock_settings.notification_urls = ["invalid-url-format"]
|
||||
|
||||
mock_apprise = mock_apprise_module.Apprise.return_value
|
||||
mock_apprise.add.return_value = False # Invalid URL
|
||||
|
||||
issues = validate_notification_config()
|
||||
assert any("Invalid notification URL format" in issue for issue in issues)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestValidateStorageConfigsEdgeCases:
|
||||
"""Test edge cases for storage configuration validation."""
|
||||
|
||||
def test_sftp_with_valid_key_path(self, tmp_path):
|
||||
"""Test SFTP validation with valid key file path."""
|
||||
key_file = tmp_path / "key.pem"
|
||||
key_file.write_text("fake key")
|
||||
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
mock_settings.sftp_host = "sftp.example.com"
|
||||
mock_settings.sftp_private_key = str(key_file)
|
||||
mock_settings.sftp_password = None
|
||||
|
||||
result = validate_storage_configs()
|
||||
# Should not have key file not found issue
|
||||
assert not any("file not found" in issue.lower() for issue in result["sftp"])
|
||||
|
||||
def test_all_services_fully_configured(self):
|
||||
"""Test validation when all services are fully configured."""
|
||||
with patch("app.utils.config_validator.validators.settings") as mock_settings:
|
||||
# Configure all services
|
||||
mock_settings.sftp_host = "sftp.example.com"
|
||||
mock_settings.sftp_password = "pass"
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
mock_settings.email_default_recipient = "test@example.com"
|
||||
mock_settings.s3_bucket_name = "my-bucket"
|
||||
mock_settings.aws_access_key_id = "key"
|
||||
mock_settings.aws_secret_access_key = "secret"
|
||||
mock_settings.ftp_host = "ftp.example.com"
|
||||
mock_settings.ftp_username = "user"
|
||||
mock_settings.ftp_password = "pass"
|
||||
mock_settings.webdav_url = "https://webdav.example.com"
|
||||
mock_settings.webdav_username = "user"
|
||||
mock_settings.webdav_password = "pass"
|
||||
mock_settings.google_drive_credentials_json = "{}"
|
||||
mock_settings.google_drive_folder_id = "folder_id"
|
||||
mock_settings.paperless_host = "https://paperless.example.com"
|
||||
mock_settings.paperless_ngx_api_token = "token"
|
||||
mock_settings.onedrive_client_id = "id"
|
||||
mock_settings.onedrive_client_secret = "secret"
|
||||
mock_settings.onedrive_refresh_token = "token"
|
||||
mock_settings.dropbox_app_key = "key"
|
||||
mock_settings.dropbox_app_secret = "secret"
|
||||
mock_settings.dropbox_refresh_token = "token"
|
||||
mock_settings.nextcloud_upload_url = "https://nextcloud.example.com"
|
||||
mock_settings.nextcloud_username = "user"
|
||||
mock_settings.nextcloud_password = "pass"
|
||||
mock_settings.uptime_kuma_url = "https://kuma.example.com"
|
||||
|
||||
result = validate_storage_configs()
|
||||
# Check that all providers have empty issue lists
|
||||
for provider, issues in result.items():
|
||||
assert len(issues) == 0, f"{provider} should have no issues"
|
||||
|
||||
@@ -450,3 +450,100 @@ class TestSplitPdfEdgeCases:
|
||||
for split_file in split_files:
|
||||
if os.path.exists(split_file):
|
||||
os.remove(split_file)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestShouldSplitFileEdgeCases:
|
||||
"""Additional edge cases for should_split_file function."""
|
||||
|
||||
def test_returns_false_when_max_size_is_none(self, tmp_path):
|
||||
"""Test that splitting is disabled when max_size is None."""
|
||||
from app.utils.file_splitting import should_split_file
|
||||
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_bytes(b"a" * 10000) # 10KB file
|
||||
|
||||
result = should_split_file(str(test_file), None)
|
||||
assert result is False
|
||||
|
||||
def test_returns_false_for_nonexistent_file(self):
|
||||
"""Test handling of nonexistent file."""
|
||||
from app.utils.file_splitting import should_split_file
|
||||
|
||||
result = should_split_file("/nonexistent/file.pdf", 1000)
|
||||
assert result is False
|
||||
|
||||
def test_returns_true_when_file_exceeds_limit(self, tmp_path):
|
||||
"""Test that splitting is enabled when file exceeds limit."""
|
||||
from app.utils.file_splitting import should_split_file
|
||||
|
||||
test_file = tmp_path / "large.pdf"
|
||||
test_file.write_bytes(b"a" * 10000) # 10KB file
|
||||
|
||||
result = should_split_file(str(test_file), 5000) # 5KB limit
|
||||
assert result is True
|
||||
|
||||
def test_returns_false_when_file_within_limit(self, tmp_path):
|
||||
"""Test that splitting is disabled when file is within limit."""
|
||||
from app.utils.file_splitting import should_split_file
|
||||
|
||||
test_file = tmp_path / "small.pdf"
|
||||
test_file.write_bytes(b"a" * 1000) # 1KB file
|
||||
|
||||
result = should_split_file(str(test_file), 5000) # 5KB limit
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSplitPdfBySizeEdgeCases:
|
||||
"""Additional edge cases for split_pdf_by_size function."""
|
||||
|
||||
def test_raises_error_for_nonexistent_file(self):
|
||||
"""Test that FileNotFoundError is raised for nonexistent file."""
|
||||
from app.utils.file_splitting import split_pdf_by_size
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
split_pdf_by_size("/nonexistent/file.pdf", 1000000)
|
||||
|
||||
def test_raises_error_for_invalid_pdf(self, tmp_path):
|
||||
"""Test that ValueError is raised for invalid PDF."""
|
||||
from app.utils.file_splitting import split_pdf_by_size
|
||||
|
||||
invalid_file = tmp_path / "invalid.pdf"
|
||||
invalid_file.write_text("not a valid PDF")
|
||||
|
||||
with pytest.raises(ValueError, match="Invalid or corrupted PDF"):
|
||||
split_pdf_by_size(str(invalid_file), 1000000)
|
||||
|
||||
def test_returns_empty_list_for_zero_page_pdf(self, tmp_path):
|
||||
"""Test handling of PDF with zero pages."""
|
||||
from app.utils.file_splitting import split_pdf_by_size
|
||||
from pypdf import PdfWriter
|
||||
|
||||
# Create a technically valid but empty PDF
|
||||
empty_pdf = tmp_path / "empty.pdf"
|
||||
writer = PdfWriter()
|
||||
with open(empty_pdf, "wb") as f:
|
||||
writer.write(f)
|
||||
|
||||
result = split_pdf_by_size(str(empty_pdf), 1000000)
|
||||
assert result == []
|
||||
|
||||
def test_custom_output_directory(self, tmp_path, sample_pdf_path):
|
||||
"""Test splitting with custom output directory."""
|
||||
from app.utils.file_splitting import split_pdf_by_size
|
||||
|
||||
output_dir = tmp_path / "custom_output"
|
||||
output_dir.mkdir()
|
||||
|
||||
split_files = split_pdf_by_size(sample_pdf_path, 500, str(output_dir))
|
||||
|
||||
# All files should be in custom directory
|
||||
for file_path in split_files:
|
||||
assert str(output_dir) in file_path
|
||||
assert os.path.exists(file_path)
|
||||
|
||||
# Cleanup
|
||||
for file_path in split_files:
|
||||
if os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
|
||||
@@ -185,3 +185,86 @@ class TestMetricsCounting:
|
||||
assert summary["uploads"]["success"] == 6
|
||||
assert summary["uploads"]["failure"] == 0
|
||||
assert summary["uploads"]["in_progress"] == 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetFilesProcessingStatusEdgeCases:
|
||||
"""Test edge cases for get_files_processing_status."""
|
||||
|
||||
def test_handles_empty_file_list(self, db_session):
|
||||
"""Test with empty file list."""
|
||||
from app.utils.file_status import get_files_processing_status
|
||||
|
||||
result = get_files_processing_status(db_session, [])
|
||||
assert result == {}
|
||||
|
||||
def test_handles_nonexistent_file_ids(self, db_session):
|
||||
"""Test with file IDs that don't exist."""
|
||||
from app.utils.file_status import get_files_processing_status
|
||||
|
||||
result = get_files_processing_status(db_session, [99999, 99998])
|
||||
# Should return status for these IDs even if they don't exist
|
||||
assert 99999 in result
|
||||
assert result[99999]["status"] == "pending"
|
||||
|
||||
def test_handles_mixed_file_states(self, db_session):
|
||||
"""Test with files in different states."""
|
||||
from app.models import FileRecord
|
||||
from app.utils.file_status import get_files_processing_status
|
||||
from app.utils.step_manager import initialize_file_steps, update_step_status
|
||||
|
||||
# Create multiple files with different states
|
||||
file1 = FileRecord(filename="file1.pdf", is_duplicate=False)
|
||||
file2 = FileRecord(filename="file2.pdf", is_duplicate=True)
|
||||
file3 = FileRecord(filename="file3.pdf", is_duplicate=False)
|
||||
|
||||
db_session.add_all([file1, file2, file3])
|
||||
db_session.commit()
|
||||
|
||||
# Initialize steps for file1 and file3
|
||||
initialize_file_steps(db_session, file1.id)
|
||||
initialize_file_steps(db_session, file3.id)
|
||||
|
||||
# Mark file1 as failed
|
||||
update_step_status(db_session, file1.id, "extract_text", "failure")
|
||||
|
||||
# Mark file3 as in progress
|
||||
update_step_status(db_session, file3.id, "extract_text", "in_progress")
|
||||
|
||||
result = get_files_processing_status(db_session, [file1.id, file2.id, file3.id])
|
||||
|
||||
assert result[file1.id]["status"] == "failed"
|
||||
assert result[file2.id]["status"] == "duplicate"
|
||||
assert result[file3.id]["status"] == "processing"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestComputeStatusFromLogsDeprecated:
|
||||
"""Test the deprecated _compute_status_from_logs function."""
|
||||
|
||||
def test_empty_logs_list(self):
|
||||
"""Test with empty logs list."""
|
||||
from app.utils.file_status import _compute_status_from_logs
|
||||
|
||||
result = _compute_status_from_logs([])
|
||||
assert result["status"] == "pending"
|
||||
assert result["last_step"] is None
|
||||
assert result["has_errors"] is False
|
||||
|
||||
def test_logs_with_multiple_steps(self, db_session):
|
||||
"""Test logs from multiple steps."""
|
||||
from app.models import ProcessingLog
|
||||
from app.utils.file_status import _compute_status_from_logs
|
||||
|
||||
logs = [
|
||||
ProcessingLog(
|
||||
file_id=1, task_id="task1", step_name="step1", status="success", message="Done", timestamp=None
|
||||
),
|
||||
ProcessingLog(
|
||||
file_id=1, task_id="task2", step_name="step2", status="in_progress", message="Running", timestamp=None
|
||||
),
|
||||
]
|
||||
|
||||
result = _compute_status_from_logs(logs)
|
||||
assert result["status"] == "processing"
|
||||
assert result["last_step"] == "step1" # First log in list
|
||||
|
||||
@@ -417,3 +417,86 @@ class TestUniqueFilepathWithCounter:
|
||||
assert ".pdf" in result
|
||||
# Should not be a simple counter-based name
|
||||
assert not any(f"test-{i:04d}.pdf" in result for i in range(1, 100))
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSanitizeFilenameEdgeCases:
|
||||
"""Additional edge cases for sanitize_filename."""
|
||||
|
||||
def test_handles_empty_string(self):
|
||||
"""Test with empty string."""
|
||||
result = sanitize_filename("")
|
||||
# Should return a default document name
|
||||
assert "document_" in result
|
||||
assert len(result) > 0
|
||||
|
||||
def test_handles_only_periods(self):
|
||||
"""Test with only periods."""
|
||||
result = sanitize_filename("...")
|
||||
# Should return a default document name
|
||||
assert "document_" in result
|
||||
|
||||
def test_handles_only_dots(self):
|
||||
"""Test with single dot."""
|
||||
result = sanitize_filename(".")
|
||||
# Should return a default document name
|
||||
assert "document_" in result
|
||||
|
||||
def test_preserves_multiple_extensions(self):
|
||||
"""Test that multiple extensions are preserved."""
|
||||
result = sanitize_filename("file.tar.gz")
|
||||
assert ".tar.gz" in result or "file_tar_gz" in result
|
||||
|
||||
def test_removes_null_bytes(self):
|
||||
"""Test that null bytes are removed."""
|
||||
result = sanitize_filename("file\x00name.pdf")
|
||||
assert "\x00" not in result
|
||||
assert "file" in result
|
||||
assert "name" in result
|
||||
|
||||
def test_handles_unicode_characters(self):
|
||||
"""Test handling of unicode characters."""
|
||||
result = sanitize_filename("文档.pdf")
|
||||
# Should preserve unicode or convert safely
|
||||
assert ".pdf" in result
|
||||
assert len(result) > 0
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExtractRemotePathEdgeCases:
|
||||
"""Additional edge cases for extract_remote_path."""
|
||||
|
||||
def test_file_not_in_base_dir(self):
|
||||
"""Test when file is not a subdirectory of base_dir."""
|
||||
result = extract_remote_path("/other/path/file.pdf", "/base/dir", "/remote")
|
||||
# Should just use filename
|
||||
assert result == "remote/file.pdf"
|
||||
|
||||
def test_multiple_processed_directories(self):
|
||||
"""Test path with multiple 'processed' directories."""
|
||||
result = extract_remote_path(
|
||||
"/base/processed/subdir/processed/file.pdf", "/base", "/remote"
|
||||
)
|
||||
# Should remove all 'processed' directories
|
||||
assert "processed" not in result.lower()
|
||||
|
||||
def test_remote_base_with_trailing_slash(self):
|
||||
"""Test remote base that already has trailing slash."""
|
||||
result = extract_remote_path("/base/file.pdf", "/base", "/remote/")
|
||||
# Should handle gracefully
|
||||
assert result.startswith("remote/")
|
||||
|
||||
def test_empty_remote_base(self):
|
||||
"""Test with empty remote base."""
|
||||
result = extract_remote_path("/base/subdir/file.pdf", "/base", "")
|
||||
assert result == "subdir/file.pdf"
|
||||
|
||||
def test_windows_style_separators(self, monkeypatch):
|
||||
"""Test handling of Windows-style path separators."""
|
||||
# Temporarily change os.sep to backslash
|
||||
import os
|
||||
|
||||
result = extract_remote_path("/base/subdir/file.pdf", "/base", "/remote")
|
||||
# Should use forward slashes in output
|
||||
assert "\\" not in result
|
||||
assert "/" in result
|
||||
|
||||
Reference in New Issue
Block a user