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

"""Tests for thread_settings module."""

import threading
from unittest.mock import MagicMock

import pytest

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


class TestNoSettingsContextError:
    """Tests for NoSettingsContextError exception."""

    def test_is_exception(self):
        """Should be an Exception subclass."""
        assert issubclass(NoSettingsContextError, Exception)

    def test_can_be_raised_with_message(self):
        """Should accept error message."""
        with pytest.raises(NoSettingsContextError) as exc_info:
            raise NoSettingsContextError("test message")
        assert "test message" in str(exc_info.value)


class TestSetSettingsContext:
    """Tests for set_settings_context function."""

    def test_sets_context_on_thread_local(self, clean_thread_local):
        """Should set settings_context on thread local."""
        context = MagicMock()
        set_settings_context(context)
        assert _thread_local.settings_context is context

    def test_overwrites_previous_context(self, clean_thread_local):
        """Should overwrite previous context."""
        context1 = MagicMock()
        context2 = MagicMock()
        set_settings_context(context1)
        set_settings_context(context2)
        assert _thread_local.settings_context is context2


class TestGetSettingsContext:
    """Tests for get_settings_context function."""

    def test_returns_context_when_set(self, clean_thread_local):
        """Should return context when set."""
        context = MagicMock()
        set_settings_context(context)
        assert get_settings_context() is context

    def test_returns_none_when_not_set(self, clean_thread_local):
        """Should return None when no context set."""
        assert get_settings_context() is None


class TestGetSettingFromSnapshot:
    """Tests for get_setting_from_snapshot function."""

    def test_returns_value_from_snapshot(self, clean_thread_local):
        """Should return value from settings_snapshot."""
        snapshot = {"test.key": "test_value"}
        result = get_setting_from_snapshot(
            "test.key", settings_snapshot=snapshot
        )
        assert result == "test_value"

    def test_handles_full_format_value(self, clean_thread_local):
        """Should extract value from full format dict."""
        snapshot = {
            "test.key": {"value": "extracted_value", "ui_element": "text"}
        }
        result = get_setting_from_snapshot(
            "test.key", settings_snapshot=snapshot
        )
        assert result == "extracted_value"

    def test_returns_default_when_key_not_found(self, clean_thread_local):
        """Should return default when key not in snapshot."""
        snapshot = {"other.key": "value"}
        result = get_setting_from_snapshot(
            "test.key", default="default_val", settings_snapshot=snapshot
        )
        assert result == "default_val"

    def test_builds_dict_from_child_keys(self, clean_thread_local):
        """Should build dict from child keys."""
        snapshot = {
            "parent.child1": "value1",
            "parent.child2": "value2",
        }
        result = get_setting_from_snapshot("parent", settings_snapshot=snapshot)
        assert result == {"child1": "value1", "child2": "value2"}

    def test_extracts_value_from_child_key_full_format(
        self, clean_thread_local
    ):
        """Should extract value from child keys with full format."""
        snapshot = {
            "parent.child": {"value": "extracted", "ui_element": "text"},
        }
        result = get_setting_from_snapshot("parent", settings_snapshot=snapshot)
        assert result == {"child": "extracted"}

    def test_uses_thread_context_when_no_snapshot(
        self, clean_thread_local, mock_settings_context
    ):
        """Should use thread-local context when no snapshot."""
        set_settings_context(mock_settings_context)
        mock_settings_context.get_setting.return_value = "context_value"
        result = get_setting_from_snapshot("test.key")
        assert result == "context_value"
        mock_settings_context.get_setting.assert_called_with("test.key", None)

    def test_extracts_value_from_context_dict(
        self, clean_thread_local, mock_settings_context
    ):
        """Should extract value from context dict format."""
        set_settings_context(mock_settings_context)
        mock_settings_context.get_setting.return_value = {
            "value": "extracted_value"
        }
        result = get_setting_from_snapshot("test.key")
        assert result == "extracted_value"

    def test_raises_when_no_context_and_no_default(self, clean_thread_local):
        """Should raise NoSettingsContextError when no context and no default."""
        with pytest.raises(NoSettingsContextError) as exc_info:
            get_setting_from_snapshot("test.key")
        assert "test.key" in str(exc_info.value)

    def test_returns_default_instead_of_raising(self, clean_thread_local):
        """Should return default instead of raising when provided."""
        result = get_setting_from_snapshot("test.key", default="my_default")
        assert result == "my_default"

    def test_returns_none_when_snapshot_value_is_none(self, clean_thread_local):
        """Key present in snapshot with value None should return None,
        not raise NoSettingsContextError. Regression for #4208: the
        OpenAI embeddings test endpoint blew up here because
        embeddings.openai.dimensions defaults to JSON null."""
        snapshot = {
            "embeddings.openai.dimensions": {
                "value": None,
                "ui_element": "number",
            }
        }
        result = get_setting_from_snapshot(
            "embeddings.openai.dimensions",
            default=None,
            settings_snapshot=snapshot,
        )
        assert result is None

    def test_returns_none_when_simplified_snapshot_value_is_none(
        self, clean_thread_local
    ):
        """Simplified-format snapshot (raw value, not {value: ...} dict)
        with explicit None should also return None, not raise."""
        snapshot = {"some.key": None}
        result = get_setting_from_snapshot(
            "some.key", default=None, settings_snapshot=snapshot
        )
        assert result is None


class TestClearSettingsContext:
    """Tests for clear_settings_context function."""

    def test_clears_existing_context(self, clean_thread_local):
        """Should clear context when set."""
        set_settings_context(MagicMock())
        clear_settings_context()
        assert get_settings_context() is None

    def test_noop_when_no_context(self, clean_thread_local):
        """Should not raise when no context is set."""
        clear_settings_context()


class TestSettingsContextManager:
    """Tests for settings_context context manager."""

    def test_sets_and_clears_context(self, clean_thread_local):
        """Context should be set inside and cleared after the block."""
        ctx = MagicMock()
        with settings_context(ctx):
            assert get_settings_context() is ctx

        assert get_settings_context() is None

    def test_clears_on_exception(self, clean_thread_local):
        """Context should be cleared even when an exception occurs."""
        ctx = MagicMock()
        with pytest.raises(ValueError):
            with settings_context(ctx):
                raise ValueError("boom")

        assert get_settings_context() is None

    def test_works_in_thread(self, clean_thread_local):
        """Context manager should work correctly in a separate thread."""
        results = []

        def worker():
            ctx = MagicMock()
            ctx.name = "thread-ctx"
            with settings_context(ctx):
                results.append(get_settings_context().name)
            results.append(get_settings_context())

        t = threading.Thread(target=worker)
        t.start()
        t.join()

        assert results[0] == "thread-ctx"
        assert results[1] is None


class TestThreadIsolation:
    """Tests for thread isolation of settings context."""

    def test_contexts_isolated_between_threads(self, clean_thread_local):
        """Should isolate contexts between threads."""
        results = {}

        def thread_func(name, context_value):
            context = MagicMock()
            context.get_setting.return_value = context_value
            set_settings_context(context)
            # Small delay to ensure overlap
            import time

            time.sleep(0.01)
            results[name] = get_settings_context().get_setting("key", None)

        thread1 = threading.Thread(target=thread_func, args=("t1", "value1"))
        thread2 = threading.Thread(target=thread_func, args=("t2", "value2"))

        thread1.start()
        thread2.start()
        thread1.join()
        thread2.join()

        assert results["t1"] == "value1"
        assert results["t2"] == "value2"