From 80a645895a3939e648bdbb5382b063dc7b6b72ab Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 14 Feb 2026 00:37:32 +0000 Subject: [PATCH] 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> --- tests/test_config_validators.py | 149 ++++++++++++++++++++++++++++++++ tests/test_file_splitting.py | 97 +++++++++++++++++++++ tests/test_file_status_fix.py | 83 ++++++++++++++++++ tests/test_filename_utils.py | 83 ++++++++++++++++++ 4 files changed, 412 insertions(+) diff --git a/tests/test_config_validators.py b/tests/test_config_validators.py index f4e7045c..0e0eb061 100644 --- a/tests/test_config_validators.py +++ b/tests/test_config_validators.py @@ -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" diff --git a/tests/test_file_splitting.py b/tests/test_file_splitting.py index 35952e87..54fc822e 100644 --- a/tests/test_file_splitting.py +++ b/tests/test_file_splitting.py @@ -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) diff --git a/tests/test_file_status_fix.py b/tests/test_file_status_fix.py index 8a6122b0..f3ab92bf 100644 --- a/tests/test_file_status_fix.py +++ b/tests/test_file_status_fix.py @@ -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 diff --git a/tests/test_filename_utils.py b/tests/test_filename_utils.py index eab4c17e..1a7c8e75 100644 --- a/tests/test_filename_utils.py +++ b/tests/test_filename_utils.py @@ -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