ss-tools/backend/tests/test_agent/test_run.py

# #region Test.AgentChat.Run [C:3] [TYPE Module] [SEMANTICS test,agent,run,entrypoint]
# @BRIEF Tests for agent/run.py — _find_free_port and _fetch_llm_config.
# @RELATION BINDS_TO -> [AgentChat.Run]

from pathlib import Path
import sys

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

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


# #region test_find_free_port [C:2] [TYPE Function]
# @BRIEF Test _find_free_port for port scanning behavior.
class TestFindFreePort:
    def test_returns_free_port(self):
        from src.agent.run import _find_free_port
        with patch("socket.socket") as mock_socket:
            mock_instance = MagicMock()
            mock_socket.return_value.__enter__.return_value = mock_instance
            result = _find_free_port(8000, 10)
            assert result == 8000
            mock_instance.bind.assert_called_once_with(("", 8000))

    def test_skips_busy_ports(self):
        from src.agent.run import _find_free_port
        with patch("socket.socket") as mock_socket:
            mock_instance = MagicMock()
            mock_socket.return_value.__enter__.return_value = mock_instance
            # Ports 8000-8002 busy, 8003 free
            mock_instance.bind.side_effect = [
                OSError("Address in use"),  # 8000
                OSError("Address in use"),  # 8001
                OSError("Address in use"),  # 8002
                None,  # 8003 — success
            ]
            result = _find_free_port(8000, 10)
            assert result == 8003
            assert mock_instance.bind.call_count == 4

    def test_raises_when_all_busy(self):
        from src.agent.run import _find_free_port
        with patch("socket.socket") as mock_socket:
            mock_instance = MagicMock()
            mock_socket.return_value.__enter__.return_value = mock_instance
            mock_instance.bind.side_effect = OSError("Address in use")
            with pytest.raises(OSError, match="No free port found"):
                _find_free_port(8000, 3)
            assert mock_instance.bind.call_count == 3
# #endregion test_find_free_port


# #region test_fetch_llm_config [C:2] [TYPE Function]
# @BRIEF Test _fetch_llm_config with retry and fallback behavior.
class TestFetchLlmConfig:
    def test_returns_config_on_success(self):
        from src.agent.run import _fetch_llm_config
        with patch("src.agent.run.httpx.get") as mock_get:
            mock_response = MagicMock()
            mock_response.json.return_value = {"configured": True, "provider_type": "openai", "default_model": "gpt-4o"}
            mock_get.return_value = mock_response
            result = _fetch_llm_config()
            assert result is not None
            assert result["configured"] is True

    def test_returns_none_when_not_configured(self):
        from src.agent.run import _fetch_llm_config
        with patch("src.agent.run.httpx.get") as mock_get:
            mock_response = MagicMock()
            mock_response.json.return_value = {"configured": False, "reason": "no provider"}
            mock_get.return_value = mock_response
            result = _fetch_llm_config()
            assert result is None

    def test_retries_on_failure(self):
        from src.agent.run import _fetch_llm_config
        import time as time_module
        with patch("src.agent.run.httpx.get") as mock_get, \
             patch.object(time_module, "sleep") as mock_sleep:
            mock_get.side_effect = Exception("Connection refused")
            result = _fetch_llm_config()
            assert result is None
            assert mock_get.call_count == 6

    def test_retries_then_returns_config(self):
        from src.agent.run import _fetch_llm_config
        import time as time_module
        with patch("src.agent.run.httpx.get") as mock_get, \
             patch.object(time_module, "sleep") as mock_sleep:
            mock_get.side_effect = [
                Exception("Timeout"),       # Attempt 1
                Exception("Timeout"),       # Attempt 2
                MagicMock(json=lambda: {"configured": True, "provider_type": "openai"}),  # Attempt 3
            ]
            result = _fetch_llm_config()
            assert result is not None
            assert result["configured"] is True

    def test_returns_none_after_max_retries_with_http_error(self):
        from src.agent.run import _fetch_llm_config
        import time as time_module
        with patch("src.agent.run.httpx.get") as mock_get, \
             patch.object(time_module, "sleep") as mock_sleep:
            mock_response = MagicMock()
            mock_response.raise_for_status.side_effect = Exception("HTTP 500")
            mock_get.return_value = mock_response
            result = _fetch_llm_config()
            assert result is None
            assert mock_get.call_count == 6

    def test_uses_service_token_header(self):
        from src.agent.run import _fetch_llm_config
        with patch("src.agent.run.httpx.get") as mock_get, \
             patch("src.agent.run.os.getenv", return_value="test-token"):
            mock_response = MagicMock()
            mock_response.json.return_value = {"configured": True}
            mock_get.return_value = mock_response
            result = _fetch_llm_config()
            assert result is not None
            # Verify Authorization header was sent
            call_kwargs = mock_get.call_args[1]
            assert call_kwargs["headers"].get("Authorization") == "Bearer test-token"


# #endregion test_fetch_llm_config


# #region test_main_block [C:2] [TYPE Function]
# @BRIEF Test if __name__ == '__main__' block — service JWT, LLM config, port fallback, OSError.
class TestMainBlock:
    """Test the if __name__ == '__main__' entry point block via importlib.util fresh module."""

    def _run_as_main(self, monkeypatch, env_overrides=None, llm_configured=False,
                     port_bind_sequence=None, port_always_fail=False):
        """Execute run.py as __main__ with given mocking configuration."""
        import importlib.util
        from pathlib import Path

        run_path = Path(__file__).parent.parent.parent / "src" / "agent" / "run.py"
        spec = importlib.util.spec_from_file_location("__main__", str(run_path))

        # Apply env overrides
        env_overrides = env_overrides or {}
        for k, v in env_overrides.items():
            monkeypatch.setenv(k, v)

        with patch('httpx.get') as mock_httpx_get, \
             patch('socket.socket') as mock_socket_cls, \
             patch('asyncio.run') as mock_asyncio_run, \
             patch('src.agent.app.create_chat_interface') as mock_create_ci, \
             patch('src.agent.context.set_service_jwt') as mock_set_jwt, \
             patch('src.agent.langgraph_setup.configure_from_api') as mock_configure, \
             patch('src.agent.langgraph_setup.init_checkpointer'):
            mock_asyncio_run.side_effect = lambda coro: coro.close() if hasattr(coro, "close") else None

            # httpx for _fetch_llm_config
            mock_resp = MagicMock()
            if llm_configured:
                mock_resp.json.return_value = {
                    "configured": True, "provider_type": "openai",
                    "default_model": "gpt-4o", "api_key": "sk-test",
                }
            else:
                mock_resp.json.return_value = {"configured": False}
            mock_httpx_get.return_value = mock_resp

            # socket for _find_free_port
            mock_sock = MagicMock()
            mock_sock.__enter__.return_value = mock_sock
            mock_socket_cls.return_value = mock_sock
            if port_always_fail:
                mock_sock.bind.side_effect = OSError("all ports busy")
            elif port_bind_sequence is not None:
                mock_sock.bind.side_effect = port_bind_sequence
            else:
                mock_sock.bind.side_effect = [None]  # first bind succeeds

            mock_demo = MagicMock()
            mock_create_ci.return_value = mock_demo

            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)

        return {
            'set_jwt': mock_set_jwt,
            'configure': mock_configure,
            'demo': mock_demo,
        }

    def test_main_block_basic(self, monkeypatch):
        """Main block with default env, no SERVICE_JWT, no LLM config."""
        # Ensure SERVICE_JWT is NOT set (some tests leak it via os.environ)
        monkeypatch.delenv("SERVICE_JWT", raising=False)
        result = self._run_as_main(monkeypatch,
                                   env_overrides={"GRADIO_SERVER_PORT": "27860"})
        result['set_jwt'].assert_not_called()
        result['configure'].assert_not_called()
        result['demo'].launch.assert_called_once()

    def test_main_block_with_service_jwt(self, monkeypatch):
        """Main block sets service JWT via ContextVar."""
        result = self._run_as_main(monkeypatch, env_overrides={
            "SERVICE_JWT": "test-service-token",
            "GRADIO_SERVER_PORT": "27861",
        })
        result['set_jwt'].assert_called_once_with("test-service-token")

    def test_main_block_with_llm_config(self, monkeypatch):
        """Main block calls configure_from_api when LLM config is active."""
        result = self._run_as_main(monkeypatch,
                                   env_overrides={"GRADIO_SERVER_PORT": "27862"},
                                   llm_configured=True)
        result['configure'].assert_called_once()

    def test_main_block_port_fallback(self, monkeypatch):
        """Port in use triggers fallback warning (logged but continues)."""
        # Ports 27863, 27864 busy → 27865 free
        with patch('src.agent.run.logger') as mock_logger:
            result = self._run_as_main(monkeypatch,
                                       env_overrides={
                                           "GRADIO_SERVER_PORT": "27863",
                                           "GRADIO_ALLOW_PORT_FALLBACK": "true",
                                       },
                                       port_bind_sequence=[OSError("in use"), OSError("in use"), None])
            result['demo'].launch.assert_called_once()

    def test_main_block_port_oserror(self, monkeypatch):
        """OSError during port finding raises in main block."""
        with patch('src.agent.run.logger') as mock_logger:
            with pytest.raises(OSError):
                self._run_as_main(monkeypatch,
                                  env_overrides={
                                      "GRADIO_SERVER_PORT": "27866",
                                      "GRADIO_ALLOW_PORT_FALLBACK": "true",
                                  },
                                  port_always_fail=True)
# #endregion test_main_block
# #endregion Test.AgentChat.Run