learningcircuit--local-deep-research/tests/test_api_settings_advanced.py

"""Advanced tests for programmatic API settings functionality."""

import threading

from local_deep_research.api.settings_utils import (
    InMemorySettingsManager,
    create_settings_snapshot,
    extract_setting_value,
)


class TestSettingsEdgeCases:
    """Test edge cases and error handling for settings."""

    def test_deeply_nested_settings_override(self):
        """Test overriding deeply nested settings."""
        snapshot = create_settings_snapshot(
            overrides={
                "search.engines.arxiv.enabled": True,
                "search.engines.wikipedia.max_results": 20,
                "llm.anthropic.model": "claude-3-opus-20240229",
            }
        )

        # Check nested settings were created properly
        assert "search.engines.arxiv.enabled" in snapshot
        assert snapshot["search.engines.arxiv.enabled"]["value"] is True

        assert "search.engines.wikipedia.max_results" in snapshot
        assert snapshot["search.engines.wikipedia.max_results"]["value"] == 20

        assert "llm.anthropic.model" in snapshot
        assert (
            snapshot["llm.anthropic.model"]["value"] == "claude-3-opus-20240229"
        )

    def test_invalid_settings_handling(self):
        """Test handling of invalid settings."""
        manager = InMemorySettingsManager()

        # Test various invalid operations
        assert manager.get_setting(None, "default") == "default"
        assert manager.get_setting("", "default") == "default"
        assert manager.set_setting(None, "value") is False
        assert manager.set_setting("", "value") is False
        assert manager.delete_setting(None) is False
        assert manager.delete_setting("") is False

    def test_settings_type_preservation(self):
        """Test that setting types are preserved correctly."""
        snapshot = create_settings_snapshot(
            overrides={
                "llm.temperature": 0.7,  # float
                "search.max_results": 10,  # int
                "search.safe_search": True,  # bool
                "llm.provider": "openai",  # string
                "custom.list": ["a", "b", "c"],  # list
                "custom.dict": {"key": "value"},  # dict
            }
        )

        # Check types are preserved
        assert isinstance(snapshot["llm.temperature"]["value"], float)
        assert isinstance(snapshot["search.max_results"]["value"], int)
        assert isinstance(snapshot["search.safe_search"]["value"], bool)
        assert isinstance(snapshot["llm.provider"]["value"], str)
        assert isinstance(snapshot["custom.list"]["value"], list)
        assert isinstance(snapshot["custom.dict"]["value"], dict)

    def test_settings_immutability(self):
        """Test that returned settings are immutable copies."""
        manager = InMemorySettingsManager()
        settings1 = manager.get_all_settings()
        settings2 = manager.get_all_settings()

        # Modify settings1
        if "llm.provider" in settings1:
            settings1["llm.provider"]["value"] = "modified"

        # settings2 should not be affected
        assert settings2["llm.provider"]["value"] != "modified"

    def test_empty_and_none_values(self):
        """Test handling of empty and None values."""
        snapshot = create_settings_snapshot(
            overrides={
                "empty.string": "",
                "none.value": None,
                "zero.value": 0,
                "false.value": False,
                "empty.list": [],
                "empty.dict": {},
            }
        )

        assert snapshot["empty.string"]["value"] == ""
        assert snapshot["none.value"]["value"] is None
        assert snapshot["zero.value"]["value"] == 0
        assert snapshot["false.value"]["value"] is False
        assert snapshot["empty.list"]["value"] == []
        assert snapshot["empty.dict"]["value"] == {}


class TestSettingsThreadSafety:
    """Test thread safety of settings management."""

    def test_concurrent_settings_access(self):
        """Test concurrent access to settings doesn't cause issues."""
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: replace with deterministic concurrent invariant test).
        manager = InMemorySettingsManager()
        results = []
        errors = []

        def access_settings(thread_id):
            try:
                # Each thread tries to read and write settings
                for i in range(10):
                    # Read
                    _ = manager.get_setting("llm.temperature")

                    # Write (if setting exists)
                    new_value = 0.1 * thread_id + 0.01 * i
                    manager.set_setting("llm.temperature", new_value)

                    # Read again
                    read_value = manager.get_setting("llm.temperature")
                    results.append((thread_id, read_value))
            except Exception as e:
                errors.append((thread_id, str(e)))

        # Create multiple threads
        threads = []
        for i in range(5):
            t = threading.Thread(target=access_settings, args=(i,))
            threads.append(t)
            t.start()

        # Wait for all threads
        for t in threads:
            t.join()

        # Should have no errors
        assert len(errors) == 0

        # Should have results from all threads
        assert len(results) > 0

    def test_settings_snapshot_isolation(self):
        """Test that settings snapshots are isolated between threads."""
        snapshots = {}

        def create_snapshot(thread_id):
            # Each thread creates its own snapshot with unique settings
            snapshot = create_settings_snapshot(
                provider=f"provider_{thread_id}",
                temperature=0.1 * thread_id,
                overrides={f"thread.{thread_id}.custom": f"value_{thread_id}"},
            )
            snapshots[thread_id] = snapshot

        # Create snapshots in parallel
        threads = []
        for i in range(5):
            t = threading.Thread(target=create_snapshot, args=(i,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        # Verify each snapshot has its unique values
        for thread_id, snapshot in snapshots.items():
            assert snapshot["llm.provider"]["value"] == f"provider_{thread_id}"
            assert snapshot["llm.temperature"]["value"] == 0.1 * thread_id
            assert (
                snapshot[f"thread.{thread_id}.custom"]["value"]
                == f"value_{thread_id}"
            )


class TestSettingsIntegration:
    """Test integration between settings and other components."""

    def test_settings_propagation_through_api(self):
        """Test that settings are properly propagated through the API."""
        # Create custom settings
        settings_snapshot = create_settings_snapshot(
            provider="anthropic",
            temperature=0.3,
            overrides={
                "llm.anthropic.api_key": "test-key",
                "llm.anthropic.model": "claude-3-opus-20240229",
                "search.tool": "searxng",
                "search.max_results": 15,
                "search.region": "us-en",
            },
        )

        # Test that settings can be extracted correctly
        assert (
            extract_setting_value(settings_snapshot, "llm.provider")
            == "anthropic"
        )
        assert (
            extract_setting_value(settings_snapshot, "llm.temperature") == 0.3
        )
        assert (
            extract_setting_value(settings_snapshot, "llm.anthropic.api_key")
            == "test-key"
        )
        assert (
            extract_setting_value(settings_snapshot, "search.tool") == "searxng"
        )
        assert (
            extract_setting_value(settings_snapshot, "search.max_results") == 15
        )

        # Test that settings structure is correct for use by components
        assert "llm.provider" in settings_snapshot
        assert isinstance(settings_snapshot["llm.provider"], dict)
        assert "value" in settings_snapshot["llm.provider"]

        # Test settings isolation
        snapshot2 = create_settings_snapshot(provider="openai", temperature=0.9)

        # Original snapshot should not be affected
        assert (
            extract_setting_value(settings_snapshot, "llm.provider")
            == "anthropic"
        )
        assert (
            extract_setting_value(settings_snapshot, "llm.temperature") == 0.3
        )
        assert extract_setting_value(snapshot2, "llm.provider") == "openai"
        assert extract_setting_value(snapshot2, "llm.temperature") == 0.9

    def test_extract_setting_value_edge_cases(self):
        """Test extract_setting_value with various edge cases."""
        # Test with None snapshot
        assert extract_setting_value(None, "any.key", "default") == "default"

        # Test with empty snapshot
        assert extract_setting_value({}, "any.key", "default") == "default"

        # Test with nested structures
        complex_snapshot = {
            "simple": "direct_value",
            "nested": {"value": "nested_value", "extra": "data"},
            "deep": {"value": {"nested": {"value": "deep_value"}}},
            "list": {"value": [1, 2, 3]},
            "null": {"value": None},
        }

        assert (
            extract_setting_value(complex_snapshot, "simple") == "direct_value"
        )
        assert (
            extract_setting_value(complex_snapshot, "nested") == "nested_value"
        )
        assert extract_setting_value(complex_snapshot, "deep") == {
            "nested": {"value": "deep_value"}
        }
        assert extract_setting_value(complex_snapshot, "list") == [1, 2, 3]
        assert extract_setting_value(complex_snapshot, "null") is None


class TestComplexSettingsScenarios:
    """Test complex real-world settings scenarios."""

    def test_multiple_provider_configuration(self):
        """Test configuring multiple LLM providers in one snapshot."""
        snapshot = create_settings_snapshot(
            provider="openai",  # Primary provider
            api_key="openai-key",
            overrides={
                # Configure multiple providers
                "llm.anthropic.api_key": "anthropic-key",
                "llm.anthropic.model": "claude-3-opus-20240229",
                "llm.ollama.base_url": "http://localhost:11434",
                "llm.ollama.model": "llama2",
                # Provider-specific settings
                "llm.openai.model": "gpt-4",
                "llm.openai.max_tokens": 4000,
                "llm.anthropic.max_tokens": 3000,
            },
        )

        # Verify all providers are configured
        assert snapshot["llm.provider"]["value"] == "openai"
        assert snapshot["llm.openai.api_key"]["value"] == "openai-key"
        assert snapshot["llm.openai.model"]["value"] == "gpt-4"
        assert snapshot["llm.openai.max_tokens"]["value"] == 4000

        assert snapshot["llm.anthropic.api_key"]["value"] == "anthropic-key"
        assert (
            snapshot["llm.anthropic.model"]["value"] == "claude-3-opus-20240229"
        )
        assert snapshot["llm.anthropic.max_tokens"]["value"] == 3000

        assert (
            snapshot["llm.ollama.base_url"]["value"] == "http://localhost:11434"
        )
        assert snapshot["llm.ollama.model"]["value"] == "llama2"

    def test_search_engine_configuration_matrix(self):
        """Test configuring multiple search engines with different settings."""
        snapshot = create_settings_snapshot(
            overrides={
                # Enable/disable specific engines
                "search.engines.searxng.enabled": True,
                "search.engines.duckduckgo.enabled": True,
                "search.engines.wikipedia.enabled": True,
                "search.engines.arxiv.enabled": False,
                # Engine-specific settings
                "search.engines.searxng.base_url": "https://searx.example.com",
                "search.engines.searxng.timeout": 10,
                "search.engines.duckduckgo.region": "us-en",
                "search.engines.duckduckgo.safe_search": True,
                "search.engines.wikipedia.language": "en",
                "search.engines.wikipedia.max_chars": 1000,
                # Global search settings
                "search.max_results": 20,
                "search.max_filtered_results": 10,
                "search.snippets_only": False,
            }
        )

        # Verify engine configurations
        assert snapshot["search.engines.searxng.enabled"]["value"] is True
        assert (
            snapshot["search.engines.searxng.base_url"]["value"]
            == "https://searx.example.com"
        )
        assert (
            snapshot["search.engines.duckduckgo.safe_search"]["value"] is True
        )
        assert snapshot["search.engines.arxiv.enabled"]["value"] is False

        # Verify global settings
        assert snapshot["search.max_results"]["value"] == 20
        assert snapshot["search.snippets_only"]["value"] is False

    def test_research_mode_specific_settings(self):
        """Test settings for different research modes."""
        # Quick mode settings
        quick_snapshot = create_settings_snapshot(
            overrides={
                "research.quick.max_iterations": 1,
                "research.quick.max_search_queries": 3,
                "research.quick.enable_deep_analysis": False,
                "llm.temperature": 0.7,  # Higher for creativity
            }
        )

        # Detailed mode settings
        detailed_snapshot = create_settings_snapshot(
            overrides={
                "research.detailed.max_iterations": 5,
                "research.detailed.max_search_queries": 10,
                "research.detailed.enable_deep_analysis": True,
                "llm.temperature": 0.3,  # Lower for accuracy
                "search.max_results": 30,  # More results for detailed research
            }
        )

        # Verify different configurations
        assert quick_snapshot["research.quick.max_iterations"]["value"] == 1
        assert quick_snapshot["llm.temperature"]["value"] == 0.7

        assert (
            detailed_snapshot["research.detailed.max_iterations"]["value"] == 5
        )
        assert detailed_snapshot["llm.temperature"]["value"] == 0.3
        assert detailed_snapshot["search.max_results"]["value"] == 30

    def test_environment_variable_override_priority(self, monkeypatch):
        """Test that env vars have correct priority over defaults and overrides."""
        # Set environment variables
        monkeypatch.setenv("LDR_LLM_PROVIDER", "env_provider")
        monkeypatch.setenv("LDR_LLM_TEMPERATURE", "0.9")

        # Create manager (picks up env vars)
        manager = InMemorySettingsManager()

        # Env vars should override defaults
        assert manager.get_setting("llm.provider") == "env_provider"
        assert manager.get_setting("llm.temperature") == 0.9

        # But explicit overrides should take precedence over env vars
        snapshot = create_settings_snapshot(
            provider="explicit_provider", temperature=0.1
        )

        assert snapshot["llm.provider"]["value"] == "explicit_provider"
        assert snapshot["llm.temperature"]["value"] == 0.1


class TestSettingsPersistence:
    """Test settings persistence and isolation."""

    def test_in_memory_settings_not_persisted(self):
        """Test that in-memory settings are not persisted between instances."""
        # Create manager and modify settings
        manager1 = InMemorySettingsManager()
        original_temp = manager1.get_setting("llm.temperature")
        manager1.set_setting("llm.temperature", 0.999)

        # Create new manager
        manager2 = InMemorySettingsManager()

        # Should have original default, not modified value
        assert manager2.get_setting("llm.temperature") == original_temp
        assert manager2.get_setting("llm.temperature") != 0.999

    def test_settings_snapshot_independence(self):
        """Test that settings snapshots are independent of each other."""
        # Create base snapshot
        base_snapshot = create_settings_snapshot(provider="openai")

        # Create another snapshot with same base
        snapshot1 = create_settings_snapshot(
            base_settings=base_snapshot, temperature=0.5
        )

        # Create third snapshot with same base
        snapshot2 = create_settings_snapshot(
            base_settings=base_snapshot, temperature=0.9
        )

        # Verify independence
        assert base_snapshot["llm.provider"]["value"] == "openai"
        assert (
            "llm.temperature" not in base_snapshot
            or base_snapshot["llm.temperature"]["value"] != 0.5
        )
        assert snapshot1["llm.temperature"]["value"] == 0.5
        assert snapshot2["llm.temperature"]["value"] == 0.9

        # Modifying one shouldn't affect others
        snapshot1["llm.provider"]["value"] = "modified"
        assert base_snapshot["llm.provider"]["value"] == "openai"
        assert snapshot2["llm.provider"]["value"] == "openai"