From d9abbc1c998668ffae485493f27e82aed4fd1527 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 25 Feb 2026 15:24:42 +0000 Subject: [PATCH] test: improve coverage for process_with_ocr, api/settings, and settings_service to 90%+ Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- tests/test_api_settings_coverage.py | 278 ++++++++++++++++++++++ tests/test_process_with_ocr_comparison.py | 207 ++++++++++++++++ tests/test_settings_service_extended.py | 170 +++++++++++++ 3 files changed, 655 insertions(+) create mode 100644 tests/test_api_settings_coverage.py create mode 100644 tests/test_process_with_ocr_comparison.py create mode 100644 tests/test_settings_service_extended.py diff --git a/tests/test_api_settings_coverage.py b/tests/test_api_settings_coverage.py new file mode 100644 index 00000000..cb723d7f --- /dev/null +++ b/tests/test_api_settings_coverage.py @@ -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 diff --git a/tests/test_process_with_ocr_comparison.py b/tests/test_process_with_ocr_comparison.py new file mode 100644 index 00000000..052419d6 --- /dev/null +++ b/tests/test_process_with_ocr_comparison.py @@ -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<>\nendobj\n" + b"2 0 obj\n<>\nendobj\n" + b"3 0 obj\n<>\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<>\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 diff --git a/tests/test_settings_service_extended.py b/tests/test_settings_service_extended.py new file mode 100644 index 00000000..49ac09a5 --- /dev/null +++ b/tests/test_settings_service_extended.py @@ -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"