learningcircuit--local-deep-research/tests/config/test_thread_settings_extended.py

"""Extended tests for the thread_settings module.

Covers thread-context management, get_setting_from_snapshot logic,
and get_bool_setting_from_snapshot boolean conversion.
"""

import threading
from unittest.mock import MagicMock, patch

import pytest

from local_deep_research.config.thread_settings import (
    NoSettingsContextError,
    clear_settings_context,
    get_bool_setting_from_snapshot,
    get_setting_from_snapshot,
    get_settings_context,
    set_settings_context,
    settings_context,
)


# ---------------------------------------------------------------------------
# Thread context management
# ---------------------------------------------------------------------------


class TestThreadContextManagement:
    """Tests for set/get/clear thread-local settings context."""

    def test_set_get_round_trip(self, clean_thread_local):
        """Setting a context and immediately getting it returns the same object."""
        ctx = MagicMock()
        set_settings_context(ctx)
        assert get_settings_context() is ctx

    def test_clear_removes_context(self, clean_thread_local):
        """After clearing, get_settings_context returns None."""
        set_settings_context(MagicMock())
        clear_settings_context()
        assert get_settings_context() is None

    def test_clear_when_nothing_set_does_not_error(self, clean_thread_local):
        """Clearing without a prior set must not raise."""
        clear_settings_context()  # should be a no-op

    def test_get_returns_none_when_not_set(self, clean_thread_local):
        """get_settings_context returns None when no context has been set."""
        assert get_settings_context() is None

    def test_settings_context_manager_sets_and_clears(self, clean_thread_local):
        """The settings_context context manager sets context on entry and
        clears it on exit."""
        ctx = MagicMock()
        with settings_context(ctx):
            assert get_settings_context() is ctx
        assert get_settings_context() is None

    def test_settings_context_manager_clears_on_exception(
        self, clean_thread_local
    ):
        """Context is cleared even when the body raises an exception."""
        ctx = MagicMock()
        with pytest.raises(RuntimeError):
            with settings_context(ctx):
                raise RuntimeError("boom")
        assert get_settings_context() is None

    def test_overwrite_previous_context(self, clean_thread_local):
        """Setting a second context replaces the first."""
        ctx1 = MagicMock()
        ctx2 = MagicMock()
        set_settings_context(ctx1)
        set_settings_context(ctx2)
        assert get_settings_context() is ctx2

    def test_thread_isolation(self, clean_thread_local):
        """Contexts set in different threads are independent."""
        barrier = threading.Barrier(2)
        results = {}

        def worker(name, value):
            ctx = MagicMock()
            ctx.value = value
            set_settings_context(ctx)
            barrier.wait()  # force both threads to overlap
            results[name] = get_settings_context().value

        t1 = threading.Thread(target=worker, args=("t1", "val1"))
        t2 = threading.Thread(target=worker, args=("t2", "val2"))
        t1.start()
        t2.start()
        t1.join()
        t2.join()

        assert results["t1"] == "val1"
        assert results["t2"] == "val2"


# ---------------------------------------------------------------------------
# get_setting_from_snapshot
# ---------------------------------------------------------------------------


class TestGetSettingFromSnapshotDirect:
    """Tests for direct key lookups in get_setting_from_snapshot."""

    def test_direct_value_from_snapshot(self, clean_thread_local):
        """A plain (non-dict) value stored under the exact key is returned."""
        snapshot = {"my_key": "hello"}
        assert (
            get_setting_from_snapshot("my_key", settings_snapshot=snapshot)
            == "hello"
        )

    def test_direct_integer_value(self, clean_thread_local):
        """Integer values are returned without coercion."""
        snapshot = {"count": 42}
        assert (
            get_setting_from_snapshot("count", settings_snapshot=snapshot) == 42
        )

    @patch("local_deep_research.config.thread_settings.get_typed_setting_value")
    def test_dict_format_with_value_key(self, mock_gtsv, clean_thread_local):
        """A dict containing 'value' triggers get_typed_setting_value."""
        mock_gtsv.return_value = "typed_result"
        snapshot = {"key": {"value": "raw", "ui_element": "select"}}
        result = get_setting_from_snapshot("key", settings_snapshot=snapshot)
        mock_gtsv.assert_called_once_with("key", "raw", "select")
        assert result == "typed_result"

    @patch("local_deep_research.config.thread_settings.get_typed_setting_value")
    def test_dict_format_defaults_ui_element_to_text(
        self, mock_gtsv, clean_thread_local
    ):
        """When ui_element is missing from the dict, 'text' is used."""
        mock_gtsv.return_value = "typed"
        snapshot = {"k": {"value": 10}}
        get_setting_from_snapshot("k", settings_snapshot=snapshot)
        mock_gtsv.assert_called_once_with("k", 10, "text")

    def test_raw_non_dict_value(self, clean_thread_local):
        """A non-dict raw value (e.g. list) is returned as-is."""
        snapshot = {"tags": ["a", "b"]}
        result = get_setting_from_snapshot("tags", settings_snapshot=snapshot)
        assert result == ["a", "b"]

    def test_dict_without_value_key_returned_as_is(self, clean_thread_local):
        """A dict that does NOT contain a 'value' key is returned unchanged
        (it is treated as a raw value, not as the full-format dict)."""
        snapshot = {"cfg": {"host": "localhost", "port": 8080}}
        result = get_setting_from_snapshot("cfg", settings_snapshot=snapshot)
        assert result == {"host": "localhost", "port": 8080}


class TestGetSettingFromSnapshotChildKeys:
    """Tests for child-key aggregation logic."""

    def test_child_key_aggregation(self, clean_thread_local):
        """Keys starting with 'parent.' are collected into a dict."""
        snapshot = {
            "parent.sub1": "v1",
            "parent.sub2": "v2",
        }
        result = get_setting_from_snapshot("parent", settings_snapshot=snapshot)
        assert result == {"sub1": "v1", "sub2": "v2"}

    def test_single_child_key(self, clean_thread_local):
        """A single matching child key still produces a dict."""
        snapshot = {"section.only_child": 99}
        result = get_setting_from_snapshot(
            "section", settings_snapshot=snapshot
        )
        assert result == {"only_child": 99}

    @patch("local_deep_research.config.thread_settings.get_typed_setting_value")
    def test_child_key_dict_format(self, mock_gtsv, clean_thread_local):
        """Child keys in full dict format trigger get_typed_setting_value."""
        mock_gtsv.return_value = "typed_child"
        snapshot = {
            "parent.child": {"value": "raw_child", "ui_element": "checkbox"},
        }
        result = get_setting_from_snapshot("parent", settings_snapshot=snapshot)
        mock_gtsv.assert_called_once_with("child", "raw_child", "checkbox")
        assert result == {"child": "typed_child"}

    def test_no_partial_prefix_confusion(self, clean_thread_local):
        """'search_extra.x' must NOT match parent key 'search'."""
        snapshot = {
            "search_extra.engine": "nope",
            "search.engine": "yes",
        }
        result = get_setting_from_snapshot("search", settings_snapshot=snapshot)
        assert result == {"engine": "yes"}

    def test_malformed_child_keys_are_skipped(self, clean_thread_local):
        """Malformed child rows (legacy 'parent.' / 'parent..' / 'parent. ')
        must not be aggregated — otherwise a stray key wraps a leaf read into
        an [object Object] dict (#4840). Mirrors the DB read path."""
        snapshot = {
            "parent.real": "v",
            "parent.": "junk",  # trailing dot
            "parent..": "junk",  # empty segment
            "parent. ": "junk",  # whitespace segment
        }
        result = get_setting_from_snapshot("parent", settings_snapshot=snapshot)
        assert result == {"real": "v"}

    def test_only_malformed_child_returns_default(self, clean_thread_local):
        """A prefix whose only matching row is malformed resolves to the
        default, not a `{'.': ...}` wrapper dict."""
        snapshot = {"parent..": "junk"}
        result = get_setting_from_snapshot(
            "parent", settings_snapshot=snapshot, default="fallback"
        )
        assert result == "fallback"


class TestGetSettingFromSnapshotThreadContext:
    """Tests for falling back to the thread-local settings context."""

    def test_falls_back_to_thread_context(self, clean_thread_local):
        """When key not in snapshot, the thread context is consulted."""
        ctx = MagicMock()
        ctx.get_setting.return_value = "from_ctx"
        set_settings_context(ctx)
        result = get_setting_from_snapshot(
            "missing_key", settings_snapshot={"other": 1}
        )
        assert result == "from_ctx"
        ctx.get_setting.assert_called_once_with("missing_key", None)

    def test_thread_context_dict_format_extraction(self, clean_thread_local):
        """If the thread context returns a dict with 'value', the inner value
        is extracted."""
        ctx = MagicMock()
        ctx.get_setting.return_value = {"value": "inner"}
        set_settings_context(ctx)
        result = get_setting_from_snapshot("key")
        assert result == "inner"

    def test_thread_context_plain_value(self, clean_thread_local):
        """Thread context returning a plain value passes it through."""
        ctx = MagicMock()
        ctx.get_setting.return_value = 7
        set_settings_context(ctx)
        result = get_setting_from_snapshot("key")
        assert result == 7

    def test_thread_context_with_default_passes_default(
        self, clean_thread_local
    ):
        """The default argument is forwarded to context.get_setting."""
        ctx = MagicMock()
        ctx.get_setting.return_value = "fallback"
        set_settings_context(ctx)
        get_setting_from_snapshot("k", default="fallback")
        ctx.get_setting.assert_called_once_with("k", "fallback")

    def test_empty_snapshot_falls_back_to_context(self, clean_thread_local):
        """An empty snapshot dict (truthy={} is falsy) skips snapshot lookup
        and goes to thread context."""
        ctx = MagicMock()
        ctx.get_setting.return_value = "ctx_val"
        set_settings_context(ctx)
        # Note: {} is falsy in Python, so `if settings_snapshot` is False
        result = get_setting_from_snapshot("k", settings_snapshot={})
        assert result == "ctx_val"


class TestGetSettingFromSnapshotDefaults:
    """Tests for default value handling and error raising."""

    def test_returns_default_when_nothing_found(self, clean_thread_local):
        """When snapshot misses, no context, and a default is given, return it."""
        result = get_setting_from_snapshot("key", default="def_val")
        assert result == "def_val"

    def test_raises_no_settings_context_error_when_no_default(
        self, clean_thread_local
    ):
        """With no snapshot, no context, and no default, raises
        NoSettingsContextError."""
        with pytest.raises(NoSettingsContextError, match="key_xyz"):
            get_setting_from_snapshot("key_xyz")

    def test_none_snapshot_and_no_context_with_default(
        self, clean_thread_local
    ):
        """None snapshot + no context + default provided -> returns default."""
        result = get_setting_from_snapshot(
            "x", default="safe", settings_snapshot=None
        )
        assert result == "safe"

    def test_none_default_still_raises(self, clean_thread_local):
        """Explicitly passing default=None still raises (it is the sentinel
        checked via `if default is not None`)."""
        with pytest.raises(NoSettingsContextError):
            get_setting_from_snapshot("missing", default=None)


# ---------------------------------------------------------------------------
# get_bool_setting_from_snapshot
# ---------------------------------------------------------------------------


class TestGetBoolSettingFromSnapshot:
    """Tests for get_bool_setting_from_snapshot boolean conversion."""

    @patch("local_deep_research.config.thread_settings.to_bool")
    def test_calls_to_bool_with_value_and_default(
        self, mock_to_bool, clean_thread_local
    ):
        """Should pass the retrieved value and the default to to_bool."""
        mock_to_bool.return_value = True
        snapshot = {"flag": "yes"}
        get_bool_setting_from_snapshot(
            "flag", default=False, settings_snapshot=snapshot
        )
        mock_to_bool.assert_called_once_with("yes", False)

    def test_converts_string_true_to_true(self, clean_thread_local):
        """String 'true' is converted to boolean True."""
        snapshot = {"flag": "true"}
        assert (
            get_bool_setting_from_snapshot("flag", settings_snapshot=snapshot)
            is True
        )

    def test_converts_string_false_to_false(self, clean_thread_local):
        """String 'false' is converted to boolean False."""
        snapshot = {"flag": "false"}
        assert (
            get_bool_setting_from_snapshot("flag", settings_snapshot=snapshot)
            is False
        )

    def test_uses_default_when_not_found(self, clean_thread_local):
        """When key is missing, the default boolean is returned."""
        result = get_bool_setting_from_snapshot(
            "missing", default=True, settings_snapshot={"other": 1}
        )
        assert result is True

    def test_passes_through_boolean_true(self, clean_thread_local):
        """A native boolean True in the snapshot passes through unchanged."""
        snapshot = {"flag": True}
        assert (
            get_bool_setting_from_snapshot("flag", settings_snapshot=snapshot)
            is True
        )

    def test_passes_through_boolean_false(self, clean_thread_local):
        """A native boolean False in the snapshot passes through unchanged."""
        snapshot = {"flag": False}
        # False is falsy but still a valid value for the snapshot; however
        # the code checks `if value is not None` before returning so False
        # goes through the normal return -> to_bool(False, default) -> False
        assert (
            get_bool_setting_from_snapshot(
                "flag", default=True, settings_snapshot=snapshot
            )
            is False
        )

    def test_converts_string_one_to_true(self, clean_thread_local):
        """String '1' is converted to boolean True by to_bool."""
        snapshot = {"flag": "1"}
        assert (
            get_bool_setting_from_snapshot("flag", settings_snapshot=snapshot)
            is True
        )

    def test_converts_string_zero_to_false(self, clean_thread_local):
        """String '0' is converted to boolean False by to_bool."""
        snapshot = {"flag": "0"}
        assert (
            get_bool_setting_from_snapshot("flag", settings_snapshot=snapshot)
            is False
        )