136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
"""Unit coverage for persisted processing log retention settings behavior."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
from unittest.mock import patch
|
|
|
|
|
|
BACKEND_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(BACKEND_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_ROOT))
|
|
|
|
if "pydantic_settings" not in sys.modules:
|
|
pydantic_settings_stub = ModuleType("pydantic_settings")
|
|
|
|
class _BaseSettings:
|
|
"""Minimal BaseSettings replacement for dependency-light unit test execution."""
|
|
|
|
def __init__(self, **kwargs: object) -> None:
|
|
for key, value in kwargs.items():
|
|
setattr(self, key, value)
|
|
|
|
def _settings_config_dict(**kwargs: object) -> dict[str, object]:
|
|
"""Returns configuration values using dict semantics expected by settings module."""
|
|
|
|
return kwargs
|
|
|
|
pydantic_settings_stub.BaseSettings = _BaseSettings
|
|
pydantic_settings_stub.SettingsConfigDict = _settings_config_dict
|
|
sys.modules["pydantic_settings"] = pydantic_settings_stub
|
|
|
|
from app.schemas.settings import AppSettingsUpdateRequest, ProcessingLogRetentionSettingsUpdateRequest
|
|
from app.services import app_settings
|
|
|
|
|
|
def _sample_current_payload() -> dict:
|
|
"""Builds a sanitized payload used as in-memory persistence fixture for update tests."""
|
|
|
|
return app_settings._sanitize_settings(app_settings._default_settings())
|
|
|
|
|
|
class ProcessingLogRetentionSettingsTests(unittest.TestCase):
|
|
"""Verifies defaulting, sanitization, schema mapping, and update merge behavior."""
|
|
|
|
def test_sanitize_settings_uses_default_retention_values(self) -> None:
|
|
"""Defaults are restored when persisted payload omits retention settings."""
|
|
|
|
sanitized = app_settings._sanitize_settings({})
|
|
self.assertEqual(
|
|
sanitized["processing_log_retention"],
|
|
{
|
|
"keep_document_sessions": 2,
|
|
"keep_unbound_entries": 80,
|
|
},
|
|
)
|
|
|
|
def test_sanitize_settings_clamps_retention_values(self) -> None:
|
|
"""Retention values are clamped to same bounds enforced by trim endpoint query rules."""
|
|
|
|
sanitized = app_settings._sanitize_settings(
|
|
{
|
|
"processing_log_retention": {
|
|
"keep_document_sessions": 99,
|
|
"keep_unbound_entries": -5,
|
|
}
|
|
}
|
|
)
|
|
self.assertEqual(
|
|
sanitized["processing_log_retention"],
|
|
{
|
|
"keep_document_sessions": 20,
|
|
"keep_unbound_entries": 0,
|
|
},
|
|
)
|
|
|
|
def test_update_request_schema_accepts_processing_log_retention_payload(self) -> None:
|
|
"""Settings PATCH schema keeps retention fields in serialized payloads."""
|
|
|
|
request_payload = AppSettingsUpdateRequest(
|
|
processing_log_retention=ProcessingLogRetentionSettingsUpdateRequest(
|
|
keep_document_sessions=7,
|
|
)
|
|
)
|
|
self.assertEqual(
|
|
request_payload.model_dump(exclude_none=True)["processing_log_retention"],
|
|
{"keep_document_sessions": 7},
|
|
)
|
|
|
|
def test_update_app_settings_merges_retention_block_and_sanitizes_values(self) -> None:
|
|
"""Settings updates merge partial retention values and persist sanitized results."""
|
|
|
|
current_payload = _sample_current_payload()
|
|
|
|
with (
|
|
patch.object(app_settings, "_read_raw_settings", return_value=current_payload),
|
|
patch.object(app_settings, "read_app_settings", return_value={"processing_log_retention": {}}),
|
|
patch.object(app_settings, "_write_settings") as write_settings_mock,
|
|
):
|
|
app_settings.update_app_settings(
|
|
processing_log_retention={
|
|
"keep_document_sessions": 9,
|
|
"keep_unbound_entries": 999,
|
|
}
|
|
)
|
|
|
|
written_payload = write_settings_mock.call_args.args[0]
|
|
self.assertEqual(
|
|
written_payload["processing_log_retention"],
|
|
{
|
|
"keep_document_sessions": 9,
|
|
"keep_unbound_entries": 400,
|
|
},
|
|
)
|
|
|
|
def test_read_processing_log_retention_settings_returns_defaults_when_key_missing(self) -> None:
|
|
"""Reader falls back to defaults when persisted payload omits retention key."""
|
|
|
|
payload_without_retention = _sample_current_payload()
|
|
payload_without_retention.pop("processing_log_retention", None)
|
|
with patch.object(app_settings, "_read_raw_settings", return_value=payload_without_retention):
|
|
retention = app_settings.read_processing_log_retention_settings()
|
|
self.assertEqual(
|
|
retention,
|
|
{
|
|
"keep_document_sessions": 2,
|
|
"keep_unbound_entries": 80,
|
|
},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|