Merge pull request #295 from christianlouis/copilot/increase-test-coverage-95-percent
Add test coverage for critical modules (config_validator, settings API, upload tasks, notifications)
This commit is contained in:
Christian Krakau-Louis
@@ -0,0 +1,629 @@
|
||||
"""
|
||||
Tests for improving coverage on config_validator, settings API, license routes,
|
||||
diagnostic API, and OpenAI API endpoints.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.api.settings import require_admin
|
||||
from app.main import app as fastapi_app
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _override_admin():
|
||||
"""Dependency override that simulates an admin user."""
|
||||
return {"is_admin": True, "name": "admin"}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 1. app/utils/config_validator.py (backward-compatible re-export wrapper)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestConfigValidatorReExports:
|
||||
"""Verify the backward-compatible wrapper re-exports all expected symbols."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_imports_from_wrapper(self):
|
||||
"""All public names are importable from the wrapper module."""
|
||||
from app.utils import config_validator as cv
|
||||
|
||||
assert callable(cv.mask_sensitive_value)
|
||||
assert callable(cv.get_provider_status)
|
||||
assert callable(cv.dump_all_settings)
|
||||
assert callable(cv.get_settings_for_display)
|
||||
assert callable(cv.validate_email_config)
|
||||
assert callable(cv.validate_storage_configs)
|
||||
assert callable(cv.validate_notification_config)
|
||||
assert callable(cv.validate_auth_config)
|
||||
assert callable(cv.check_all_configs)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_all_list_matches_expected_exports(self):
|
||||
"""__all__ contains exactly the expected names."""
|
||||
from app.utils import config_validator as cv
|
||||
|
||||
expected = {
|
||||
"validate_email_config",
|
||||
"validate_storage_configs",
|
||||
"validate_notification_config",
|
||||
"validate_auth_config",
|
||||
"mask_sensitive_value",
|
||||
"get_provider_status",
|
||||
"get_settings_for_display",
|
||||
"dump_all_settings",
|
||||
"check_all_configs",
|
||||
}
|
||||
assert set(cv.__all__) == expected
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mask_sensitive_value_callable(self):
|
||||
"""mask_sensitive_value from wrapper returns a result."""
|
||||
from app.utils.config_validator import mask_sensitive_value
|
||||
|
||||
result = mask_sensitive_value("secret-token-12345")
|
||||
assert isinstance(result, str)
|
||||
# Should mask part of the value
|
||||
assert result != "secret-token-12345"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_provider_status_returns_dict(self):
|
||||
"""get_provider_status returns a dictionary."""
|
||||
from app.utils.config_validator import get_provider_status
|
||||
|
||||
result = get_provider_status()
|
||||
assert isinstance(result, dict)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_check_all_configs_returns_dict(self):
|
||||
"""check_all_configs returns a dictionary of validation results."""
|
||||
from app.utils.config_validator import check_all_configs
|
||||
|
||||
result = check_all_configs()
|
||||
assert isinstance(result, dict)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. app/api/settings.py (admin-only settings CRUD)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRequireAdminDependency:
|
||||
"""Tests for the require_admin dependency itself."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_raises_when_no_session_user(self):
|
||||
"""require_admin raises 403 when there is no user in session."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = None
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
require_admin(mock_request)
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_raises_when_user_not_admin(self):
|
||||
"""require_admin raises 403 when user is not admin."""
|
||||
from fastapi import HTTPException
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"name": "user", "is_admin": False}
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
require_admin(mock_request)
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_require_admin_returns_user_when_admin(self):
|
||||
"""require_admin returns user dict when user is admin."""
|
||||
mock_request = MagicMock()
|
||||
admin_user = {"name": "admin", "is_admin": True}
|
||||
mock_request.session.get.return_value = admin_user
|
||||
result = require_admin(mock_request)
|
||||
assert result == admin_user
|
||||
|
||||
|
||||
class TestSettingsGetAll:
|
||||
"""GET /api/settings/ - list all settings."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_settings_success(self, client):
|
||||
"""Successfully retrieve all settings as admin."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
response = client.get("/api/settings/")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "settings" in data
|
||||
assert "categories" in data
|
||||
assert "db_settings" in data
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_settings_error_handling(self, client):
|
||||
"""500 error when internal exception occurs."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.get_all_settings_from_db", side_effect=RuntimeError("db error")):
|
||||
response = client.get("/api/settings/")
|
||||
assert response.status_code == 500
|
||||
assert "Failed to retrieve settings" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
|
||||
class TestSettingsGetOne:
|
||||
"""GET /api/settings/{key} - get a specific setting."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_setting_known_key(self, client):
|
||||
"""Retrieve a known setting key."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
response = client.get("/api/settings/workdir")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["key"] == "workdir"
|
||||
assert "metadata" in data
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_setting_unknown_key(self, client):
|
||||
"""Retrieve an unknown setting key returns value=None."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
response = client.get("/api/settings/nonexistent_key_xyz")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["key"] == "nonexistent_key_xyz"
|
||||
assert data["value"] is None
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_setting_internal_error(self, client):
|
||||
"""500 error when get_setting_metadata raises."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.get_setting_metadata", side_effect=RuntimeError("boom")):
|
||||
response = client.get("/api/settings/workdir")
|
||||
assert response.status_code == 500
|
||||
assert "Failed to retrieve setting" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
|
||||
class TestSettingsUpdate:
|
||||
"""POST /api/settings/{key} - update a setting."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_setting_success(self, client):
|
||||
"""Successfully update a setting."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.save_setting_to_db", return_value=True), \
|
||||
patch("app.api.settings.validate_setting_value", return_value=(True, None)):
|
||||
response = client.post(
|
||||
"/api/settings/workdir",
|
||||
json={"key": "workdir", "value": "/new/path"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert data["key"] == "workdir"
|
||||
assert data["value"] == "/new/path"
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_setting_validation_failure(self, client):
|
||||
"""400 error when validation fails."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(False, "Invalid value")):
|
||||
response = client.post(
|
||||
"/api/settings/workdir",
|
||||
json={"key": "workdir", "value": "bad"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert "Invalid value" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_setting_save_failure(self, client):
|
||||
"""500 error when save_setting_to_db returns False."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(True, None)), \
|
||||
patch("app.api.settings.save_setting_to_db", return_value=False):
|
||||
response = client.post(
|
||||
"/api/settings/workdir",
|
||||
json={"key": "workdir", "value": "/tmp"},
|
||||
)
|
||||
assert response.status_code == 500
|
||||
assert "Failed to save setting" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_setting_with_none_value(self, client):
|
||||
"""Update a setting with None value (delete semantics)."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.save_setting_to_db", return_value=True):
|
||||
response = client.post(
|
||||
"/api/settings/workdir",
|
||||
json={"key": "workdir", "value": None},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["success"] is True
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_update_setting_unexpected_error(self, client):
|
||||
"""500 error when an unexpected exception is raised."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(True, None)), \
|
||||
patch("app.api.settings.save_setting_to_db", side_effect=RuntimeError("unexpected")):
|
||||
response = client.post(
|
||||
"/api/settings/workdir",
|
||||
json={"key": "workdir", "value": "/tmp"},
|
||||
)
|
||||
assert response.status_code == 500
|
||||
assert "Failed to update setting" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
|
||||
class TestSettingsDelete:
|
||||
"""DELETE /api/settings/{key} - delete a setting."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_setting_success(self, client):
|
||||
"""Successfully delete a setting."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.delete_setting_from_db", return_value=True):
|
||||
response = client.delete("/api/settings/workdir")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["success"] is True
|
||||
assert "deleted" in data["message"].lower()
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_setting_not_found(self, client):
|
||||
"""404 error when setting not found in DB."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.delete_setting_from_db", return_value=False):
|
||||
response = client.delete("/api/settings/nonexistent")
|
||||
assert response.status_code == 404
|
||||
assert "not found" in response.json()["detail"].lower()
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_delete_setting_unexpected_error(self, client):
|
||||
"""500 error when an unexpected exception is raised."""
|
||||
fastapi_app.dependency_overrides[require_admin] = _override_admin
|
||||
try:
|
||||
with patch("app.api.settings.delete_setting_from_db", side_effect=RuntimeError("db crash")):
|
||||
response = client.delete("/api/settings/workdir")
|
||||
assert response.status_code == 500
|
||||
assert "Failed to delete setting" in response.json()["detail"]
|
||||
finally:
|
||||
fastapi_app.dependency_overrides.pop(require_admin, None)
|
||||
|
||||
|
||||
class TestSettingsBulkUpdate:
|
||||
"""Tests for bulk_update_settings handler.
|
||||
|
||||
The /bulk-update route is defined after /{key} in the router, so FastAPI
|
||||
matches /{key} first. We test the async handler function directly.
|
||||
"""
|
||||
|
||||
def _make_mock_db(self):
|
||||
return MagicMock()
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_all_success(self):
|
||||
"""Successfully bulk update multiple settings."""
|
||||
from app.api.settings import SettingUpdate, bulk_update_settings
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = self._make_mock_db()
|
||||
mock_admin = {"is_admin": True, "name": "admin"}
|
||||
updates = [
|
||||
SettingUpdate(key="workdir", value="/tmp/a"),
|
||||
SettingUpdate(key="external_hostname", value="example.com"),
|
||||
]
|
||||
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(True, None)), \
|
||||
patch("app.api.settings.save_setting_to_db", return_value=True):
|
||||
result = await bulk_update_settings(updates, mock_request, mock_db, mock_admin)
|
||||
assert result["success"] is True
|
||||
assert len(result["updated"]) == 2
|
||||
assert len(result["errors"]) == 0
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_with_validation_error(self):
|
||||
"""Bulk update skips invalid settings and reports errors."""
|
||||
from app.api.settings import SettingUpdate, bulk_update_settings
|
||||
|
||||
def mock_validate(key, value):
|
||||
if key == "bad_key":
|
||||
return (False, "Invalid value for bad_key")
|
||||
return (True, None)
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = self._make_mock_db()
|
||||
updates = [
|
||||
SettingUpdate(key="workdir", value="/tmp"),
|
||||
SettingUpdate(key="bad_key", value="invalid"),
|
||||
]
|
||||
|
||||
with patch("app.api.settings.validate_setting_value", side_effect=mock_validate), \
|
||||
patch("app.api.settings.save_setting_to_db", return_value=True):
|
||||
result = await bulk_update_settings(updates, mock_request, mock_db, {"is_admin": True})
|
||||
assert result["success"] is False
|
||||
assert len(result["updated"]) == 1
|
||||
assert len(result["errors"]) == 1
|
||||
assert result["errors"][0]["key"] == "bad_key"
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_save_failure(self):
|
||||
"""Bulk update reports errors when save_setting_to_db returns False."""
|
||||
from app.api.settings import SettingUpdate, bulk_update_settings
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = self._make_mock_db()
|
||||
updates = [SettingUpdate(key="workdir", value="/tmp")]
|
||||
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(True, None)), \
|
||||
patch("app.api.settings.save_setting_to_db", return_value=False):
|
||||
result = await bulk_update_settings(updates, mock_request, mock_db, {"is_admin": True})
|
||||
assert result["success"] is False
|
||||
assert len(result["errors"]) == 1
|
||||
assert "Failed to save" in result["errors"][0]["error"]
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_with_exception_during_save(self):
|
||||
"""Bulk update catches per-item exceptions and reports them."""
|
||||
from app.api.settings import SettingUpdate, bulk_update_settings
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = self._make_mock_db()
|
||||
updates = [SettingUpdate(key="workdir", value="/tmp")]
|
||||
|
||||
with patch("app.api.settings.validate_setting_value", return_value=(True, None)), \
|
||||
patch("app.api.settings.save_setting_to_db", side_effect=RuntimeError("boom")):
|
||||
result = await bulk_update_settings(updates, mock_request, mock_db, {"is_admin": True})
|
||||
assert result["success"] is False
|
||||
assert len(result["errors"]) == 1
|
||||
assert "boom" in result["errors"][0]["error"]
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.asyncio
|
||||
async def test_bulk_update_with_none_value(self):
|
||||
"""Bulk update with None value skips validation."""
|
||||
from app.api.settings import SettingUpdate, bulk_update_settings
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = self._make_mock_db()
|
||||
updates = [SettingUpdate(key="workdir", value=None)]
|
||||
|
||||
with patch("app.api.settings.save_setting_to_db", return_value=True):
|
||||
result = await bulk_update_settings(updates, mock_request, mock_db, {"is_admin": True})
|
||||
assert result["success"] is True
|
||||
assert len(result["updated"]) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. app/views/license_routes.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestLicenseRoutes:
|
||||
"""Tests for license and attribution view routes."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_lgpl_license_success(self, client):
|
||||
"""GET /licenses/lgpl.txt returns the LGPL license text."""
|
||||
response = client.get("/licenses/lgpl.txt")
|
||||
assert response.status_code == 200
|
||||
assert "text/plain" in response.headers["content-type"]
|
||||
# LGPL license files typically contain recognizable text
|
||||
assert len(response.text) > 0
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_lgpl_license_file_missing(self, client):
|
||||
"""GET /licenses/lgpl.txt returns 404 when file doesn't exist."""
|
||||
with patch("app.views.license_routes.Path") as MockPath:
|
||||
mock_path_instance = MagicMock()
|
||||
mock_path_instance.exists.return_value = False
|
||||
MockPath.return_value = mock_path_instance
|
||||
response = client.get("/licenses/lgpl.txt")
|
||||
assert response.status_code == 404
|
||||
# HTTPException returns JSON even with PlainTextResponse response_class
|
||||
assert "not found" in response.text.lower()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_serve_attribution_page(self, client):
|
||||
"""GET /attribution returns the attribution HTML page."""
|
||||
response = client.get("/attribution")
|
||||
assert response.status_code == 200
|
||||
assert "text/html" in response.headers["content-type"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. app/api/diagnostic.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestDiagnosticSettings:
|
||||
"""GET /api/diagnostic/settings - dump settings."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_diagnostic_settings_success(self, client):
|
||||
"""Returns safe subset of settings."""
|
||||
with patch("app.utils.config_validator.dump_all_settings"):
|
||||
response = client.get("/api/diagnostic/settings")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
assert "settings" in data
|
||||
assert "configured_services" in data["settings"]
|
||||
|
||||
|
||||
class TestDiagnosticTestNotification:
|
||||
"""POST /api/diagnostic/test-notification - send test notification."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_notification_no_urls_configured(self, client):
|
||||
"""Returns warning when no notification URLs are configured."""
|
||||
with patch("app.config.settings.notification_urls", new=[], create=True):
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "warning"
|
||||
assert "No notification" in data["message"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_notification_send_success(self, client):
|
||||
"""Returns success when notification is sent."""
|
||||
with patch("app.config.settings.notification_urls", new=["http://ntfy.example.com/test"], create=True), \
|
||||
patch("app.utils.notification.send_notification", return_value=True) as mock_send:
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
assert data["services_count"] == 1
|
||||
mock_send.assert_called_once()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_notification_send_failure(self, client):
|
||||
"""Returns error when send_notification returns False."""
|
||||
with patch("app.config.settings.notification_urls", new=["http://ntfy.example.com/test"], create=True), \
|
||||
patch("app.utils.notification.send_notification", return_value=False):
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "Failed" in data["message"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_notification_send_exception(self, client):
|
||||
"""Returns error when send_notification raises an exception."""
|
||||
with patch("app.config.settings.notification_urls", new=["http://ntfy.example.com/test"], create=True), \
|
||||
patch("app.utils.notification.send_notification", side_effect=RuntimeError("connection refused")):
|
||||
response = client.post("/api/diagnostic/test-notification")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "connection refused" in data["message"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. app/api/openai.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestOpenAITestEndpoint:
|
||||
"""GET /api/openai/test - test OpenAI API key."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_no_api_key(self, client):
|
||||
"""Returns error when no API key is configured."""
|
||||
with patch("app.config.settings.openai_api_key", new=""):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "No OpenAI API key" in data["message"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_valid_key(self, client):
|
||||
"""Returns success when API key is valid."""
|
||||
mock_models = MagicMock()
|
||||
mock_models.data = [MagicMock(), MagicMock(), MagicMock()]
|
||||
|
||||
mock_client_instance = MagicMock()
|
||||
mock_client_instance.models.list.return_value = mock_models
|
||||
|
||||
with patch("app.config.settings.openai_api_key", new="sk-valid-key"), \
|
||||
patch("openai.OpenAI", return_value=mock_client_instance):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "success"
|
||||
assert data["models_available"] == 3
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_auth_error(self, client):
|
||||
"""Returns error with auth flag when key is invalid."""
|
||||
mock_client_instance = MagicMock()
|
||||
mock_client_instance.models.list.side_effect = Exception("Incorrect API key provided")
|
||||
|
||||
with patch("app.config.settings.openai_api_key", new="sk-bad-key"), \
|
||||
patch("openai.OpenAI", return_value=mock_client_instance):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert data["is_auth_error"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_non_auth_error(self, client):
|
||||
"""Returns error without auth flag for non-auth failures."""
|
||||
mock_client_instance = MagicMock()
|
||||
mock_client_instance.models.list.side_effect = Exception("Connection timeout")
|
||||
|
||||
with patch("app.config.settings.openai_api_key", new="sk-valid-key"), \
|
||||
patch("openai.OpenAI", return_value=mock_client_instance):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert data["is_auth_error"] is False
|
||||
assert "Connection timeout" in data["message"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_import_error(self, client):
|
||||
"""Returns error when openai package is not installed."""
|
||||
import builtins
|
||||
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "openai":
|
||||
raise ImportError("No module named 'openai'")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch("app.config.settings.openai_api_key", new="sk-key"), \
|
||||
patch("builtins.__import__", side_effect=mock_import):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "not installed" in data["message"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_unexpected_error(self, client):
|
||||
"""Returns error for unexpected exceptions outside the inner try."""
|
||||
with patch("app.config.settings.openai_api_key", new="sk-key"), \
|
||||
patch("openai.OpenAI", side_effect=RuntimeError("unexpected crash")):
|
||||
response = client.get("/api/openai/test")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "Unexpected error" in data["message"]
|
||||
@@ -0,0 +1,876 @@
|
||||
"""
|
||||
Tests targeting uncovered lines in:
|
||||
- app/tasks/upload_to_s3.py (lines 43-46, 49-52, 67-71, 78, 99-109)
|
||||
- app/tasks/upload_to_sftp.py (lines 39-42, 62-63, 80-88, 103, 107-111, 128-132, 146-158)
|
||||
- app/utils/notification.py (lines 74-118)
|
||||
- app/tasks/send_to_all.py (lines 50, 76, 85-105, 122-146, 210-212, 228-230)
|
||||
"""
|
||||
|
||||
from contextlib import ExitStack
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.tasks.upload_to_s3 import upload_to_s3
|
||||
from app.tasks.upload_to_sftp import upload_to_sftp
|
||||
from app.tasks.send_to_all import send_to_all_destinations
|
||||
|
||||
_TEST_CRED = "test_secret" # noqa: S105
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# S3 upload tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUploadToS3Coverage:
|
||||
"""Tests for upload_to_s3 covering uncovered lines."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_no_bucket_name_raises(self, tmp_path):
|
||||
"""Lines 43-46: ValueError when s3_bucket_name is empty."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = ""
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
|
||||
with pytest.raises(ValueError, match="bucket name"):
|
||||
upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_no_aws_credentials_raises(self, tmp_path):
|
||||
"""Lines 49-52: ValueError when AWS credentials missing."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = ""
|
||||
ms.aws_secret_access_key = ""
|
||||
|
||||
with pytest.raises(ValueError, match="credentials"):
|
||||
upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_folder_prefix_appends_slash(self, tmp_path):
|
||||
"""Lines 67-71: folder prefix without trailing slash gets one added."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_s3 = MagicMock()
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.boto3") as mock_boto3,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
ms.aws_region = "us-east-1"
|
||||
ms.s3_folder_prefix = "docs" # no trailing slash
|
||||
ms.s3_storage_class = "STANDARD"
|
||||
ms.s3_acl = ""
|
||||
mock_boto3.client.return_value = mock_s3
|
||||
|
||||
result = upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
uploaded_key = mock_s3.upload_file.call_args[0][2]
|
||||
assert uploaded_key == "docs/test.pdf"
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_s3_key_without_prefix(self, tmp_path):
|
||||
"""Line 71: s3_key = filename when no folder prefix."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_s3 = MagicMock()
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.boto3") as mock_boto3,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
ms.aws_region = "us-east-1"
|
||||
ms.s3_folder_prefix = ""
|
||||
ms.s3_storage_class = "STANDARD"
|
||||
ms.s3_acl = ""
|
||||
mock_boto3.client.return_value = mock_s3
|
||||
|
||||
result = upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
assert mock_s3.upload_file.call_args[0][2] == "test.pdf"
|
||||
assert result["s3_key"] == "test.pdf"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_acl_added_when_configured(self, tmp_path):
|
||||
"""Line 78: ACL added to extra_args."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_s3 = MagicMock()
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.boto3") as mock_boto3,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
ms.aws_region = "us-east-1"
|
||||
ms.s3_folder_prefix = ""
|
||||
ms.s3_storage_class = "STANDARD"
|
||||
ms.s3_acl = "public-read"
|
||||
mock_boto3.client.return_value = mock_s3
|
||||
|
||||
upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
extra_args = mock_s3.upload_file.call_args[1]["ExtraArgs"]
|
||||
assert extra_args["ACL"] == "public-read"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_client_error_handler(self, tmp_path):
|
||||
"""Lines 99-109: ClientError is caught and re-raised."""
|
||||
from botocore.exceptions import ClientError
|
||||
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_s3 = MagicMock()
|
||||
error_response = {"Error": {"Code": "NoSuchBucket", "Message": "not found"}}
|
||||
mock_s3.upload_file.side_effect = ClientError(error_response, "PutObject")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.boto3") as mock_boto3,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
ms.aws_region = "us-east-1"
|
||||
ms.s3_folder_prefix = ""
|
||||
ms.s3_storage_class = "STANDARD"
|
||||
ms.s3_acl = ""
|
||||
mock_boto3.client.return_value = mock_s3
|
||||
|
||||
with pytest.raises(Exception, match="Failed to upload"):
|
||||
upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_generic_exception_handler(self, tmp_path):
|
||||
"""Lines 105-109: Generic exception is caught and re-raised."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_s3 = MagicMock()
|
||||
mock_s3.upload_file.side_effect = RuntimeError("network timeout")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_s3.settings") as ms,
|
||||
patch("app.tasks.upload_to_s3.boto3") as mock_boto3,
|
||||
patch("app.tasks.upload_to_s3.log_task_progress"),
|
||||
):
|
||||
ms.s3_bucket_name = "my-bucket"
|
||||
ms.aws_access_key_id = "key"
|
||||
ms.aws_secret_access_key = _TEST_CRED
|
||||
ms.aws_region = "us-east-1"
|
||||
ms.s3_folder_prefix = ""
|
||||
ms.s3_storage_class = "STANDARD"
|
||||
ms.s3_acl = ""
|
||||
mock_boto3.client.return_value = mock_s3
|
||||
|
||||
with pytest.raises(Exception, match="Error uploading"):
|
||||
upload_to_s3.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SFTP upload tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUploadToSftpCoverage:
|
||||
"""Tests for upload_to_sftp covering uncovered lines."""
|
||||
|
||||
def _sftp_settings(self, ms, password="pass", key=None, passphrase=None, disable_verify=True):
|
||||
"""Helper to configure common SFTP mock settings."""
|
||||
ms.sftp_host = "host"
|
||||
ms.sftp_port = 22
|
||||
ms.sftp_username = "user"
|
||||
ms.sftp_password = password
|
||||
ms.sftp_folder = "/upload"
|
||||
ms.sftp_disable_host_key_verification = disable_verify
|
||||
ms.sftp_private_key = key
|
||||
ms.sftp_private_key_passphrase = passphrase
|
||||
ms.workdir = "/workdir"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_missing_config_skips(self, tmp_path):
|
||||
"""Lines 39-42: Skipped when sftp config is incomplete."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
):
|
||||
ms.sftp_host = ""
|
||||
ms.sftp_port = 22
|
||||
ms.sftp_username = ""
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
assert result["status"] == "Skipped"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_system_host_keys_loaded(self, tmp_path):
|
||||
"""Lines 62-63: load_system_host_keys + RejectPolicy when verification enabled."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
mock_sftp_client.stat.return_value = MagicMock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="/upload/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", return_value="/upload/test.pdf"),
|
||||
):
|
||||
self._sftp_settings(ms, disable_verify=False)
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
mock_ssh.load_system_host_keys.assert_called_once()
|
||||
mock_ssh.set_missing_host_key_policy.assert_called_once_with(mock_paramiko.RejectPolicy())
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_key_auth_with_passphrase(self, tmp_path):
|
||||
"""Lines 80-81: key_filename and passphrase in connect_kwargs."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
key_file = tmp_path / "id_rsa"
|
||||
key_file.write_text("fake-key")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
mock_sftp_client.stat.return_value = MagicMock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="/upload/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", return_value="/upload/test.pdf"),
|
||||
):
|
||||
self._sftp_settings(ms, password="", key=str(key_file), passphrase="my-passphrase")
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
connect_kwargs = mock_ssh.connect.call_args[1]
|
||||
assert connect_kwargs["key_filename"] == str(key_file)
|
||||
assert connect_kwargs["passphrase"] == "my-passphrase"
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_password_auth(self, tmp_path):
|
||||
"""Lines 82-84: password authentication branch."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
mock_sftp_client.stat.return_value = MagicMock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="/upload/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", return_value="/upload/test.pdf"),
|
||||
):
|
||||
self._sftp_settings(ms, password="secret-pass")
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
connect_kwargs = mock_ssh.connect.call_args[1]
|
||||
assert connect_kwargs["password"] == "secret-pass"
|
||||
assert "key_filename" not in connect_kwargs
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_no_auth_method_raises(self, tmp_path):
|
||||
"""Lines 86-88: raises when no key or password."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
):
|
||||
self._sftp_settings(ms, password="")
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
with pytest.raises(Exception, match="Failed to upload"):
|
||||
upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_remote_path_slash_prepended(self, tmp_path):
|
||||
"""Line 103: slash prepended when remote_base starts with /."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
mock_sftp_client.stat.return_value = MagicMock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="upload/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", side_effect=lambda p, _: p),
|
||||
):
|
||||
self._sftp_settings(ms)
|
||||
ms.sftp_folder = "/data" # starts with /
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
assert result["sftp_path"].startswith("/")
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_remote_dir_creation(self, tmp_path):
|
||||
"""Lines 107-111, 128-132: creates remote directories when they don't exist."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
|
||||
def stat_side_effect(path):
|
||||
if path in ("/upload", "/upload/sub"):
|
||||
raise FileNotFoundError()
|
||||
return MagicMock()
|
||||
|
||||
mock_sftp_client.stat.side_effect = stat_side_effect
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="/upload/sub/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", side_effect=lambda p, _: p),
|
||||
):
|
||||
self._sftp_settings(ms)
|
||||
ms.sftp_folder = "/"
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
mock_sftp_client.mkdir.assert_any_call("/upload")
|
||||
mock_sftp_client.mkdir.assert_any_call("/upload/sub")
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mkdir_exception_logged(self, tmp_path):
|
||||
"""Lines 128-132: exception during mkdir is caught as warning."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_sftp_client = MagicMock()
|
||||
mock_ssh.open_sftp.return_value = mock_sftp_client
|
||||
mock_sftp_client.stat.side_effect = FileNotFoundError()
|
||||
mock_sftp_client.mkdir.side_effect = PermissionError("denied")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.extract_remote_path", return_value="/upload/test.pdf"),
|
||||
patch("app.tasks.upload_to_sftp.get_unique_filename", side_effect=lambda p, _: p),
|
||||
):
|
||||
self._sftp_settings(ms)
|
||||
ms.sftp_folder = "/"
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
result = upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
assert result["status"] == "Completed"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_cleanup_on_exception(self, tmp_path):
|
||||
"""Lines 146-158: connections cleaned up on exception."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_ssh = MagicMock()
|
||||
mock_ssh.connect.side_effect = ConnectionRefusedError("refused")
|
||||
|
||||
with (
|
||||
patch("app.tasks.upload_to_sftp.settings") as ms,
|
||||
patch("app.tasks.upload_to_sftp.paramiko") as mock_paramiko,
|
||||
patch("app.tasks.upload_to_sftp.log_task_progress"),
|
||||
patch("app.tasks.upload_to_sftp.sanitize_filename", return_value="test.pdf"),
|
||||
):
|
||||
self._sftp_settings(ms)
|
||||
mock_paramiko.SSHClient.return_value = mock_ssh
|
||||
|
||||
with pytest.raises(Exception, match="Failed to upload"):
|
||||
upload_to_sftp.apply(args=[str(f)], kwargs={"file_id": 1}).get()
|
||||
|
||||
mock_ssh.close.assert_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Notification tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNotificationCoverage:
|
||||
"""Tests for app/utils/notification covering lines 74-118."""
|
||||
|
||||
def _reset_apprise(self):
|
||||
import app.utils.notification as notif_module
|
||||
|
||||
notif_module._apprise = None
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_success(self):
|
||||
"""Lines 74-114: full body with successful server."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_server = MagicMock()
|
||||
mock_server.__str__ = lambda s: "json://localhost"
|
||||
mock_server.notify.return_value = True
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [mock_server]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Test Title", "Test body", notification_type="success")
|
||||
|
||||
assert result is True
|
||||
mock_server.notify.assert_called_once()
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_failure_type(self):
|
||||
"""Lines 83-84: notification_type='failure' maps to FAILURE."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_server = MagicMock()
|
||||
mock_server.__str__ = lambda s: "json://localhost"
|
||||
mock_server.notify.return_value = True
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [mock_server]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Fail", "Something failed", notification_type="failure")
|
||||
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_warning_type(self):
|
||||
"""Lines 81-82: notification_type='warn' maps to WARNING."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_server = MagicMock()
|
||||
mock_server.__str__ = lambda s: "json://localhost"
|
||||
mock_server.notify.return_value = True
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [mock_server]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Warn", "Warning msg", notification_type="warn")
|
||||
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_no_servers(self):
|
||||
"""Lines 87-89: returns False when servers list is empty."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = []
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Title", "Body")
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_server_fails(self):
|
||||
"""Lines 102-103: server.notify returns False."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_server = MagicMock()
|
||||
mock_server.__str__ = lambda s: "json://localhost"
|
||||
mock_server.notify.return_value = False
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [mock_server]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Title", "Body")
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_server_exception(self):
|
||||
"""Lines 104-105: exception from server.notify is caught."""
|
||||
self._reset_apprise()
|
||||
|
||||
mock_server = MagicMock()
|
||||
mock_server.__str__ = lambda s: "json://localhost"
|
||||
mock_server.notify.side_effect = RuntimeError("boom")
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [mock_server]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Title", "Body")
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_outer_exception(self):
|
||||
"""Lines 116-118: outer exception returns False."""
|
||||
self._reset_apprise()
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", side_effect=RuntimeError("init failed")),
|
||||
):
|
||||
ms.notification_urls = ["json://localhost"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Title", "Body")
|
||||
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_send_notification_partial_success(self):
|
||||
"""Lines 91-114: one server succeeds, one fails -> True."""
|
||||
self._reset_apprise()
|
||||
|
||||
good = MagicMock()
|
||||
good.__str__ = lambda s: "json://good"
|
||||
good.notify.return_value = True
|
||||
|
||||
bad = MagicMock()
|
||||
bad.__str__ = lambda s: "json://bad"
|
||||
bad.notify.return_value = False
|
||||
|
||||
mock_apprise_instance = MagicMock()
|
||||
mock_apprise_instance.servers = [good, bad]
|
||||
|
||||
with (
|
||||
patch("app.utils.notification.settings") as ms,
|
||||
patch("app.utils.notification.init_apprise", return_value=mock_apprise_instance),
|
||||
):
|
||||
ms.notification_urls = ["json://good", "json://bad"]
|
||||
from app.utils.notification import send_notification
|
||||
|
||||
result = send_notification("Title", "Body", notification_type="info")
|
||||
|
||||
assert result is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# send_to_all_destinations tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _all_should_upload_false():
|
||||
"""Return a list of patch context managers that set all _should_upload_* to False."""
|
||||
services = [
|
||||
"dropbox", "nextcloud", "paperless", "google_drive",
|
||||
"webdav", "ftp", "sftp", "email", "onedrive", "s3",
|
||||
]
|
||||
return [
|
||||
patch(f"app.tasks.send_to_all._should_upload_to_{s}", return_value=False)
|
||||
for s in services
|
||||
]
|
||||
|
||||
|
||||
class TestSendToAllCoverage:
|
||||
"""Tests for send_to_all_destinations covering uncovered lines."""
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_should_upload_returns_false(self, tmp_path):
|
||||
"""Line 50: service skipped when should_upload returns False."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.send_to_all.os.path.exists", return_value=True),
|
||||
patch("app.tasks.send_to_all.log_task_progress"),
|
||||
patch("app.tasks.send_to_all.settings") as ms,
|
||||
patch("app.tasks.send_to_all.get_provider_status", side_effect=Exception("unavailable")),
|
||||
patch("app.tasks.send_to_all._should_upload_to_dropbox", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_nextcloud", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_paperless", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_google_drive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_webdav", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_ftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_sftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_email", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_onedrive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_s3", return_value=False),
|
||||
patch("app.tasks.send_to_all.SessionLocal") as mock_session_cls,
|
||||
):
|
||||
ms.workdir = str(tmp_path)
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": 1}
|
||||
).get()
|
||||
|
||||
assert result["status"] == "Queued"
|
||||
assert result["tasks"] == {}
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_get_configured_services_from_validator(self):
|
||||
"""Lines 85-105: get_configured_services_from_validator maps provider status."""
|
||||
with patch("app.tasks.send_to_all.get_provider_status") as mock_prov:
|
||||
mock_prov.return_value = {
|
||||
"Dropbox": {"configured": True},
|
||||
"NextCloud": {"configured": False},
|
||||
"S3 Storage": {"configured": True},
|
||||
}
|
||||
from app.tasks.send_to_all import get_configured_services_from_validator
|
||||
|
||||
result = get_configured_services_from_validator()
|
||||
|
||||
assert result["dropbox"] is True
|
||||
assert result["nextcloud"] is False
|
||||
assert result["s3"] is True
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_file_id_lookup_from_db(self, tmp_path):
|
||||
"""Lines 136-146: file_id looked up from database when not provided."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_record = MagicMock()
|
||||
mock_record.id = 42
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = mock_record
|
||||
|
||||
mock_session_cls = MagicMock()
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with ExitStack() as stack:
|
||||
stack.enter_context(patch("app.tasks.send_to_all.os.path.exists", return_value=True))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.log_task_progress"))
|
||||
ms = stack.enter_context(patch("app.tasks.send_to_all.settings"))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.SessionLocal", mock_session_cls))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.get_provider_status", return_value={}))
|
||||
for p in _all_should_upload_false():
|
||||
stack.enter_context(p)
|
||||
ms.workdir = str(tmp_path)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": None}
|
||||
).get()
|
||||
|
||||
assert result["status"] == "Queued"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_file_id_lookup_no_record(self, tmp_path):
|
||||
"""Lines 136-146: file_id stays None when no DB record found."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
mock_session_cls = MagicMock()
|
||||
mock_session_cls.return_value.__enter__ = MagicMock(return_value=mock_db)
|
||||
mock_session_cls.return_value.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
with ExitStack() as stack:
|
||||
stack.enter_context(patch("app.tasks.send_to_all.os.path.exists", return_value=True))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.log_task_progress"))
|
||||
ms = stack.enter_context(patch("app.tasks.send_to_all.settings"))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.SessionLocal", mock_session_cls))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.get_provider_status", return_value={}))
|
||||
for p in _all_should_upload_false():
|
||||
stack.enter_context(p)
|
||||
ms.workdir = str(tmp_path)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": None}
|
||||
).get()
|
||||
|
||||
assert result["status"] == "Queued"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_validator_exception_falls_back(self, tmp_path):
|
||||
"""Lines 210-212: validator exception causes fallback to should_upload."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.send_to_all.os.path.exists", return_value=True),
|
||||
patch("app.tasks.send_to_all.log_task_progress"),
|
||||
patch("app.tasks.send_to_all.settings") as ms,
|
||||
patch("app.tasks.send_to_all.get_provider_status", side_effect=RuntimeError("fail")),
|
||||
patch("app.tasks.send_to_all._should_upload_to_dropbox", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_nextcloud", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_paperless", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_google_drive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_webdav", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_ftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_sftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_email", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_onedrive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_s3", return_value=False),
|
||||
patch("app.tasks.send_to_all.SessionLocal"),
|
||||
):
|
||||
ms.workdir = str(tmp_path)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": 1}
|
||||
).get()
|
||||
|
||||
assert result["status"] == "Queued"
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_queue_error(self, tmp_path):
|
||||
"""Lines 245-248: exception when queueing a task is caught."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
mock_upload = MagicMock()
|
||||
mock_upload.delay.side_effect = RuntimeError("broker down")
|
||||
|
||||
with ExitStack() as stack:
|
||||
stack.enter_context(patch("app.tasks.send_to_all.os.path.exists", return_value=True))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.log_task_progress"))
|
||||
ms = stack.enter_context(patch("app.tasks.send_to_all.settings"))
|
||||
stack.enter_context(
|
||||
patch(
|
||||
"app.tasks.send_to_all.get_provider_status",
|
||||
return_value={"Dropbox": {"configured": True}},
|
||||
)
|
||||
)
|
||||
stack.enter_context(patch("app.tasks.send_to_all.upload_to_dropbox", mock_upload))
|
||||
stack.enter_context(patch("app.tasks.send_to_all.SessionLocal"))
|
||||
for p in _all_should_upload_false():
|
||||
stack.enter_context(p)
|
||||
ms.workdir = str(tmp_path)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": 1}
|
||||
).get()
|
||||
|
||||
assert "dropbox_error" in result["tasks"]
|
||||
assert "broker down" in result["tasks"]["dropbox_error"]
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_should_upload_check_exception(self, tmp_path):
|
||||
"""Lines 228-230: exception in should_upload sets is_configured=False."""
|
||||
f = tmp_path / "test.pdf"
|
||||
f.write_text("data")
|
||||
|
||||
with (
|
||||
patch("app.tasks.send_to_all.os.path.exists", return_value=True),
|
||||
patch("app.tasks.send_to_all.log_task_progress"),
|
||||
patch("app.tasks.send_to_all.settings") as ms,
|
||||
patch("app.tasks.send_to_all.get_provider_status", side_effect=RuntimeError("fail")),
|
||||
patch("app.tasks.send_to_all._should_upload_to_dropbox", side_effect=RuntimeError("bad")),
|
||||
patch("app.tasks.send_to_all._should_upload_to_nextcloud", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_paperless", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_google_drive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_webdav", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_ftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_sftp", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_email", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_onedrive", return_value=False),
|
||||
patch("app.tasks.send_to_all._should_upload_to_s3", return_value=False),
|
||||
patch("app.tasks.send_to_all.SessionLocal"),
|
||||
):
|
||||
ms.workdir = str(tmp_path)
|
||||
|
||||
result = send_to_all_destinations.apply(
|
||||
args=[str(f)], kwargs={"use_validator": True, "file_id": 1}
|
||||
).get()
|
||||
|
||||
assert "dropbox_task_id" not in result["tasks"]
|
||||
assert result["status"] == "Queued"
|
||||
Reference in New Issue
Block a user