brain-of-reese/tests/unit/test_settings_overlay.py

"""Phase 168 (task 02): the DB-over-env overlay resolver.

Pure/unit level — no live DB: the ``load_overrides`` failure paths use
a stubbed engine, the kill switch is flipped via ``os.environ`` (it is
read per call), and the env base comes from the process environment
(pinned by ``tests/conftest.py`` — ``BOR_SETTINGS_DB=0``).
"""
from __future__ import annotations

import logging
from pathlib import Path

import pytest
from sqlalchemy.exc import OperationalError, ProgrammingError

from app.config import Settings
from app.core import settings_overlay
from app.core.settings_overlay import (
    _coerce,
    enabled,
    load_overrides,
    merge,
    register_invalidator,
    reset_state_for_tests,
    resolve_settings,
    run_invalidators,
)


@pytest.fixture(autouse=True)
def _clean_state() -> None:
    reset_state_for_tests()


# ---------------------------------------------------------------------------
# enabled() — the BOR_SETTINGS_DB kill switch (D6)
# ---------------------------------------------------------------------------


def test_enabled_matrix(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("BOR_SETTINGS_DB", raising=False)
    assert enabled() is True
    for value in ("0", "false", "no", "off"):
        monkeypatch.setenv("BOR_SETTINGS_DB", value)
        assert enabled() is False, value
    for value in ("1", "true"):
        monkeypatch.setenv("BOR_SETTINGS_DB", value)
        assert enabled() is True, value
    # read per call — a flip is visible without any cache clear
    monkeypatch.setenv("BOR_SETTINGS_DB", "off")
    assert enabled() is False
    monkeypatch.setenv("BOR_SETTINGS_DB", "1")
    assert enabled() is True


# ---------------------------------------------------------------------------
# _coerce — env-native strings; the one list[str] field is JSON
# ---------------------------------------------------------------------------


def test_coerce_scalars_are_raw_strings() -> None:
    # pydantic-settings lax-coerces on Settings() — the raw string goes
    # through exactly like the env var spelling.
    assert _coerce("relevance_threshold", "0.55") == "0.55"
    assert _coerce("llm_retries", "7") == "7"
    assert _coerce("stream_thinking", "false") == "false"
    assert _coerce("import_extensions", "md,txt") == "md,txt"


def test_coerce_suggestions_json() -> None:
    assert set(settings_overlay.LIST_FIELDS) == {"suggestions"}
    assert _coerce("suggestions", '["a", "b"]') == ["a", "b"]
    with pytest.raises(ValueError):
        _coerce("suggestions", "not a json array")


# ---------------------------------------------------------------------------
# merge — layer rows over the env base, fail soft per key (LOCKED A5)
# ---------------------------------------------------------------------------


def test_merge_happy_path_env_base() -> None:
    base = Settings()
    merged, skipped = merge({"relevance_threshold": "0.55"})
    assert skipped == []
    assert merged.relevance_threshold == 0.55
    assert isinstance(merged.relevance_threshold, float)
    # everything else is byte-equal to the env base
    for name in Settings.model_fields:
        if name != "relevance_threshold":
            assert getattr(merged, name) == getattr(base, name), name


def test_merge_skips_unknown_and_non_overridable() -> None:
    base = Settings()
    merged, skipped = merge(
        {
            "no_such_field": "x",  # orphan row from a removed setting
            "database_url": base.database_url,  # LOCKED A3 — read-only
            "embedding_dim": "512",  # LOCKED A3 — read-only
        }
    )
    keys = {key for key, _ in skipped}
    assert keys == {"no_such_field", "database_url", "embedding_dim"}
    # the base values stand — a hand-inserted row never hijacks them
    assert merged.database_url == base.database_url
    assert merged.embedding_dim == base.embedding_dim
    assert merged.embedding_dim == 768


def test_merge_drops_one_invalid_value_keeps_the_rest() -> None:
    base = Settings()
    merged, skipped = merge(
        {
            "relevance_threshold": "not-a-float",
            "llm_retries": "7",
        }
    )
    assert [key for key, _ in skipped] == ["relevance_threshold"]
    assert merged.llm_retries == 7
    # the bad row falls back to the env/base value
    assert merged.relevance_threshold == base.relevance_threshold


def test_merge_malformed_suggestions_json_is_skipped_not_fatal() -> None:
    base = Settings()
    merged, skipped = merge({"suggestions": "definitely not json"})
    assert [key for key, _ in skipped] == ["suggestions"]
    assert merged.suggestions == base.suggestions
    assert isinstance(merged.suggestions, list)


def test_merge_valid_suggestions_json_overrides() -> None:
    merged, skipped = merge({"suggestions": '["only this"]'})
    assert skipped == []
    assert merged.suggestions == ["only this"]


def test_merge_cross_field_violation_skipped_not_raised(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("BOR_RELEVANCE_THRESHOLD", "0.30")
    monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.15")
    # floor 0.9 > threshold 0.55 — the validator's cross-field guard
    # (app/config.py) fires; the FLOOR row is dropped, the threshold
    # override stands, the floor falls back to the env value.
    merged, skipped = merge(
        {"relevance_threshold": "0.55", "lexical_support_floor": "0.9"}
    )
    keys = {key for key, _ in skipped}
    assert "lexical_support_floor" in keys
    assert merged.relevance_threshold == 0.55
    assert merged.lexical_support_floor == 0.15  # env fallback


def test_merge_env_collision_cross_field_terminates(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # The env floor (0.15) is above the DB-overridden threshold (0.1):
    # the validator blames the FLOOR, which has no override — its value
    # comes from env and cannot be dropped. The resolver must find a
    # single override drop that heals the set (the threshold row)
    # instead of looping: one unresolvable collision costs the
    # colliding row, the rest of the table stands (LOCKED A5 fail-soft).
    monkeypatch.setenv("BOR_RELEVANCE_THRESHOLD", "0.30")
    monkeypatch.setenv("BOR_LEXICAL_SUPPORT_FLOOR", "0.15")
    base = Settings()
    merged, skipped = merge({"relevance_threshold": "0.1"})
    keys = {key for key, _ in skipped}
    assert "relevance_threshold" in keys
    assert merged.relevance_threshold == base.relevance_threshold
    assert merged.lexical_support_floor == base.lexical_support_floor


def test_merge_bool_spellings_behave_like_env() -> None:
    merged_false, skipped = merge({"stream_thinking": "false"})
    assert skipped == []
    assert merged_false.stream_thinking is False
    merged_zero, _ = merge({"stream_thinking": "0"})
    assert merged_zero.stream_thinking is False
    # untouched fields keep the base value
    base = Settings()
    assert merged_false.llm_retries == base.llm_retries


def test_merge_empty_rows_is_the_env_base() -> None:
    base = Settings()
    merged, skipped = merge({})
    assert skipped == []
    assert merged == base


def test_load_overrides_success_path(monkeypatch: pytest.MonkeyPatch) -> None:
    class _Rows:
        def __iter__(self):
            return iter((("llm_retries", "9"), ("stream_thinking", "false")))

    class _FakeConn:
        def execute(self, _stmt: object) -> _Rows:
            return _Rows()

        def __enter__(self):
            return self

        def __exit__(self, *exc: object) -> bool:
            return False

    class _FakeEngine:
        def connect(self) -> _FakeConn:  # type: ignore[no-untyped-def]
            return _FakeConn()

    monkeypatch.setattr(settings_overlay, "_get_engine", lambda: _FakeEngine())
    assert load_overrides() == {"llm_retries": "9", "stream_thinking": "false"}


def test_get_engine_creates_once_and_caches() -> None:
    # create_engine is lazy (no connection at creation) — the second
    # call must return the SAME engine (the once-per-process engine).
    first = settings_overlay._get_engine()
    second = settings_overlay._get_engine()
    assert first is second


# ---------------------------------------------------------------------------
# applied_override_keys — "which keys does the panel own?" (phase 190)
# ---------------------------------------------------------------------------


class _KeyResult:
    """The ``session.execute(select(AppSetting.key))`` shape — the one
    call :func:`applied_override_keys` makes on the caller's session."""

    def __init__(self, keys: list[str]) -> None:
        self._keys = keys

    def scalars(self) -> _KeyResult:
        return self

    def all(self) -> list[str]:
        return list(self._keys)


class _RecordingSession:
    """A stand-in session: it answers the key read and counts the calls
    (the resolver's "one read for all three string fields" contract)."""

    def __init__(self, keys: list[str]) -> None:
        self.keys = keys
        self.calls = 0

    def execute(self, _stmt: object) -> _KeyResult:
        self.calls += 1
        return _KeyResult(self.keys)


def test_applied_override_keys_filters_orphans_and_locked_rows(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from typing import Any, cast
    """Only a row the overlay would ACTUALLY apply counts: an orphan row
    (a field a later release removed) and a LOCKED A3 row (a
    hand-inserted ``database_url`` — read-only, never in effect) are
    excluded, so a row that changes nothing can never win a brand
    string from the Theme-tab fallback (phase 190 A1)."""
    monkeypatch.setenv("BOR_SETTINGS_DB", "1")
    session = _RecordingSession(
        ["app_name", "database_url", "embedding_dim", "no_such_setting"]
    )
    assert settings_overlay.applied_override_keys(
        cast(Any, session)
    ) == frozenset({"app_name"})
    assert session.calls == 1


def test_applied_override_keys_kill_switch_reads_nothing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from typing import Any, cast
    """``BOR_SETTINGS_DB=0`` means NO row is in effect (D6) — the helper
    says so WITHOUT touching the database, so ``effective_settings``
    falls through to the Theme-tab row instead of shadowing it with a
    row the resolver ignores."""
    monkeypatch.setenv("BOR_SETTINGS_DB", "0")
    session = _RecordingSession(["app_name"])
    assert settings_overlay.applied_override_keys(cast(Any, session)) == frozenset()
    assert session.calls == 0  # short-circuited before any statement


# ---------------------------------------------------------------------------
# merge — the defensive fallback (a validator change in a later release
# blames a field no override owns and no single drop heals the set: the
# env base stands, logged)
# ---------------------------------------------------------------------------


def test_merge_unresolvable_collision_returns_base_logged(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from typing import Any, cast

    from pydantic import ValidationError

    class _StubSettings:
        """Raises like a broken candidate set; the final no-args call
        (the env-base fallback) succeeds."""

        model_fields = Settings.model_fields  # merge() pre-checks it
        calls = 0

        def __init__(self, **_kwargs: object) -> None:
            _StubSettings.calls += 1
            if _StubSettings.calls < 3:
                # the runtime shape (msg/error keys) is wider than the
                # pyright stub's InitErrorDetails — cast it
                raise ValidationError.from_exception_data(
                    "Settings",
                    cast(
                        Any,
                        [
                            {
                                "type": "missing",
                                "loc": ("lexical_support_floor",),
                                "msg": "a later-release validator",
                                "error": ValueError("a later-release validator"),
                                "input_value": 0.35,
                            }
                        ],
                    ),
                )

    monkeypatch.setattr(settings_overlay, "Settings", _StubSettings)
    merged, skipped = merge({"relevance_threshold": "0.1"})
    assert merged is not None  # the env base (the stub's final call)
    reasons = {key: reason for key, reason in skipped}
    assert "unresolvable cross-field collision" in (
        reasons.get("relevance_threshold", "")
        or reasons.get("lexical_support_floor", "")
    )


# ---------------------------------------------------------------------------
# load_overrides — fail-soft, once-per-process (K3 + D5)
# ---------------------------------------------------------------------------


class _FailingEngine:
    """A stub engine whose connect() raises the given exception."""

    def __init__(self, exc: Exception) -> None:
        self._exc = exc

    def connect(self) -> None:  # type: ignore[no-untyped-def]
        raise self._exc


def test_load_overrides_generic_failure_warns_once(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    monkeypatch.setattr(
        settings_overlay,
        "_get_engine",
        lambda: _FailingEngine(
            OperationalError("SELECT 1", {}, OSError("connection refused"))
        ),
    )
    with caplog.at_level(logging.WARNING, logger="app.settings"):
        assert load_overrides() == {}
        assert load_overrides() == {}  # the second miss is silent
    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(warnings) == 1
    assert "settings overrides unavailable — env values in effect" in warnings[0].msg
    # the exception class is named, the value is not
    assert "OperationalError" in warnings[0].getMessage()


def test_load_overrides_missing_table_debugs_once(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    monkeypatch.setattr(
        settings_overlay,
        "_get_engine",
        lambda: _FailingEngine(
            ProgrammingError(
                "SELECT key, value FROM app_settings",
                {},
                Exception('relation "app_settings" does not exist'),
            )
        ),
    )
    with caplog.at_level(logging.DEBUG, logger="app.settings"):
        assert load_overrides() == {}
        assert load_overrides() == {}  # the second miss is silent
    debugs = [
        r
        for r in caplog.records
        if r.levelno == logging.DEBUG and "app_settings" in r.getMessage()
    ]
    assert len(debugs) == 1


def test_reset_state_restores_the_warning(
    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
    monkeypatch.setattr(
        settings_overlay,
        "_get_engine",
        lambda: _FailingEngine(
            OperationalError("SELECT 1", {}, OSError("connection refused"))
        ),
    )
    with caplog.at_level(logging.WARNING, logger="app.settings"):
        load_overrides()  # first miss: warns
        reset_state_for_tests()  # clears the once-per-process flag
        load_overrides()  # second miss: warns again
    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(warnings) == 2


# ---------------------------------------------------------------------------
# resolve_settings — the whole overlay, recursion guard (D2)
# ---------------------------------------------------------------------------


def test_resolve_settings_disabled_returns_env_base(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("BOR_SETTINGS_DB", "0")
    assert resolve_settings() == Settings()


def test_resolve_settings_enabled_uses_the_rows(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("BOR_SETTINGS_DB", "1")
    monkeypatch.setattr(
        settings_overlay, "load_overrides", lambda: {"llm_retries": "9"}
    )
    settings = resolve_settings()
    assert settings.llm_retries == 9


def test_resolve_settings_never_reaches_back_into_config() -> None:
    """The recursion guard, source-pinned: the overlay builds
    ``Settings()`` directly — app.config's resolver delegates HERE."""
    src = Path(settings_overlay.__file__).read_text()
    assert "get_settings(" not in src


# ---------------------------------------------------------------------------
# the invalidator registry (task 03's seam)
# ---------------------------------------------------------------------------


def test_invalidators_run_in_order_and_survive_a_raising_hook(
    caplog: pytest.LogCaptureFixture,
) -> None:
    calls: list[str] = []

    def _boom() -> None:
        raise RuntimeError("boom")

    register_invalidator(lambda: calls.append("first"))
    register_invalidator(_boom)
    register_invalidator(lambda: calls.append("third"))
    with caplog.at_level(logging.ERROR, logger="app.settings"):
        run_invalidators()  # must not raise
    assert calls == ["first", "third"]
    assert any(
        "settings invalidator failed" in r.getMessage() for r in caplog.records
    )


# ---------------------------------------------------------------------------
# LOCKED A1 — .env is never written by the app
# ---------------------------------------------------------------------------


def test_phase_files_never_write_the_env_file() -> None:
    """No ``open(`` / ``write_text(`` in the phase-168 modules — the
    panel writes Postgres only (LOCKED A1); ``app/api/admin_settings.py``
    (task 04) is pinned in the same way once it exists."""
    root = Path(settings_overlay.__file__).resolve().parents[2]
    for rel in (
        "app/core/settings_overlay.py",
        "app/config.py",
        "app/api/admin_settings.py",
    ):
        path = root / rel
        if not path.exists():
            continue
        src = path.read_text()
        assert "open(" not in src, (
            f"{rel} must never open() a file (LOCKED A1: .env is never "
            f"written by the app)"
        )
        assert "write_text(" not in src, f"{rel} must never write a file (LOCKED A1)"