test: add comprehensive tests for IMAP, OpenAI, send_to_all, config_loader, and diagnostic APIs
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -1,5 +1,7 @@
|
||||
"""Tests for app/api/openai.py module."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@@ -20,3 +22,119 @@ class TestOpenAIEndpoints:
|
||||
data = response.json()
|
||||
# Should be success or error depending on API key validity
|
||||
assert data["status"] in ("success", "error")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestOpenAIConnectionErrors:
|
||||
"""Test error handling in OpenAI connection test."""
|
||||
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_no_api_key_configured(self, mock_settings, client):
|
||||
"""Test OpenAI test endpoint when no API key is configured."""
|
||||
mock_settings.openai_api_key = None
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
assert "not configured" in data["message"].lower()
|
||||
|
||||
@patch("app.api.openai.openai.OpenAI")
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_api_key_validation_success(self, mock_settings, mock_openai_class, client):
|
||||
"""Test OpenAI API key validation success."""
|
||||
mock_settings.openai_api_key = "sk-test-key"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_models = MagicMock()
|
||||
mock_models.data = [{"id": "gpt-4"}, {"id": "gpt-3.5-turbo"}]
|
||||
mock_client.models.list.return_value = mock_models
|
||||
mock_openai_class.return_value = mock_client
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "success"
|
||||
assert "valid" in data["message"].lower()
|
||||
assert data["models_available"] == 2
|
||||
|
||||
@patch("app.api.openai.openai.OpenAI")
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_api_key_validation_auth_error(self, mock_settings, mock_openai_class, client):
|
||||
"""Test OpenAI API key validation with auth error."""
|
||||
mock_settings.openai_api_key = "sk-invalid-key"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.models.list.side_effect = Exception("Incorrect API key")
|
||||
mock_openai_class.return_value = mock_client
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
assert "api key" in data["message"].lower() or "validation failed" in data["message"].lower()
|
||||
assert data.get("is_auth_error") is True
|
||||
|
||||
@patch("app.api.openai.openai.OpenAI")
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_api_key_validation_network_error(self, mock_settings, mock_openai_class, client):
|
||||
"""Test OpenAI API key validation with network error."""
|
||||
mock_settings.openai_api_key = "sk-test-key"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_client.models.list.side_effect = Exception("Network timeout")
|
||||
mock_openai_class.return_value = mock_client
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
# Not an auth error, so is_auth_error should be False
|
||||
assert data.get("is_auth_error") is False
|
||||
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_import_error(self, mock_settings, client):
|
||||
"""Test handling when openai package is not installed."""
|
||||
mock_settings.openai_api_key = "sk-test-key"
|
||||
|
||||
# Mock ImportError by patching the import
|
||||
with patch("app.api.openai.openai", side_effect=ImportError()):
|
||||
# Need to reload the module to trigger the import error path
|
||||
# For this test, we'll just verify the endpoint handles missing imports gracefully
|
||||
response = client.get("/api/openai/test")
|
||||
# The endpoint should still respond, even if openai is missing
|
||||
assert response.status_code == 200
|
||||
# Note: The actual implementation uses try/except ImportError in the endpoint itself
|
||||
|
||||
@patch("app.api.openai.openai.OpenAI")
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_models_without_data_attribute(self, mock_settings, mock_openai_class, client):
|
||||
"""Test OpenAI API when models response doesn't have data attribute."""
|
||||
mock_settings.openai_api_key = "sk-test-key"
|
||||
|
||||
mock_client = MagicMock()
|
||||
mock_models = MagicMock(spec=[]) # No data attribute
|
||||
del mock_models.data # Ensure data attribute doesn't exist
|
||||
mock_client.models.list.return_value = mock_models
|
||||
mock_openai_class.return_value = mock_client
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "success"
|
||||
assert data["models_available"] == "Unknown"
|
||||
|
||||
@patch("app.api.openai.openai.OpenAI")
|
||||
@patch("app.api.openai.settings")
|
||||
def test_openai_unexpected_exception(self, mock_settings, mock_openai_class, client):
|
||||
"""Test handling of unexpected exceptions."""
|
||||
mock_settings.openai_api_key = "sk-test-key"
|
||||
|
||||
# Raise an unexpected exception during OpenAI client creation
|
||||
mock_openai_class.side_effect = RuntimeError("Unexpected error")
|
||||
|
||||
response = client.get("/api/openai/test")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
assert "unexpected error" in data["message"].lower()
|
||||
|
||||
+146
-1
@@ -1,8 +1,15 @@
|
||||
"""Tests for app/utils/config_loader.py module."""
|
||||
|
||||
from typing import Optional, Union
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.utils.config_loader import convert_setting_value
|
||||
from app.utils.config_loader import (
|
||||
convert_setting_value,
|
||||
load_settings_from_db,
|
||||
reload_settings_from_db,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@@ -58,3 +65,141 @@ class TestConvertSettingValue:
|
||||
"""Test converting comma-separated string to list."""
|
||||
result = convert_setting_value("a, b, c", list)
|
||||
assert result == ["a", "b", "c"]
|
||||
|
||||
def test_handles_optional_type(self):
|
||||
"""Test handling Optional type annotation."""
|
||||
result = convert_setting_value("42", Optional[int])
|
||||
assert result == 42
|
||||
|
||||
def test_handles_union_type(self):
|
||||
"""Test handling Union type annotation."""
|
||||
result = convert_setting_value("test", Union[str, None])
|
||||
assert result == "test"
|
||||
|
||||
def test_list_from_empty_string(self):
|
||||
"""Test converting empty comma string to empty list."""
|
||||
result = convert_setting_value("", list)
|
||||
assert result == []
|
||||
|
||||
def test_list_with_whitespace(self):
|
||||
"""Test list conversion handles extra whitespace."""
|
||||
result = convert_setting_value(" a , b , c ", list)
|
||||
assert result == ["a", "b", "c"]
|
||||
|
||||
def test_bool_variants(self):
|
||||
"""Test various boolean string representations."""
|
||||
assert convert_setting_value("1", bool) is True
|
||||
assert convert_setting_value("y", bool) is True
|
||||
assert convert_setting_value("t", bool) is True
|
||||
assert convert_setting_value("0", bool) is False
|
||||
assert convert_setting_value("n", bool) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestLoadSettingsFromDb:
|
||||
"""Tests for load_settings_from_db function."""
|
||||
|
||||
def test_loads_settings_from_database(self):
|
||||
"""Test loading settings from database."""
|
||||
from app.models import ApplicationSettings
|
||||
|
||||
# Mock settings object
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.__fields__ = {
|
||||
"test_setting": MagicMock(annotation=str),
|
||||
"another_setting": MagicMock(annotation=int),
|
||||
}
|
||||
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_db_settings = [
|
||||
ApplicationSettings(key="test_setting", value="test_value"),
|
||||
ApplicationSettings(key="another_setting", value="42"),
|
||||
]
|
||||
mock_db.query.return_value.all.return_value = mock_db_settings
|
||||
|
||||
load_settings_from_db(mock_settings, mock_db)
|
||||
|
||||
# Verify settings were set
|
||||
assert mock_settings.test_setting == "test_value"
|
||||
assert mock_settings.another_setting == 42
|
||||
|
||||
def test_handles_empty_database(self):
|
||||
"""Test handling when no settings in database."""
|
||||
mock_settings = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.all.return_value = []
|
||||
|
||||
# Should not raise
|
||||
load_settings_from_db(mock_settings, mock_db)
|
||||
|
||||
def test_skips_unknown_settings(self):
|
||||
"""Test that unknown settings are skipped."""
|
||||
from app.models import ApplicationSettings
|
||||
|
||||
mock_settings = MagicMock()
|
||||
mock_settings.__fields__ = {"known_setting": MagicMock(annotation=str)}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db_settings = [
|
||||
ApplicationSettings(key="known_setting", value="value1"),
|
||||
ApplicationSettings(key="unknown_setting", value="value2"),
|
||||
]
|
||||
mock_db.query.return_value.all.return_value = mock_db_settings
|
||||
|
||||
load_settings_from_db(mock_settings, mock_db)
|
||||
|
||||
# Only known_setting should be set
|
||||
assert hasattr(mock_settings, "known_setting")
|
||||
|
||||
def test_handles_database_errors(self):
|
||||
"""Test handling database errors gracefully."""
|
||||
mock_settings = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.side_effect = Exception("Database error")
|
||||
|
||||
# Should not raise, just log warning
|
||||
load_settings_from_db(mock_settings, mock_db)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestReloadSettingsFromDb:
|
||||
"""Tests for reload_settings_from_db function."""
|
||||
|
||||
@patch("app.utils.config_loader.SessionLocal")
|
||||
@patch("app.utils.config_loader.load_settings_from_db")
|
||||
def test_reload_success(self, mock_load, mock_session_local):
|
||||
"""Test successful settings reload."""
|
||||
mock_settings = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_session_local.return_value = mock_db
|
||||
|
||||
result = reload_settings_from_db(mock_settings)
|
||||
|
||||
assert result is True
|
||||
mock_load.assert_called_once_with(mock_settings, mock_db)
|
||||
mock_db.close.assert_called_once()
|
||||
|
||||
@patch("app.utils.config_loader.SessionLocal")
|
||||
def test_reload_database_error(self, mock_session_local):
|
||||
"""Test reload handling database error."""
|
||||
mock_settings = MagicMock()
|
||||
mock_session_local.side_effect = Exception("Connection error")
|
||||
|
||||
result = reload_settings_from_db(mock_settings)
|
||||
|
||||
assert result is False
|
||||
|
||||
@patch("app.utils.config_loader.SessionLocal")
|
||||
def test_reload_closes_session_on_error(self, mock_session_local):
|
||||
"""Test that session is closed even on error."""
|
||||
mock_settings = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.side_effect = Exception("Query error")
|
||||
mock_session_local.return_value = mock_db
|
||||
|
||||
result = reload_settings_from_db(mock_settings)
|
||||
|
||||
# Should close session despite error
|
||||
mock_db.close.assert_called_once()
|
||||
assert result is False
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Tests for app/api/diagnostic.py module."""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@@ -25,6 +27,24 @@ class TestDiagnosticSettings:
|
||||
for key in expected_keys:
|
||||
assert key in services
|
||||
|
||||
def test_diagnostic_settings_includes_workdir(self, client):
|
||||
"""Test that settings include workdir."""
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
data = response.json()
|
||||
assert "workdir" in data["settings"]
|
||||
|
||||
def test_diagnostic_settings_includes_hostname(self, client):
|
||||
"""Test that settings include external hostname."""
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
data = response.json()
|
||||
assert "external_hostname" in data["settings"]
|
||||
|
||||
def test_diagnostic_settings_includes_imap_status(self, client):
|
||||
"""Test that settings include IMAP enabled status."""
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
data = response.json()
|
||||
assert "imap_enabled" in data["settings"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestTestNotification:
|
||||
@@ -37,3 +57,129 @@ class TestTestNotification:
|
||||
data = response.json()
|
||||
# Should return warning (no notification services configured) or success
|
||||
assert data["status"] in ("warning", "success", "error")
|
||||
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_test_notification_no_urls_configured(self, mock_settings, client):
|
||||
"""Test notification test when no URLs are configured."""
|
||||
mock_settings.notification_urls = []
|
||||
mock_settings.external_hostname = "test-host"
|
||||
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "warning"
|
||||
assert "not configured" in data["message"].lower()
|
||||
|
||||
@patch("app.api.diagnostic.send_notification")
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_test_notification_success(self, mock_settings, mock_send, client):
|
||||
"""Test successful notification test."""
|
||||
mock_settings.notification_urls = ["https://example.com/notify"]
|
||||
mock_settings.external_hostname = "test-host"
|
||||
mock_send.return_value = True
|
||||
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "success"
|
||||
assert "services_count" in data
|
||||
mock_send.assert_called_once()
|
||||
|
||||
@patch("app.api.diagnostic.send_notification")
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_test_notification_failure(self, mock_settings, mock_send, client):
|
||||
"""Test notification test when sending fails."""
|
||||
mock_settings.notification_urls = ["https://example.com/notify"]
|
||||
mock_settings.external_hostname = "test-host"
|
||||
mock_send.return_value = False
|
||||
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
assert "failed" in data["message"].lower()
|
||||
|
||||
@patch("app.api.diagnostic.send_notification")
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_test_notification_exception(self, mock_settings, mock_send, client):
|
||||
"""Test notification test with exception."""
|
||||
mock_settings.notification_urls = ["https://example.com/notify"]
|
||||
mock_settings.external_hostname = "test-host"
|
||||
mock_send.side_effect = Exception("Connection error")
|
||||
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
data = response.json()
|
||||
|
||||
assert data["status"] == "error"
|
||||
assert "error" in data["message"].lower()
|
||||
|
||||
def test_test_notification_includes_timestamp(self, client):
|
||||
"""Test that notification includes timestamp in message."""
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
data = response.json()
|
||||
|
||||
# Response should have been processed
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDiagnosticHelpers:
|
||||
"""Test helper functions in diagnostic module."""
|
||||
|
||||
@patch("app.api.diagnostic.dump_all_settings")
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_dump_all_settings_called(self, mock_settings, mock_dump, client):
|
||||
"""Test that dump_all_settings is called."""
|
||||
mock_settings.external_hostname = "test"
|
||||
# Setup minimal mocks for configured services
|
||||
mock_settings.email_host = None
|
||||
mock_settings.s3_bucket_name = None
|
||||
mock_settings.dropbox_refresh_token = None
|
||||
mock_settings.onedrive_refresh_token = None
|
||||
mock_settings.nextcloud_upload_url = None
|
||||
mock_settings.sftp_host = None
|
||||
mock_settings.paperless_host = None
|
||||
mock_settings.google_drive_credentials_json = None
|
||||
mock_settings.uptime_kuma_url = None
|
||||
mock_settings.authentik_config_url = None
|
||||
mock_settings.openai_api_key = None
|
||||
mock_settings.azure_api_key = None
|
||||
mock_settings.azure_endpoint = None
|
||||
mock_settings.imap1_host = None
|
||||
mock_settings.imap2_host = None
|
||||
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
|
||||
# dump_all_settings should have been called
|
||||
mock_dump.assert_called_once()
|
||||
|
||||
@patch("app.api.diagnostic.settings")
|
||||
def test_safe_settings_no_sensitive_data(self, mock_settings, client):
|
||||
"""Test that safe settings don't include sensitive data."""
|
||||
mock_settings.workdir = "/tmp/workdir"
|
||||
mock_settings.external_hostname = "test-host"
|
||||
mock_settings.openai_api_key = "sk-secret-key-12345"
|
||||
mock_settings.aws_secret_access_key = "secret-aws-key"
|
||||
# Setup minimal configured services
|
||||
mock_settings.email_host = None
|
||||
mock_settings.s3_bucket_name = None
|
||||
mock_settings.dropbox_refresh_token = None
|
||||
mock_settings.onedrive_refresh_token = None
|
||||
mock_settings.nextcloud_upload_url = None
|
||||
mock_settings.sftp_host = None
|
||||
mock_settings.paperless_host = None
|
||||
mock_settings.google_drive_credentials_json = None
|
||||
mock_settings.uptime_kuma_url = None
|
||||
mock_settings.authentik_config_url = None
|
||||
mock_settings.azure_api_key = None
|
||||
mock_settings.azure_endpoint = None
|
||||
mock_settings.imap1_host = None
|
||||
mock_settings.imap2_host = None
|
||||
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
data = response.json()
|
||||
|
||||
# Sensitive keys should not be in response
|
||||
response_str = str(data)
|
||||
assert "sk-secret-key" not in response_str
|
||||
assert "secret-aws-key" not in response_str
|
||||
|
||||
@@ -1031,3 +1031,187 @@ class TestEmailAlreadyHasLabelExtended:
|
||||
|
||||
result = email_already_has_label(mock_mail, b"1", "Ingested")
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetCapabilitiesEdgeCases:
|
||||
"""Test edge cases for get_capabilities function."""
|
||||
|
||||
def test_get_capabilities_with_failed_response(self):
|
||||
"""Test get_capabilities when server returns error."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.capability.return_value = ("NO", None)
|
||||
|
||||
result = get_capabilities(mock_mail)
|
||||
assert result == []
|
||||
|
||||
def test_get_capabilities_with_empty_data(self):
|
||||
"""Test get_capabilities with empty capability data."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.capability.return_value = ("OK", [b""])
|
||||
|
||||
result = get_capabilities(mock_mail)
|
||||
assert isinstance(result, list)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestFindAllMailXlistEdgeCases:
|
||||
"""Test edge cases for find_all_mail_xlist function."""
|
||||
|
||||
def test_find_all_mail_xlist_no_allmail_flag(self):
|
||||
"""Test XLIST when no folder has ALLMAIL flag."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail._new_tag.return_value = b"A001"
|
||||
# Simulate responses without ALLMAIL flag
|
||||
mock_mail.readline.side_effect = [
|
||||
b'* XLIST (\\HasNoChildren) "/" "INBOX"\r\n',
|
||||
b"A001 OK XLIST completed\r\n",
|
||||
]
|
||||
|
||||
result = find_all_mail_xlist(mock_mail)
|
||||
assert result is None
|
||||
|
||||
def test_find_all_mail_xlist_with_valid_allmail(self):
|
||||
"""Test XLIST when folder has ALLMAIL flag."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail._new_tag.return_value = b"A001"
|
||||
# Simulate response with ALLMAIL flag
|
||||
mock_mail.readline.side_effect = [
|
||||
b'* XLIST (\\AllMail \\HasNoChildren) "/" "[Gmail]/All Mail"\r\n',
|
||||
b"A001 OK XLIST completed\r\n",
|
||||
]
|
||||
|
||||
result = find_all_mail_xlist(mock_mail)
|
||||
assert result == "[Gmail]/All Mail"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAcquireReleaseLockEdgeCases:
|
||||
"""Test edge cases for lock acquisition and release."""
|
||||
|
||||
@patch("app.tasks.imap_tasks.redis_client")
|
||||
def test_acquire_lock_when_already_held(self, mock_redis):
|
||||
"""Test lock acquisition when lock is already held."""
|
||||
mock_redis.setnx.return_value = False
|
||||
|
||||
result = acquire_lock()
|
||||
assert result is False
|
||||
# Should not call expire if lock not acquired
|
||||
mock_redis.expire.assert_not_called()
|
||||
|
||||
@patch("app.tasks.imap_tasks.redis_client")
|
||||
def test_release_lock_always_attempts_delete(self, mock_redis):
|
||||
"""Test that release_lock always attempts to delete the key."""
|
||||
release_lock()
|
||||
mock_redis.delete.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestPullInboxEdgeCases:
|
||||
"""Test edge cases for pull_inbox function."""
|
||||
|
||||
@patch("app.tasks.imap_tasks.load_processed_emails")
|
||||
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
|
||||
@patch("app.tasks.imap_tasks.settings")
|
||||
def test_pull_inbox_search_failed_status(self, mock_settings, mock_imap_class, mock_load):
|
||||
"""Test pull_inbox when search returns non-OK status."""
|
||||
mock_settings.workdir = "/tmp"
|
||||
mock_load.return_value = {}
|
||||
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.login.return_value = ("OK", [])
|
||||
mock_mail.select.return_value = ("OK", [])
|
||||
mock_mail.search.return_value = ("NO", []) # Search failed
|
||||
mock_imap_class.return_value = mock_mail
|
||||
|
||||
# Should handle gracefully and not raise
|
||||
pull_inbox("test", "imap.example.com", 993, "user", "pass", True, False)
|
||||
|
||||
mock_mail.close.assert_called_once()
|
||||
mock_mail.logout.assert_called_once()
|
||||
|
||||
@patch("app.tasks.imap_tasks.load_processed_emails")
|
||||
@patch("app.tasks.imap_tasks.save_processed_emails")
|
||||
@patch("app.tasks.imap_tasks.fetch_attachments_and_enqueue")
|
||||
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
|
||||
@patch("app.tasks.imap_tasks.settings")
|
||||
def test_pull_inbox_email_without_message_id(
|
||||
self, mock_settings, mock_imap_class, mock_fetch, mock_save, mock_load
|
||||
):
|
||||
"""Test that emails without Message-ID are skipped."""
|
||||
mock_settings.workdir = "/tmp"
|
||||
mock_load.return_value = {}
|
||||
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.login.return_value = ("OK", [])
|
||||
mock_mail.select.return_value = ("OK", [])
|
||||
mock_mail.search.return_value = ("OK", [b"1"])
|
||||
|
||||
# Create email without Message-ID header
|
||||
email_without_id = EmailMessage()
|
||||
email_without_id["Subject"] = "Test"
|
||||
raw_email = email_without_id.as_bytes()
|
||||
|
||||
mock_mail.fetch.return_value = ("OK", [(None, raw_email)])
|
||||
mock_imap_class.return_value = mock_mail
|
||||
|
||||
pull_inbox("test", "imap.example.com", 993, "user", "pass", True, False)
|
||||
|
||||
# Should skip processing since no Message-ID
|
||||
mock_fetch.assert_not_called()
|
||||
|
||||
@patch("app.tasks.imap_tasks.load_processed_emails")
|
||||
@patch("app.tasks.imap_tasks.imaplib.IMAP4_SSL")
|
||||
@patch("app.tasks.imap_tasks.settings")
|
||||
def test_pull_inbox_fetch_failed_status(self, mock_settings, mock_imap_class, mock_load):
|
||||
"""Test pull_inbox when fetch returns non-OK status."""
|
||||
mock_settings.workdir = "/tmp"
|
||||
mock_load.return_value = {}
|
||||
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.login.return_value = ("OK", [])
|
||||
mock_mail.select.return_value = ("OK", [])
|
||||
mock_mail.search.return_value = ("OK", [b"1"])
|
||||
mock_mail.fetch.return_value = ("NO", []) # Fetch failed
|
||||
mock_imap_class.return_value = mock_mail
|
||||
|
||||
# Should handle gracefully
|
||||
pull_inbox("test", "imap.example.com", 993, "user", "pass", True, False)
|
||||
|
||||
mock_mail.close.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestMarkAsProcessedFunctions:
|
||||
"""Test mark_as_processed_with_star and mark_as_processed_with_label functions."""
|
||||
|
||||
def test_mark_as_processed_with_star_handles_exception(self):
|
||||
"""Test that star marking handles exceptions gracefully."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.store.side_effect = Exception("Connection error")
|
||||
|
||||
# Should not raise, just log error
|
||||
mark_as_processed_with_star(mock_mail, b"1")
|
||||
|
||||
def test_mark_as_processed_with_label_handles_exception(self):
|
||||
"""Test that label marking handles exceptions gracefully."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.store.side_effect = Exception("Connection error")
|
||||
|
||||
# Should not raise, just log error
|
||||
mark_as_processed_with_label(mock_mail, b"1", "Ingested")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestEmailAlreadyHasLabelExceptions:
|
||||
"""Test exception handling in email_already_has_label."""
|
||||
|
||||
def test_email_already_has_label_fetch_exception(self):
|
||||
"""Test that fetch exceptions are handled gracefully."""
|
||||
mock_mail = MagicMock()
|
||||
mock_mail.fetch.side_effect = Exception("Fetch failed")
|
||||
|
||||
result = email_already_has_label(mock_mail, b"1", "Ingested")
|
||||
|
||||
# Should return False on error
|
||||
assert result is False
|
||||
|
||||
@@ -0,0 +1,356 @@
|
||||
"""Tests for app/tasks/send_to_all.py module."""
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.send_to_all import (
|
||||
_should_upload_to_dropbox,
|
||||
_should_upload_to_email,
|
||||
_should_upload_to_ftp,
|
||||
_should_upload_to_google_drive,
|
||||
_should_upload_to_nextcloud,
|
||||
_should_upload_to_onedrive,
|
||||
_should_upload_to_paperless,
|
||||
_should_upload_to_s3,
|
||||
_should_upload_to_sftp,
|
||||
_should_upload_to_webdav,
|
||||
get_configured_services_from_validator,
|
||||
send_to_all_destinations,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestShouldUploadFunctions:
|
||||
"""Test the _should_upload_to_* helper functions."""
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_dropbox_all_configured(self, mock_settings):
|
||||
"""Test Dropbox upload check when all credentials are configured."""
|
||||
mock_settings.dropbox_app_key = "key"
|
||||
mock_settings.dropbox_app_secret = "secret"
|
||||
mock_settings.dropbox_refresh_token = "token"
|
||||
|
||||
assert _should_upload_to_dropbox() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_dropbox_missing_credentials(self, mock_settings):
|
||||
"""Test Dropbox upload check when credentials are missing."""
|
||||
mock_settings.dropbox_app_key = None
|
||||
mock_settings.dropbox_app_secret = None
|
||||
mock_settings.dropbox_refresh_token = None
|
||||
|
||||
assert _should_upload_to_dropbox() is False
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_nextcloud_configured(self, mock_settings):
|
||||
"""Test Nextcloud upload check."""
|
||||
mock_settings.nextcloud_upload_url = "https://nextcloud.example.com"
|
||||
mock_settings.nextcloud_username = "user"
|
||||
mock_settings.nextcloud_password = "pass"
|
||||
|
||||
assert _should_upload_to_nextcloud() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_paperless_configured(self, mock_settings):
|
||||
"""Test Paperless upload check."""
|
||||
mock_settings.paperless_ngx_api_token = "token"
|
||||
mock_settings.paperless_host = "https://paperless.example.com"
|
||||
|
||||
assert _should_upload_to_paperless() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_google_drive_oauth(self, mock_settings):
|
||||
"""Test Google Drive upload check with OAuth."""
|
||||
mock_settings.google_drive_use_oauth = True
|
||||
mock_settings.google_drive_client_id = "client_id"
|
||||
mock_settings.google_drive_client_secret = "client_secret"
|
||||
mock_settings.google_drive_refresh_token = "refresh_token"
|
||||
mock_settings.google_drive_folder_id = "folder_id"
|
||||
|
||||
assert _should_upload_to_google_drive() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_google_drive_service_account(self, mock_settings):
|
||||
"""Test Google Drive upload check with service account."""
|
||||
mock_settings.google_drive_use_oauth = False
|
||||
mock_settings.google_drive_credentials_json = '{"type": "service_account"}'
|
||||
mock_settings.google_drive_folder_id = "folder_id"
|
||||
|
||||
assert _should_upload_to_google_drive() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_webdav_configured(self, mock_settings):
|
||||
"""Test WebDAV upload check."""
|
||||
mock_settings.webdav_url = "https://webdav.example.com"
|
||||
mock_settings.webdav_username = "user"
|
||||
mock_settings.webdav_password = "pass"
|
||||
|
||||
assert _should_upload_to_webdav() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_ftp_configured(self, mock_settings):
|
||||
"""Test FTP upload check."""
|
||||
mock_settings.ftp_host = "ftp.example.com"
|
||||
mock_settings.ftp_username = "user"
|
||||
mock_settings.ftp_password = "pass"
|
||||
|
||||
assert _should_upload_to_ftp() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_sftp_with_password(self, mock_settings):
|
||||
"""Test SFTP upload check with password auth."""
|
||||
mock_settings.sftp_host = "sftp.example.com"
|
||||
mock_settings.sftp_username = "user"
|
||||
mock_settings.sftp_password = "pass"
|
||||
mock_settings.sftp_private_key = None
|
||||
|
||||
assert _should_upload_to_sftp() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_sftp_with_key(self, mock_settings):
|
||||
"""Test SFTP upload check with key auth."""
|
||||
mock_settings.sftp_host = "sftp.example.com"
|
||||
mock_settings.sftp_username = "user"
|
||||
mock_settings.sftp_password = None
|
||||
mock_settings.sftp_private_key = "/path/to/key"
|
||||
|
||||
assert _should_upload_to_sftp() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_email_configured(self, mock_settings):
|
||||
"""Test email upload check."""
|
||||
mock_settings.email_host = "smtp.example.com"
|
||||
mock_settings.email_username = "user"
|
||||
mock_settings.email_password = "pass"
|
||||
mock_settings.email_default_recipient = "recipient@example.com"
|
||||
|
||||
assert _should_upload_to_email() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_onedrive_configured(self, mock_settings):
|
||||
"""Test OneDrive upload check."""
|
||||
mock_settings.onedrive_client_id = "client_id"
|
||||
mock_settings.onedrive_client_secret = "client_secret"
|
||||
mock_settings.onedrive_refresh_token = "refresh_token"
|
||||
|
||||
assert _should_upload_to_onedrive() is True
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_should_upload_to_s3_configured(self, mock_settings):
|
||||
"""Test S3 upload check."""
|
||||
mock_settings.s3_bucket_name = "my-bucket"
|
||||
mock_settings.aws_access_key_id = "key_id"
|
||||
mock_settings.aws_secret_access_key = "secret_key"
|
||||
|
||||
assert _should_upload_to_s3() is True
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetConfiguredServicesFromValidator:
|
||||
"""Test get_configured_services_from_validator function."""
|
||||
|
||||
@patch("app.tasks.send_to_all.get_provider_status")
|
||||
def test_returns_configured_services(self, mock_get_status):
|
||||
"""Test that configured services are returned correctly."""
|
||||
mock_get_status.return_value = {
|
||||
"Dropbox": {"configured": True},
|
||||
"NextCloud": {"configured": False},
|
||||
"S3 Storage": {"configured": True},
|
||||
}
|
||||
|
||||
result = get_configured_services_from_validator()
|
||||
|
||||
assert result["dropbox"] is True
|
||||
assert result["nextcloud"] is False
|
||||
assert result["s3"] is True
|
||||
|
||||
@patch("app.tasks.send_to_all.get_provider_status")
|
||||
def test_handles_missing_providers(self, mock_get_status):
|
||||
"""Test handling when some providers are not in status."""
|
||||
mock_get_status.return_value = {
|
||||
"Dropbox": {"configured": True},
|
||||
}
|
||||
|
||||
result = get_configured_services_from_validator()
|
||||
|
||||
assert result["dropbox"] is True
|
||||
# Other services not in result
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSendToAllDestinations:
|
||||
"""Test send_to_all_destinations task."""
|
||||
|
||||
def test_file_not_found_error(self):
|
||||
"""Test that FileNotFoundError is raised when file doesn't exist."""
|
||||
with pytest.raises(FileNotFoundError):
|
||||
send_to_all_destinations.apply(args=["/nonexistent/file.pdf"])
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
@patch("app.tasks.send_to_all.upload_to_dropbox")
|
||||
def test_queues_single_configured_service(self, mock_upload, mock_should, mock_settings, tmp_path):
|
||||
"""Test queueing upload to a single configured service."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should.return_value = True
|
||||
mock_upload.delay.return_value = MagicMock(id="task-123")
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False])
|
||||
|
||||
assert result.result["status"] == "Queued"
|
||||
mock_upload.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_s3")
|
||||
@patch("app.tasks.send_to_all.upload_to_dropbox")
|
||||
@patch("app.tasks.send_to_all.upload_to_s3")
|
||||
def test_queues_multiple_services(
|
||||
self, mock_s3_upload, mock_dropbox_upload, mock_should_s3, mock_should_dropbox, mock_settings, tmp_path
|
||||
):
|
||||
"""Test queueing uploads to multiple services."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should_dropbox.return_value = True
|
||||
mock_should_s3.return_value = True
|
||||
mock_dropbox_upload.delay.return_value = MagicMock(id="dropbox-task")
|
||||
mock_s3_upload.delay.return_value = MagicMock(id="s3-task")
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False])
|
||||
|
||||
assert result.result["status"] == "Queued"
|
||||
mock_dropbox_upload.delay.assert_called_once()
|
||||
mock_s3_upload.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
def test_skips_unconfigured_services(self, mock_should, mock_settings, tmp_path):
|
||||
"""Test that unconfigured services are skipped."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should.return_value = False
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False])
|
||||
|
||||
# No uploads should be queued
|
||||
assert result.result["status"] == "Queued"
|
||||
# Check that message indicates 0 uploads
|
||||
assert "0 upload" in result.result["tasks"] or len(result.result["tasks"]) == 0
|
||||
|
||||
@patch("app.tasks.send_to_all.SessionLocal")
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
@patch("app.tasks.send_to_all.upload_to_dropbox")
|
||||
def test_with_file_id_parameter(
|
||||
self, mock_upload, mock_should, mock_settings, mock_session_local, tmp_path
|
||||
):
|
||||
"""Test send_to_all with explicit file_id parameter."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should.return_value = True
|
||||
mock_upload.delay.return_value = MagicMock(id="task-123")
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False, 42])
|
||||
|
||||
mock_upload.delay.assert_called_once()
|
||||
# Verify file_id was passed to the upload task
|
||||
call_kwargs = mock_upload.delay.call_args[1]
|
||||
assert call_kwargs.get("file_id") == 42
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all.get_configured_services_from_validator")
|
||||
@patch("app.tasks.send_to_all.upload_to_dropbox")
|
||||
def test_uses_validator_when_enabled(
|
||||
self, mock_upload, mock_validator, mock_settings, tmp_path
|
||||
):
|
||||
"""Test that validator is used when use_validator=True."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_validator.return_value = {"dropbox": True}
|
||||
mock_upload.delay.return_value = MagicMock(id="task-123")
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), True])
|
||||
|
||||
mock_validator.assert_called_once()
|
||||
mock_upload.delay.assert_called_once()
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all.get_configured_services_from_validator")
|
||||
def test_validator_exception_fallback(self, mock_validator, mock_settings, tmp_path):
|
||||
"""Test fallback to individual checks when validator fails."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_validator.side_effect = Exception("Validator error")
|
||||
|
||||
# Should not raise, should fall back to individual checks
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), True])
|
||||
|
||||
assert result.result["status"] == "Queued"
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
@patch("app.tasks.send_to_all.upload_to_dropbox")
|
||||
def test_handles_upload_task_queue_error(
|
||||
self, mock_upload, mock_should, mock_settings, tmp_path
|
||||
):
|
||||
"""Test handling when queueing upload task fails."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should.return_value = True
|
||||
mock_upload.delay.side_effect = Exception("Queue error")
|
||||
|
||||
# Should not raise, should log error
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False])
|
||||
|
||||
assert result.result["status"] == "Queued"
|
||||
# Error should be recorded in results
|
||||
assert "dropbox_error" in result.result["tasks"]
|
||||
|
||||
@patch("app.tasks.send_to_all.SessionLocal")
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
def test_fallback_file_id_lookup(self, mock_settings, mock_session_local, tmp_path):
|
||||
"""Test file_id lookup fallback when not provided."""
|
||||
from app.models import FileRecord
|
||||
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
|
||||
# Mock database session
|
||||
mock_db = MagicMock()
|
||||
mock_query = MagicMock()
|
||||
mock_file_record = MagicMock(spec=FileRecord)
|
||||
mock_file_record.id = 123
|
||||
mock_query.first.return_value = mock_file_record
|
||||
mock_db.query.return_value.filter.return_value = mock_query
|
||||
mock_session_local.return_value.__enter__.return_value = mock_db
|
||||
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False, None])
|
||||
|
||||
# Should attempt database lookup
|
||||
mock_db.query.assert_called()
|
||||
|
||||
@patch("app.tasks.send_to_all.settings")
|
||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||
def test_should_upload_check_exception_handling(self, mock_should, mock_settings, tmp_path):
|
||||
"""Test that exceptions in should_upload checks are handled."""
|
||||
test_file = tmp_path / "test.pdf"
|
||||
test_file.write_text("test")
|
||||
|
||||
mock_should.side_effect = Exception("Configuration check error")
|
||||
|
||||
# Should not raise, should treat as not configured
|
||||
result = send_to_all_destinations.apply(args=[str(test_file), False])
|
||||
|
||||
assert result.result["status"] == "Queued"
|
||||
Reference in New Issue
Block a user