Merge pull request #416 from christianlouis/copilot/improve-test-coverage-mid-range-files
test: improve coverage for process_with_ocr, api/settings, and settings_service to 90%+
This commit is contained in:
Christian Krakau-Louis
@@ -0,0 +1,278 @@
|
||||
"""Tests for uncovered paths in app/api/settings.py.
|
||||
|
||||
Covers:
|
||||
- audit-log endpoint error path (lines 181-183)
|
||||
- export-env endpoint error path (lines 237-239)
|
||||
- delete setting HTTPException re-raise (line 341, 353)
|
||||
- install-ocr-languages endpoint (lines 424-443)
|
||||
- key history endpoint error path (lines 461-463)
|
||||
- rollback endpoint error path and HTTPException re-raise (lines 509-513)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestAuditLogEndpoint:
|
||||
"""Tests for GET /settings/audit-log endpoint."""
|
||||
|
||||
@patch("app.api.settings.get_audit_log")
|
||||
def test_audit_log_success(self, mock_audit_log):
|
||||
"""Test successful retrieval of audit log."""
|
||||
from app.api.settings import list_audit_log
|
||||
|
||||
mock_audit_log.return_value = [
|
||||
{"id": 1, "key": "debug", "old_value": "false", "new_value": "true", "changed_by": "admin"}
|
||||
]
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(list_audit_log(mock_request, mock_db, mock_admin, limit=100, offset=0))
|
||||
|
||||
assert result["entries"] == mock_audit_log.return_value
|
||||
assert result["limit"] == 100
|
||||
assert result["offset"] == 0
|
||||
|
||||
@patch("app.api.settings.get_audit_log")
|
||||
def test_audit_log_error_raises_500(self, mock_audit_log):
|
||||
"""Test that audit log errors raise HTTP 500."""
|
||||
from app.api.settings import list_audit_log
|
||||
|
||||
mock_audit_log.side_effect = Exception("DB connection lost")
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(list_audit_log(mock_request, mock_db, mock_admin))
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestExportEnvEndpoint:
|
||||
"""Tests for GET /settings/export-env endpoint."""
|
||||
|
||||
@patch("app.utils.settings_service.get_settings_for_export")
|
||||
def test_export_env_db_source_success(self, mock_export):
|
||||
"""Test successful export with db source."""
|
||||
from app.api.settings import export_env_settings
|
||||
|
||||
mock_export.return_value = {"DEBUG": "true", "WORKDIR": "/tmp"}
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(export_env_settings(mock_request, mock_db, mock_admin, source="db"))
|
||||
|
||||
assert result.status_code == 200
|
||||
assert "text/plain" in result.media_type
|
||||
body = result.body.decode()
|
||||
assert "DEBUG=true" in body
|
||||
assert "WORKDIR=/tmp" in body
|
||||
|
||||
@patch("app.utils.settings_service.get_settings_for_export")
|
||||
def test_export_env_error_raises_500(self, mock_export):
|
||||
"""Test that export errors raise HTTP 500."""
|
||||
from app.api.settings import export_env_settings
|
||||
|
||||
mock_export.side_effect = Exception("Export failed")
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(export_env_settings(mock_request, mock_db, mock_admin, source="db"))
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
def test_export_env_invalid_source_raises_400(self):
|
||||
"""Test that invalid source parameter raises HTTP 400."""
|
||||
from app.api.settings import export_env_settings
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(export_env_settings(mock_request, mock_db, mock_admin, source="invalid"))
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDeleteSettingEndpoint:
|
||||
"""Tests for DELETE /settings/{key} endpoint."""
|
||||
|
||||
@patch("app.api.settings.notify_settings_updated")
|
||||
@patch("app.api.settings.delete_setting_from_db")
|
||||
@patch("app.api.settings.validate_setting_key")
|
||||
def test_delete_setting_not_found_raises_404(self, mock_validate, mock_delete, mock_notify):
|
||||
"""Test that deleting a non-existent setting raises 404."""
|
||||
from app.api.settings import delete_setting
|
||||
|
||||
mock_delete.return_value = False
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"username": "admin", "is_admin": True}
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(delete_setting(key="nonexistent", request=mock_request, db=mock_db, admin=mock_admin))
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
@patch("app.api.settings.delete_setting_from_db")
|
||||
@patch("app.api.settings.validate_setting_key")
|
||||
def test_delete_setting_unexpected_error_raises_500(self, mock_validate, mock_delete):
|
||||
"""Test that unexpected errors during delete raise 500."""
|
||||
from app.api.settings import delete_setting
|
||||
|
||||
mock_delete.side_effect = RuntimeError("Unexpected DB error")
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"username": "admin", "is_admin": True}
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(delete_setting(key="some_key", request=mock_request, db=mock_db, admin=mock_admin))
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestInstallOCRLanguagesEndpoint:
|
||||
"""Tests for POST /settings/install-ocr-languages endpoint."""
|
||||
|
||||
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_from_settings")
|
||||
def test_install_ocr_languages_all_ok(self, mock_ensure):
|
||||
"""Test successful OCR language installation."""
|
||||
from app.api.settings import install_ocr_languages
|
||||
|
||||
mock_ensure.return_value = {"tesseract_missing": [], "easyocr_failed": []}
|
||||
mock_request = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(install_ocr_languages(mock_request, mock_admin))
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["tesseract_missing"] == []
|
||||
assert result["easyocr_failed"] == []
|
||||
|
||||
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_from_settings")
|
||||
def test_install_ocr_languages_some_missing(self, mock_ensure):
|
||||
"""Test partial failure in OCR language installation."""
|
||||
from app.api.settings import install_ocr_languages
|
||||
|
||||
mock_ensure.return_value = {"tesseract_missing": ["jpn"], "easyocr_failed": ["ar"]}
|
||||
mock_request = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(install_ocr_languages(mock_request, mock_admin))
|
||||
|
||||
assert result["success"] is False
|
||||
assert "jpn" in result["tesseract_missing"]
|
||||
assert "ar" in result["easyocr_failed"]
|
||||
|
||||
@patch("app.utils.ocr_language_manager.ensure_ocr_languages_from_settings")
|
||||
def test_install_ocr_languages_error_raises_500(self, mock_ensure):
|
||||
"""Test that errors during installation raise HTTP 500."""
|
||||
from app.api.settings import install_ocr_languages
|
||||
|
||||
mock_ensure.side_effect = Exception("Installation crashed")
|
||||
mock_request = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(install_ocr_languages(mock_request, mock_admin))
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestKeyHistoryEndpoint:
|
||||
"""Tests for GET /settings/{key}/history endpoint."""
|
||||
|
||||
@patch("app.api.settings.get_setting_history")
|
||||
@patch("app.api.settings.validate_setting_key_format")
|
||||
def test_key_history_success(self, mock_validate, mock_history):
|
||||
"""Test successful retrieval of setting history."""
|
||||
from app.api.settings import get_key_history
|
||||
|
||||
mock_history.return_value = [{"id": 1, "key": "debug", "old_value": "false", "new_value": "true"}]
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(get_key_history("debug", mock_request, mock_db, mock_admin))
|
||||
|
||||
assert result["key"] == "debug"
|
||||
assert len(result["history"]) == 1
|
||||
|
||||
@patch("app.api.settings.get_setting_history")
|
||||
@patch("app.api.settings.validate_setting_key_format")
|
||||
def test_key_history_error_raises_500(self, mock_validate, mock_history):
|
||||
"""Test that history retrieval errors raise HTTP 500."""
|
||||
from app.api.settings import get_key_history
|
||||
|
||||
mock_history.side_effect = Exception("History lookup failed")
|
||||
mock_request = MagicMock()
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(get_key_history("debug", mock_request, mock_db, mock_admin))
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRollbackSettingEndpoint:
|
||||
"""Tests for POST /settings/{key}/rollback/{history_id} endpoint."""
|
||||
|
||||
@patch("app.api.settings.notify_settings_updated")
|
||||
@patch("app.api.settings.rollback_setting")
|
||||
@patch("app.api.settings.validate_setting_key_format")
|
||||
def test_rollback_success(self, mock_validate, mock_rollback, mock_notify):
|
||||
"""Test successful setting rollback."""
|
||||
from app.api.settings import rollback_setting_to_history
|
||||
|
||||
mock_rollback.return_value = True
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"username": "admin", "is_admin": True}
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
result = asyncio.run(rollback_setting_to_history("debug", 42, mock_request, mock_db, mock_admin))
|
||||
|
||||
assert result["success"] is True
|
||||
|
||||
@patch("app.api.settings.rollback_setting")
|
||||
@patch("app.api.settings.validate_setting_key_format")
|
||||
def test_rollback_not_found_raises_404(self, mock_validate, mock_rollback):
|
||||
"""Test that rollback with invalid history_id raises 404."""
|
||||
from app.api.settings import rollback_setting_to_history
|
||||
|
||||
mock_rollback.return_value = False
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"username": "admin", "is_admin": True}
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(rollback_setting_to_history("debug", 999, mock_request, mock_db, mock_admin))
|
||||
assert exc_info.value.status_code == 404
|
||||
|
||||
@patch("app.api.settings.rollback_setting")
|
||||
@patch("app.api.settings.validate_setting_key_format")
|
||||
def test_rollback_unexpected_error_raises_500(self, mock_validate, mock_rollback):
|
||||
"""Test that unexpected errors during rollback raise 500."""
|
||||
from app.api.settings import rollback_setting_to_history
|
||||
|
||||
mock_rollback.side_effect = RuntimeError("Unexpected error")
|
||||
mock_request = MagicMock()
|
||||
mock_request.session.get.return_value = {"username": "admin", "is_admin": True}
|
||||
mock_db = MagicMock()
|
||||
mock_admin = {"is_admin": True}
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(rollback_setting_to_history("debug", 1, mock_request, mock_db, mock_admin))
|
||||
assert exc_info.value.status_code == 500
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Tests for the head-to-head text comparison logic in process_with_ocr.
|
||||
|
||||
Covers lines 154-232 of app/tasks/process_with_ocr.py:
|
||||
- original_text provided with both texts non-empty (comparison preferred=original)
|
||||
- original_text provided with both texts non-empty (comparison preferred=ocr)
|
||||
- comparison raises an exception → keep OCR text
|
||||
- original_text provided but OCR returns empty text → fallback to original
|
||||
- original_text provided but is empty → skip comparison, use OCR output
|
||||
"""
|
||||
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.utils.text_quality import TextComparisonResult
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestProcessWithOCRTextComparison:
|
||||
"""Tests for the head-to-head quality comparison in process_with_ocr."""
|
||||
|
||||
_MINIMAL_PDF = (
|
||||
b"%PDF-1.4\n"
|
||||
b"1 0 obj\n<</Type /Catalog /Pages 2 0 R>>\nendobj\n"
|
||||
b"2 0 obj\n<</Type /Pages /Kids [3 0 R] /Count 1>>\nendobj\n"
|
||||
b"3 0 obj\n<</Type /Page /Parent 2 0 R /MediaBox [0 0 612 792]>>\nendobj\n"
|
||||
b"xref\n0 4\n"
|
||||
b"0000000000 65535 f \n"
|
||||
b"0000000009 00000 n \n"
|
||||
b"0000000058 00000 n \n"
|
||||
b"0000000115 00000 n \n"
|
||||
b"trailer\n<</Size 4 /Root 1 0 R>>\n"
|
||||
b"startxref\n190\n%%EOF\n"
|
||||
)
|
||||
|
||||
def _setup_ocr_task(self, tmp_path, extracted_text, searchable_pdf_path=None):
|
||||
"""Helper to set up common mocks for OCR task tests."""
|
||||
from app.utils.ocr_provider import OCRResult
|
||||
|
||||
tmp_dir = tmp_path / "tmp"
|
||||
tmp_dir.mkdir(exist_ok=True)
|
||||
pdf_file = tmp_dir / "test.pdf"
|
||||
pdf_file.write_bytes(self._MINIMAL_PDF)
|
||||
|
||||
mock_result = OCRResult(
|
||||
provider="azure",
|
||||
text=extracted_text,
|
||||
searchable_pdf_path=searchable_pdf_path or str(pdf_file),
|
||||
)
|
||||
|
||||
provider_mock = Mock()
|
||||
provider_mock.name = "azure"
|
||||
provider_mock.__class__.__name__ = "AzureOCRProvider"
|
||||
provider_mock.process.return_value = mock_result
|
||||
|
||||
return pdf_file, provider_mock
|
||||
|
||||
@patch("app.tasks.process_with_ocr.log_task_progress")
|
||||
@patch("app.tasks.process_with_ocr.rotate_pdf_pages")
|
||||
@patch("app.tasks.process_with_ocr.compare_text_quality")
|
||||
def test_comparison_prefers_original_text(self, mock_compare, mock_rotate, mock_log, tmp_path):
|
||||
"""When comparison prefers original, final_text should be original_text."""
|
||||
from app.tasks.process_with_ocr import process_with_ocr
|
||||
|
||||
pdf_file, provider_mock = self._setup_ocr_task(tmp_path, "OCR extracted text")
|
||||
mock_rotate.delay = Mock()
|
||||
|
||||
mock_compare.return_value = TextComparisonResult(
|
||||
preferred="original",
|
||||
original_score=90,
|
||||
ocr_score=70,
|
||||
explanation="Original text is cleaner",
|
||||
)
|
||||
|
||||
with (
|
||||
patch("app.tasks.process_with_ocr.settings") as mock_settings,
|
||||
patch("app.tasks.process_with_ocr.get_ocr_providers") as mock_providers,
|
||||
patch(
|
||||
"app.tasks.process_with_ocr.merge_ocr_results",
|
||||
return_value=("OCR extracted text", str(pdf_file), {}),
|
||||
),
|
||||
):
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
mock_providers.return_value = [provider_mock]
|
||||
|
||||
result = process_with_ocr.run("test.pdf", file_id=1, original_text="Original embedded text")
|
||||
|
||||
assert result["cleaned_text"] == "Original embedded text"
|
||||
mock_compare.assert_called_once_with("Original embedded text", "OCR extracted text")
|
||||
|
||||
@patch("app.tasks.process_with_ocr.log_task_progress")
|
||||
@patch("app.tasks.process_with_ocr.rotate_pdf_pages")
|
||||
@patch("app.tasks.process_with_ocr.compare_text_quality")
|
||||
def test_comparison_prefers_ocr_text(self, mock_compare, mock_rotate, mock_log, tmp_path):
|
||||
"""When comparison prefers OCR, final_text should be the OCR text."""
|
||||
from app.tasks.process_with_ocr import process_with_ocr
|
||||
|
||||
pdf_file, provider_mock = self._setup_ocr_task(tmp_path, "High quality OCR text")
|
||||
mock_rotate.delay = Mock()
|
||||
|
||||
mock_compare.return_value = TextComparisonResult(
|
||||
preferred="ocr",
|
||||
original_score=50,
|
||||
ocr_score=95,
|
||||
explanation="OCR text is more complete",
|
||||
)
|
||||
|
||||
with (
|
||||
patch("app.tasks.process_with_ocr.settings") as mock_settings,
|
||||
patch("app.tasks.process_with_ocr.get_ocr_providers") as mock_providers,
|
||||
patch(
|
||||
"app.tasks.process_with_ocr.merge_ocr_results",
|
||||
return_value=("High quality OCR text", str(pdf_file), {}),
|
||||
),
|
||||
):
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
mock_providers.return_value = [provider_mock]
|
||||
|
||||
result = process_with_ocr.run("test.pdf", file_id=1, original_text="Low quality original")
|
||||
|
||||
assert result["cleaned_text"] == "High quality OCR text"
|
||||
mock_compare.assert_called_once()
|
||||
|
||||
@patch("app.tasks.process_with_ocr.log_task_progress")
|
||||
@patch("app.tasks.process_with_ocr.rotate_pdf_pages")
|
||||
@patch("app.tasks.process_with_ocr.compare_text_quality")
|
||||
def test_comparison_exception_keeps_ocr_text(self, mock_compare, mock_rotate, mock_log, tmp_path):
|
||||
"""When comparison raises an exception, the OCR text should be kept."""
|
||||
from app.tasks.process_with_ocr import process_with_ocr
|
||||
|
||||
pdf_file, provider_mock = self._setup_ocr_task(tmp_path, "OCR text after error")
|
||||
mock_rotate.delay = Mock()
|
||||
|
||||
mock_compare.side_effect = RuntimeError("AI comparison failed")
|
||||
|
||||
with (
|
||||
patch("app.tasks.process_with_ocr.settings") as mock_settings,
|
||||
patch("app.tasks.process_with_ocr.get_ocr_providers") as mock_providers,
|
||||
patch(
|
||||
"app.tasks.process_with_ocr.merge_ocr_results",
|
||||
return_value=("OCR text after error", str(pdf_file), {}),
|
||||
),
|
||||
):
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
mock_providers.return_value = [provider_mock]
|
||||
|
||||
result = process_with_ocr.run("test.pdf", file_id=1, original_text="Some original text")
|
||||
|
||||
assert result["cleaned_text"] == "OCR text after error"
|
||||
|
||||
@patch("app.tasks.process_with_ocr.log_task_progress")
|
||||
@patch("app.tasks.process_with_ocr.rotate_pdf_pages")
|
||||
def test_ocr_empty_falls_back_to_original(self, mock_rotate, mock_log, tmp_path):
|
||||
"""When OCR returns empty text but original is non-empty, fallback to original."""
|
||||
from app.tasks.process_with_ocr import process_with_ocr
|
||||
|
||||
pdf_file, provider_mock = self._setup_ocr_task(tmp_path, "")
|
||||
mock_rotate.delay = Mock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.process_with_ocr.settings") as mock_settings,
|
||||
patch("app.tasks.process_with_ocr.get_ocr_providers") as mock_providers,
|
||||
patch(
|
||||
"app.tasks.process_with_ocr.merge_ocr_results",
|
||||
return_value=("", str(pdf_file), {}),
|
||||
),
|
||||
):
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
mock_providers.return_value = [provider_mock]
|
||||
|
||||
result = process_with_ocr.run("test.pdf", file_id=1, original_text="Fallback original text")
|
||||
|
||||
assert result["cleaned_text"] == "Fallback original text"
|
||||
|
||||
@patch("app.tasks.process_with_ocr.log_task_progress")
|
||||
@patch("app.tasks.process_with_ocr.rotate_pdf_pages")
|
||||
def test_original_text_empty_skips_comparison(self, mock_rotate, mock_log, tmp_path):
|
||||
"""When original_text is empty string, skip comparison and use OCR output."""
|
||||
from app.tasks.process_with_ocr import process_with_ocr
|
||||
|
||||
pdf_file, provider_mock = self._setup_ocr_task(tmp_path, "Good OCR text")
|
||||
mock_rotate.delay = Mock()
|
||||
|
||||
with (
|
||||
patch("app.tasks.process_with_ocr.settings") as mock_settings,
|
||||
patch("app.tasks.process_with_ocr.get_ocr_providers") as mock_providers,
|
||||
patch(
|
||||
"app.tasks.process_with_ocr.merge_ocr_results",
|
||||
return_value=("Good OCR text", str(pdf_file), {}),
|
||||
),
|
||||
patch("app.tasks.process_with_ocr.compare_text_quality") as mock_compare,
|
||||
):
|
||||
mock_settings.workdir = str(tmp_path)
|
||||
mock_providers.return_value = [provider_mock]
|
||||
|
||||
result = process_with_ocr.run("test.pdf", file_id=1, original_text="")
|
||||
|
||||
assert result["cleaned_text"] == "Good OCR text"
|
||||
mock_compare.assert_not_called()
|
||||
|
||||
# Verify "skipped" was logged for compare_ocr_quality
|
||||
skip_calls = [
|
||||
c
|
||||
for c in mock_log.call_args_list
|
||||
if len(c[0]) >= 3 and c[0][1] == "compare_ocr_quality" and c[0][2] == "skipped"
|
||||
]
|
||||
assert len(skip_calls) == 1
|
||||
@@ -0,0 +1,170 @@
|
||||
"""Tests for uncovered paths in app/utils/settings_service.py.
|
||||
|
||||
Covers:
|
||||
- save_setting_to_db: decryption of old_storage_value for sensitive keys (lines 1247-1252)
|
||||
- delete_setting_from_db: decryption of stored value for sensitive keys (lines 1329-1334)
|
||||
- get_audit_log: SQLAlchemyError handling (lines 1469-1471)
|
||||
- get_setting_history: SQLAlchemyError handling (lines 1510-1512)
|
||||
- rollback_setting: SQLAlchemyError handling (lines 1553-1556)
|
||||
- get_settings_for_export: DB value takes precedence in effective export (line 1582)
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.exc import SQLAlchemyError
|
||||
|
||||
from app.models import ApplicationSettings, SettingsAuditLog
|
||||
from app.utils.settings_service import (
|
||||
delete_setting_from_db,
|
||||
get_audit_log,
|
||||
get_setting_history,
|
||||
get_settings_for_export,
|
||||
rollback_setting,
|
||||
save_setting_to_db,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestSaveSettingDecryptsOldValue:
|
||||
"""Tests for save_setting_to_db decrypting old value for audit log."""
|
||||
|
||||
@patch("app.utils.settings_service.get_setting_metadata")
|
||||
def test_save_sensitive_setting_decrypts_old_value_for_audit(self, mock_metadata, db_session):
|
||||
"""When updating a sensitive setting, old value is decrypted for the audit log."""
|
||||
mock_metadata.return_value = {"sensitive": True}
|
||||
|
||||
# First, save an encrypted value directly
|
||||
setting = ApplicationSettings(key="openai_api_key", value="enc:old_encrypted")
|
||||
db_session.add(setting)
|
||||
db_session.commit()
|
||||
|
||||
with (
|
||||
patch("app.utils.encryption.is_encryption_available", return_value=True),
|
||||
patch("app.utils.encryption.encrypt_value", return_value="enc:new_encrypted"),
|
||||
patch("app.utils.encryption.decrypt_value", return_value="old_plaintext_key") as mock_decrypt,
|
||||
):
|
||||
result = save_setting_to_db(db_session, "openai_api_key", "new_api_key", changed_by="admin")
|
||||
|
||||
assert result is True
|
||||
mock_decrypt.assert_called_once_with("enc:old_encrypted")
|
||||
|
||||
# Verify audit log has the decrypted old value
|
||||
audit = db_session.query(SettingsAuditLog).filter_by(key="openai_api_key").first()
|
||||
assert audit is not None
|
||||
assert audit.old_value == "old_plaintext_key"
|
||||
|
||||
@patch("app.utils.settings_service.get_setting_metadata")
|
||||
def test_save_sensitive_setting_decryption_fails_uses_raw(self, mock_metadata, db_session):
|
||||
"""When decryption of old value fails, the raw stored value is used."""
|
||||
mock_metadata.return_value = {"sensitive": True}
|
||||
|
||||
setting = ApplicationSettings(key="openai_api_key", value="enc:corrupted")
|
||||
db_session.add(setting)
|
||||
db_session.commit()
|
||||
|
||||
with (
|
||||
patch("app.utils.encryption.is_encryption_available", return_value=True),
|
||||
patch("app.utils.encryption.encrypt_value", return_value="enc:new"),
|
||||
patch("app.utils.encryption.decrypt_value", side_effect=Exception("Decryption failed")),
|
||||
):
|
||||
result = save_setting_to_db(db_session, "openai_api_key", "new_key", changed_by="admin")
|
||||
|
||||
assert result is True
|
||||
|
||||
# Audit log should have the raw encrypted value as fallback
|
||||
audit = db_session.query(SettingsAuditLog).filter_by(key="openai_api_key").first()
|
||||
assert audit is not None
|
||||
assert audit.old_value == "enc:corrupted"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDeleteSettingDecryptsOldValue:
|
||||
"""Tests for delete_setting_from_db decrypting stored value for audit log."""
|
||||
|
||||
@patch("app.utils.settings_service.get_setting_metadata")
|
||||
def test_delete_sensitive_setting_decrypts_for_audit(self, mock_metadata, db_session):
|
||||
"""When deleting a sensitive setting, old value is decrypted for the audit log."""
|
||||
mock_metadata.return_value = {"sensitive": True}
|
||||
|
||||
setting = ApplicationSettings(key="openai_api_key", value="enc:secret_value")
|
||||
db_session.add(setting)
|
||||
db_session.commit()
|
||||
|
||||
with patch("app.utils.encryption.decrypt_value", return_value="decrypted_secret") as mock_decrypt:
|
||||
result = delete_setting_from_db(db_session, "openai_api_key", changed_by="admin")
|
||||
|
||||
assert result is True
|
||||
mock_decrypt.assert_called_once_with("enc:secret_value")
|
||||
|
||||
audit = db_session.query(SettingsAuditLog).filter_by(key="openai_api_key").first()
|
||||
assert audit is not None
|
||||
assert audit.old_value == "decrypted_secret"
|
||||
assert audit.action == "delete"
|
||||
|
||||
@patch("app.utils.settings_service.get_setting_metadata")
|
||||
def test_delete_sensitive_setting_decryption_fails_uses_raw(self, mock_metadata, db_session):
|
||||
"""When decryption fails during delete, the raw value is used in audit."""
|
||||
mock_metadata.return_value = {"sensitive": True}
|
||||
|
||||
setting = ApplicationSettings(key="openai_api_key", value="enc:bad_data")
|
||||
db_session.add(setting)
|
||||
db_session.commit()
|
||||
|
||||
with patch("app.utils.encryption.decrypt_value", side_effect=Exception("Bad key")):
|
||||
result = delete_setting_from_db(db_session, "openai_api_key", changed_by="admin")
|
||||
|
||||
assert result is True
|
||||
|
||||
audit = db_session.query(SettingsAuditLog).filter_by(key="openai_api_key").first()
|
||||
assert audit is not None
|
||||
assert audit.old_value == "enc:bad_data"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetAuditLogError:
|
||||
"""Tests for get_audit_log SQLAlchemyError handling."""
|
||||
|
||||
def test_get_audit_log_returns_empty_on_db_error(self, db_session):
|
||||
"""get_audit_log returns empty list when a database error occurs."""
|
||||
with patch.object(db_session, "query", side_effect=SQLAlchemyError("Connection lost")):
|
||||
result = get_audit_log(db_session)
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetSettingHistoryError:
|
||||
"""Tests for get_setting_history SQLAlchemyError handling."""
|
||||
|
||||
def test_get_setting_history_returns_empty_on_db_error(self, db_session):
|
||||
"""get_setting_history returns empty list when a database error occurs."""
|
||||
with patch.object(db_session, "query", side_effect=SQLAlchemyError("Timeout")):
|
||||
result = get_setting_history(db_session, "debug")
|
||||
assert result == []
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestRollbackSettingError:
|
||||
"""Tests for rollback_setting SQLAlchemyError handling."""
|
||||
|
||||
def test_rollback_setting_returns_false_on_db_error(self, db_session):
|
||||
"""rollback_setting returns False when a database error occurs."""
|
||||
with patch.object(db_session, "query", side_effect=SQLAlchemyError("DB locked")):
|
||||
result = rollback_setting(db_session, "debug", 1, changed_by="admin")
|
||||
assert result is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetSettingsForExport:
|
||||
"""Tests for get_settings_for_export effective mode."""
|
||||
|
||||
def test_effective_export_db_value_takes_precedence(self, db_session):
|
||||
"""In effective mode, DB value takes precedence over ENV/default."""
|
||||
# Save a value in the DB
|
||||
save_setting_to_db(db_session, "debug", "true", changed_by="test")
|
||||
|
||||
with patch("app.utils.settings_service.SETTING_METADATA", {"debug": {"type": "bool"}}):
|
||||
result = get_settings_for_export(db_session, source="effective")
|
||||
|
||||
assert "DEBUG" in result
|
||||
assert result["DEBUG"] == "true"
|
||||
Reference in New Issue
Block a user