learningcircuit--local-deep-research/tests/utilities/test_llm_utils.py

"""Tests for llm_utils module."""

from unittest.mock import Mock, patch

from local_deep_research.utilities.llm_utils import (
    fetch_ollama_models,
    get_model_identifier,
    get_ollama_base_url,
    get_server_url,
)


class _FakeModelName:
    def __init__(self, model_name):
        self.model_name = model_name


class _FakeModelOnly:
    def __init__(self, model):
        self.model = model


class _FakeBoth:
    def __init__(self, model_name, model):
        self.model_name = model_name
        self.model = model


class _FakeNeither:
    pass


class _FakeWrapper:
    def __init__(self, base_llm):
        self.base_llm = base_llm


class TestGetModelIdentifier:
    """Regression tests for the identifier helper that drives the
    Tier 4 LLM cache predicate. Earlier code used ``getattr(llm, "name",
    str(llm))`` which always fell back to ``str(wrapper)`` — a repr
    string that poisoned the quality_model column."""

    def test_prefers_model_name(self):
        assert get_model_identifier(_FakeModelName("gpt-4")) == "gpt-4"

    def test_falls_back_to_model(self):
        assert (
            get_model_identifier(_FakeModelOnly("claude-opus-4-7"))
            == "claude-opus-4-7"
        )

    def test_prefers_model_name_over_model(self):
        # If both exist, model_name wins. ChatOpenAI / ChatAnthropic
        # expose model_name; ChatOllama exposes model.
        llm = _FakeBoth(model_name="gpt-4", model="ignored")
        assert get_model_identifier(llm) == "gpt-4"

    def test_class_name_fallback(self):
        # Never return a repr() with object address — fall back cleanly.
        result = get_model_identifier(_FakeNeither())
        assert result == "_FakeNeither"
        assert "object at 0x" not in result

    def test_unwraps_processing_wrapper(self):
        wrapped = _FakeWrapper(_FakeModelOnly("mistral"))
        assert get_model_identifier(wrapped) == "mistral"

    def test_none_values_skipped(self):
        class _NonePair:
            model_name = None
            model = "real-model"

        assert get_model_identifier(_NonePair()) == "real-model"


class TestGetOllamaBaseUrl:
    """Tests for get_ollama_base_url function."""

    def test_returns_default_without_settings(self):
        """Should return default URL when no settings provided."""
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:
            mock_get.return_value = None
            result = get_ollama_base_url()
            assert result == "http://localhost:11434"

    def test_uses_embeddings_ollama_url(self):
        """Should use embeddings.ollama.url setting."""
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:
            mock_get.return_value = "http://custom:11434"
            result = get_ollama_base_url()
            assert result == "http://custom:11434"

    def test_normalizes_url(self):
        """Should normalize URL without scheme."""
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:
            mock_get.return_value = "localhost:11434"
            result = get_ollama_base_url()
            assert result == "http://localhost:11434"

    def test_passes_settings_snapshot(self):
        """Should pass settings snapshot to get_setting_from_snapshot."""
        snapshot = {"key": "value"}
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:
            mock_get.return_value = "http://localhost:11434"
            get_ollama_base_url(settings_snapshot=snapshot)
            # Should have been called with snapshot
            assert any(
                call.kwargs.get("settings_snapshot") == snapshot
                for call in mock_get.call_args_list
            )


class TestGetServerUrl:
    """Tests for get_server_url function."""

    def test_returns_default_without_settings(self):
        """Should return default URL when no settings provided."""
        result = get_server_url()
        assert result == "http://127.0.0.1:5000/"

    def test_uses_server_url_from_snapshot(self):
        """Should use direct server_url from snapshot."""
        snapshot = {"server_url": "https://custom.example.com/"}
        result = get_server_url(settings_snapshot=snapshot)
        assert result == "https://custom.example.com/"

    def test_uses_system_server_url(self):
        """Should use system.server_url setting."""
        snapshot = {"system": {"server_url": "https://system.example.com/"}}
        result = get_server_url(settings_snapshot=snapshot)
        assert result == "https://system.example.com/"

    def test_constructs_url_from_web_settings(self):
        """Should construct URL from web.host, web.port, web.use_https."""
        # Need to provide snapshot so it goes through web settings path
        snapshot = {"_trigger_web_settings": True}
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:

            def side_effect(key, snapshot_arg=None, default=None):
                settings = {
                    "web.host": "0.0.0.0",
                    "web.port": 8080,
                    "web.use_https": True,
                }
                return settings.get(key, default)

            mock_get.side_effect = side_effect
            result = get_server_url(settings_snapshot=snapshot)
            # 0.0.0.0 should be converted to 127.0.0.1
            assert result == "https://127.0.0.1:8080/"

    def test_uses_http_when_use_https_false(self):
        """Should use http scheme when use_https is False."""
        snapshot = {"_trigger_web_settings": True}
        with patch(
            "local_deep_research.utilities.llm_utils.get_setting_from_snapshot"
        ) as mock_get:

            def side_effect(key, snapshot_arg=None, default=None):
                settings = {
                    "web.host": "192.168.1.1",  # Not localhost, so it won't default
                    "web.port": 5000,
                    "web.use_https": False,
                }
                return settings.get(key, default)

            mock_get.side_effect = side_effect
            result = get_server_url(settings_snapshot=snapshot)
            assert result == "http://192.168.1.1:5000/"

    def test_priority_order(self):
        """Should check server_url before system before web settings."""
        # Direct server_url takes priority
        snapshot = {
            "server_url": "https://direct/",
            "system": {"server_url": "https://system/"},
        }
        result = get_server_url(settings_snapshot=snapshot)
        assert result == "https://direct/"

    def test_returns_fallback_with_trailing_slash(self):
        """Fallback URL should have trailing slash."""
        result = get_server_url()
        assert result.endswith("/")


class TestFetchOllamaModels:
    """Tests for fetch_ollama_models function."""

    def test_returns_empty_list_on_connection_error(self):
        """Should return empty list on connection error."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_get.side_effect = Exception("Connection refused")
            result = fetch_ollama_models("http://localhost:11434")
            assert result == []

    def test_returns_empty_list_on_non_200(self):
        """Should return empty list on non-200 status."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 404
            mock_get.return_value = mock_response
            result = fetch_ollama_models("http://localhost:11434")
            assert result == []

    def test_parses_models_from_response(self):
        """Should parse models from API response."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {
                "models": [
                    {"name": "llama2"},
                    {"name": "mistral"},
                ]
            }
            mock_get.return_value = mock_response

            result = fetch_ollama_models("http://localhost:11434")

            assert len(result) == 2
            assert {"value": "llama2", "label": "llama2"} in result
            assert {"value": "mistral", "label": "mistral"} in result

    def test_handles_older_api_format(self):
        """Should handle older API format (list directly)."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = [
                {"name": "model1"},
                {"name": "model2"},
            ]
            mock_get.return_value = mock_response

            result = fetch_ollama_models("http://localhost:11434")

            assert len(result) == 2

    def test_skips_models_without_name(self):
        """Should skip models without name field."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {
                "models": [
                    {"name": "valid"},
                    {"other": "field"},  # No name
                    {"name": ""},  # Empty name
                ]
            }
            mock_get.return_value = mock_response

            result = fetch_ollama_models("http://localhost:11434")

            assert len(result) == 1
            assert result[0]["value"] == "valid"

    def test_uses_custom_timeout(self):
        """Should use custom timeout."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {"models": []}
            mock_get.return_value = mock_response

            fetch_ollama_models("http://localhost:11434", timeout=10.0)

            mock_get.assert_called_once()
            assert mock_get.call_args.kwargs["timeout"] == 10.0

    def test_uses_auth_headers(self):
        """Should pass auth headers to request."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {"models": []}
            mock_get.return_value = mock_response

            headers = {"Authorization": "Bearer token"}
            fetch_ollama_models("http://localhost:11434", auth_headers=headers)

            mock_get.assert_called_once()
            # safe_get wraps requests.get and injects a project User-Agent
            # alongside caller-provided headers, so we check the auth
            # header is present rather than doing an exact dict match.
            sent_headers = mock_get.call_args.kwargs["headers"]
            assert sent_headers["Authorization"] == "Bearer token"

    def test_constructs_correct_url(self):
        """Should construct correct API URL."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {"models": []}
            mock_get.return_value = mock_response

            fetch_ollama_models("http://localhost:11434")

            mock_get.assert_called_once()
            assert (
                mock_get.call_args.args[0] == "http://localhost:11434/api/tags"
            )

    def test_returns_correct_format(self):
        """Should return models in correct format with value and label."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {
                "models": [{"name": "test-model"}]
            }
            mock_get.return_value = mock_response

            result = fetch_ollama_models("http://localhost:11434")

            assert len(result) == 1
            assert "value" in result[0]
            assert "label" in result[0]
            assert result[0]["value"] == result[0]["label"]

    def test_handles_empty_models_list(self):
        """Should handle empty models list."""
        import requests

        with patch.object(requests, "get") as mock_get:
            mock_response = Mock()
            mock_response.status_code = 200
            mock_response.json.return_value = {"models": []}
            mock_get.return_value = mock_response

            result = fetch_ollama_models("http://localhost:11434")

            assert result == []