ss-tools/agent/tests/test_agent/test_langgraph_setup.py

# #region Test.AgentChat.LangGraph.Setup [C:3] [TYPE Module] [SEMANTICS test,agent,langgraph,setup]
# @BRIEF Tests for agent/langgraph_setup.py — configure_from_api, create_agent.
# @RELATION BINDS_TO -> [AgentChat.LangGraph.Setup]

from pathlib import Path
import sys

sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))

from unittest.mock import AsyncMock, MagicMock, patch
import pytest


@pytest.fixture
def anyio_backend():
    return "asyncio"


# #region Test.AgentChat.TestConfigureFromApi [C:2] [TYPE Function]
# @BRIEF Test configure_from_api updates global config.
class TestConfigureFromApi:
    def test_sets_llm_config(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None  # Reset
        ls.configure_from_api({"configured": True, "api_key": "sk-test", "default_model": "gpt-4o"})
        assert ls._llm_config is not None
        assert ls._llm_config["configured"] is True
        # Reset for other tests
        ls.configure_from_api(None)
        ls._llm_config = None

    def test_overwrites_previous_config(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None  # Reset
        ls.configure_from_api({"configured": True, "api_key": "sk-1"})
        ls.configure_from_api({"configured": False})
        assert ls._llm_config["configured"] is False
        ls._llm_config = None
# #endregion Test.AgentChat.TestConfigureFromApi


# #region Test.AgentChat.TestLlmDiagnostics [C:2] [TYPE Function] [SEMANTICS test,agent,llm,observability]
# @BRIEF Diagnostics identify the configured provider but never expose API credentials or full URL paths.
def test_llm_diagnostics_redacts_api_key_and_path():
    import ss_tools.agent.langgraph_setup as ls

    diagnostics = ls.llm_diagnostics({
        "configured": True,
        "provider_id": "provider-1",
        "provider_name": "litellm",
        "provider_type": "litellm",
        "base_url": "https://key:secret@lite.ai.rusal.com/v1/private",
        "api_key": "must-not-appear",
        "default_model": "qwen-flash",
        "selection_source": "assistant_planner_provider",
    })

    assert diagnostics["provider_host"] == "lite.ai.rusal.com"
    assert diagnostics["provider_scheme"] == "https"
    assert diagnostics["model"] == "qwen-flash"
    assert "api_key" not in diagnostics
    assert "base_url" not in diagnostics
    assert all("secret" not in str(value) for value in diagnostics.values())
# #endregion Test.AgentChat.TestLlmDiagnostics


# #region Test.AgentChat.TestCreateAgent [C:2] [TYPE Function]
# @BRIEF Test create_agent with various LLM config states.
class TestCreateAgent:
    @pytest.mark.anyio
    async def test_creates_agent_with_api_config(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None  # Reset
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-api-config",
            "base_url": "https://custom.api.com/v1",
            "default_model": "gpt-4o-mini",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI") as mock_llm, \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)):
            mock_create.return_value = MagicMock()
            result = await ls.create_agent([MagicMock()])
            assert result is mock_create.return_value
            call_kwargs = mock_llm.call_args[1]
            assert call_kwargs["api_key"] == "sk-api-config"
            assert call_kwargs["base_url"] == "https://custom.api.com/v1"
            assert call_kwargs["model"] == "gpt-4o-mini"
            assert "http_async_client" in call_kwargs
            assert "http_client" not in call_kwargs
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_raises_error_when_no_llm_configured(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None  # Reset
        with patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=None)):
            with pytest.raises(RuntimeError, match="No LLM provider configured in backend"):
                await ls.create_agent([])
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_creates_agent_with_partial_api_config(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-key-only",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI") as mock_llm, \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)):
            mock_create.return_value = MagicMock()
            result = await ls.create_agent([])
            assert result is mock_create.return_value
            call_kwargs = mock_llm.call_args[1]
            assert call_kwargs["api_key"] == "sk-key-only"
            assert call_kwargs["base_url"] is None
            assert call_kwargs["model"] is None
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_uses_inmemory_saver(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-test",
            "base_url": "",
            "default_model": "gpt-4o-mini",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI") as mock_llm, \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)):
            mock_create.return_value = MagicMock()
            await ls.create_agent([])
            call_kwargs = mock_create.call_args[1]
            from langgraph.checkpoint.memory import InMemorySaver
            assert isinstance(call_kwargs["checkpointer"], InMemorySaver)
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_uses_empty_interrupt_list_by_default(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-test",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI"), \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)):
            mock_create.return_value = MagicMock()
            await ls.create_agent([])
            assert mock_create.call_args[1]["interrupt_before"] == []
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_confirm_tools_env_interrupts_before_tools_node(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-test",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI"), \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)), \
             patch("ss_tools.agent.langgraph_setup.AGENT_CONFIRM_TOOLS", True):
            mock_create.return_value = MagicMock()
            await ls.create_agent([])
            assert mock_create.call_args[1]["interrupt_before"] == ["tools"]
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_uses_env_configured_interrupt_nodes(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-test",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI"), \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)), \
             patch("ss_tools.agent.langgraph_setup._INTERRUPT_BEFORE", "tools"):
            mock_create.return_value = MagicMock()
            await ls.create_agent([])
            assert mock_create.call_args[1]["interrupt_before"] == ["tools"]
        ls._llm_config = None

    @pytest.mark.anyio
    async def test_interrupt_override_bypasses_env_guardrail(self):
        import ss_tools.agent.langgraph_setup as ls
        ls._llm_config = None
        ls.configure_from_api({
            "configured": True,
            "api_key": "sk-test",
        })
        with patch("ss_tools.agent.langgraph_setup.ChatOpenAI"), \
             patch("ss_tools.agent.langgraph_setup.create_react_agent") as mock_create, \
             patch("ss_tools.agent.langgraph_setup._fetch_llm_config", new=AsyncMock(return_value=ls._llm_config)), \
             patch("ss_tools.agent.langgraph_setup.AGENT_CONFIRM_TOOLS", True):
            mock_create.return_value = MagicMock()
            await ls.create_agent([], interrupt_before=[])
            assert mock_create.call_args[1]["interrupt_before"] == []
        ls._llm_config = None
# #endregion Test.AgentChat.TestCreateAgent
# #endregion Test.AgentChat.LangGraph.Setup