"""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 )