tests/test_cli.py - main - githubmirror/sgr-agent-core

tests/test_cli.py
378 строк13 KB
Pavel Rykov
refactor(skills): address review — BaseSkill, drop MCP prompts, default .agent/skills
14 июл 2026, 19:48
14 июл 2026, 19:488a0a7e7
"""Tests for CLI command sgrsh."""

import asyncio
import sys
from unittest.mock import AsyncMock, Mock, patch

import pytest

from sgr_agent_core.cli.sgrsh import chat_loop, find_config_file, format_skills_listing, main, run_agent
from sgr_agent_core.models import AgentStatesEnum
from sgr_agent_core.skills import BaseSkill, SkillMetadata


class TestFormatSkillsListing:
    """Test the skills listing formatter used by --list-skills."""

    def test_empty(self):
        assert "No skills" in format_skills_listing([])

    def test_lists_name_and_description(self):
        skills = [BaseSkill(metadata=SkillMetadata(name="greet", description="Greets people."))]
        out = format_skills_listing(skills)
        assert "greet" in out
        assert "Greets people." in out

    def test_excludes_non_user_invocable(self):
        skills = [
            BaseSkill(metadata=SkillMetadata(name="shown", description="Shown.")),
            BaseSkill(metadata=SkillMetadata(name="hidden", description="Hidden.", user_invocable=False)),
        ]
        out = format_skills_listing(skills)
        assert "shown" in out
        assert "hidden" not in out


class TestFindConfigFile:
    """Test find_config_file function."""

    def test_find_config_file_explicit_path_exists(self, tmp_path):
        """Test finding config file with explicit path."""
        config_file = tmp_path / "test_config.yaml"
        config_file.write_text("test: config")

        result = find_config_file(str(config_file))
        assert result == config_file.resolve()

    def test_find_config_file_explicit_path_not_exists(self, tmp_path):
        """Test finding config file with explicit non-existent path raises."""
        config_file = tmp_path / "nonexistent.yaml"

        with pytest.raises(FileNotFoundError, match="Config file not found"):
            find_config_file(str(config_file))

    def test_find_config_file_current_directory(self, tmp_path, monkeypatch):
        """Test finding config.yaml in current directory."""
        monkeypatch.chdir(tmp_path)
        config_file = tmp_path / "config.yaml"
        config_file.write_text("test: config")

        result = find_config_file(None)
        assert result == config_file.resolve()

    def test_find_config_file_not_found(self, tmp_path, monkeypatch):
        """Test when config.yaml not found in current directory raises."""
        monkeypatch.chdir(tmp_path)

        with pytest.raises(FileNotFoundError, match="Config file not found"):
            find_config_file(None)


class TestRunAgent:
    """Test run_agent function."""

    @pytest.mark.asyncio
    async def test_run_agent_success(self):
        """Test successful agent execution."""
        mock_agent = Mock()
        mock_agent.execute = AsyncMock(return_value="Test result")
        mock_agent._context = Mock()
        mock_agent._context.state = AgentStatesEnum.COMPLETED
        mock_agent.log = []

        result = await run_agent(mock_agent)

        assert result == "Test result"
        mock_agent.execute.assert_called_once()

    @pytest.mark.asyncio
    async def test_run_agent_with_clarification(self, monkeypatch):
        """Test agent execution with clarification request."""
        mock_agent = Mock()
        mock_agent._context = Mock()
        mock_agent._context.state = AgentStatesEnum.WAITING_FOR_CLARIFICATION
        mock_agent._context.execution_result = "Question 1?\nQuestion 2?"
        mock_agent.log = []
        mock_agent.provide_clarification = AsyncMock()
        mock_agent.cancel = AsyncMock()

        # Track if clarification was provided
        clarification_provided = False

        # Mock execute to simulate waiting for clarification
        async def mock_execute():
            nonlocal clarification_provided
            # First call - waiting for clarification
            if mock_agent._context.state == AgentStatesEnum.WAITING_FOR_CLARIFICATION:
                await asyncio.sleep(0.1)  # Simulate waiting
            # After clarification provided, return result
            if clarification_provided:
                return "Final result"
            # Otherwise keep waiting
            await asyncio.sleep(0.1)
            return "Final result"

        mock_agent.execute = AsyncMock(side_effect=mock_execute)

        # Mock user input - return answer once (patch _read_user_input, not input)
        user_input_called = False

        def mock_read_user_input(prompt: str) -> str:
            nonlocal user_input_called, clarification_provided
            if not user_input_called:
                user_input_called = True
                asyncio.create_task(mock_agent.provide_clarification([{"role": "user", "content": "User answer"}]))
                mock_agent._context.state = AgentStatesEnum.COMPLETED
                clarification_provided = True
                return "User answer"
            return ""

        monkeypatch.setattr("sgr_agent_core.cli.sgrsh._read_user_input", mock_read_user_input)

        result = await run_agent(mock_agent)

        # Should eventually get result after clarification
        assert result == "Final result" or result is None  # Allow None if timing issues

    @pytest.mark.asyncio
    async def test_run_agent_cancel_on_empty_input(self, monkeypatch):
        """Test agent cancellation on empty clarification input."""
        mock_agent = Mock()
        mock_agent._context = Mock()
        mock_agent._context.state = AgentStatesEnum.WAITING_FOR_CLARIFICATION
        mock_agent._context.execution_result = "Question?"
        mock_agent.log = []
        mock_agent.provide_clarification = AsyncMock()
        mock_agent.cancel = AsyncMock()

        async def mock_execute():
            await asyncio.sleep(0.1)
            return "Result"

        mock_agent.execute = AsyncMock(side_effect=mock_execute)

        # Mock empty user input (patch _read_user_input, not input)
        monkeypatch.setattr("sgr_agent_core.cli.sgrsh._read_user_input", lambda _: "")

        result = await run_agent(mock_agent)

        assert result is None
        mock_agent.cancel.assert_called_once()

    @pytest.mark.asyncio
    async def test_run_agent_execution_error(self):
        """Test agent execution error handling."""
        mock_agent = Mock()
        mock_agent._context = Mock()
        mock_agent._context.state = AgentStatesEnum.COMPLETED
        mock_agent.log = []
        mock_agent.execute = AsyncMock(side_effect=Exception("Test error"))

        result = await run_agent(mock_agent)

        assert result is None

    @pytest.mark.asyncio
    async def test_run_agent_keyboard_interrupt_during_input(self, monkeypatch):
        """Test agent cancellation on KeyboardInterrupt during user input."""
        mock_agent = Mock()
        mock_agent._context = Mock()
        mock_agent._context.state = AgentStatesEnum.WAITING_FOR_CLARIFICATION
        mock_agent._context.execution_result = "Question?"
        mock_agent.log = []
        mock_agent.provide_clarification = AsyncMock()
        mock_agent.cancel = AsyncMock()

        async def mock_execute():
            await asyncio.sleep(0.1)
            return "Result"

        mock_agent.execute = AsyncMock(side_effect=mock_execute)

        # Mock _read_user_input to raise KeyboardInterrupt
        def mock_read_user_input_raise_interrupt(_: str) -> str:
            raise KeyboardInterrupt()

        monkeypatch.setattr("sgr_agent_core.cli.sgrsh._read_user_input", mock_read_user_input_raise_interrupt)

        result = await run_agent(mock_agent)

        assert result is None
        mock_agent.cancel.assert_called_once()


class TestChatLoopMultipleRequests:
    """Test chat_loop with multiple user requests (conversation history)."""

    @pytest.mark.asyncio
    async def test_chat_loop_multiple_requests_then_quit(self, monkeypatch):
        """Test that multiple requests are sent to the agent and history
        grows."""
        inputs = iter(["First request", "Second request", "quit"])

        def mock_read_user_input(prompt: str) -> str:
            return next(inputs)

        create_calls = []

        async def mock_create(agent_def, *, task_messages):
            create_calls.append(list(task_messages))
            mock_agent = Mock()
            # Return different result per call: first request -> first response, etc.
            if len(create_calls) == 1:
                mock_agent.execute = AsyncMock(return_value="First response")
            else:
                mock_agent.execute = AsyncMock(return_value="Second response")
            mock_agent._context = Mock()
            mock_agent._context.state = AgentStatesEnum.COMPLETED
            mock_agent.log = []
            return mock_agent

        mock_config = Mock()
        mock_config.agents = {"test_agent": Mock()}

        with (
            patch("sgr_agent_core.cli.sgrsh._read_user_input", side_effect=mock_read_user_input),
            patch("sgr_agent_core.cli.sgrsh.AgentFactory") as mock_factory,
        ):
            mock_factory.create = mock_create
            await chat_loop("test_agent", mock_config)

        assert len(create_calls) == 2
        assert create_calls[0] == [{"role": "user", "content": "First request"}]
        assert create_calls[1] == [
            {"role": "user", "content": "First request"},
            {"role": "assistant", "content": "First response"},
            {"role": "user", "content": "Second request"},
        ]


class TestMain:
    """Test main CLI function."""

    @pytest.mark.asyncio
    async def test_main_with_query(self, tmp_path, monkeypatch):
        """Test main function with query argument."""
        monkeypatch.chdir(tmp_path)
        config_file = tmp_path / "config.yaml"
        config_file.write_text(
            """
llm:
  api_key: "test-key"
  base_url: "https://api.test.com/v1"
  model: "test-model"

agents:
  test_agent:
    base_class: "sgr_agent_core.agents.sgr_agent.SGRAgent"
    tools:
      - "final_answer_tool"
"""
        )

        with (
            patch("sgr_agent_core.cli.sgrsh.GlobalConfig") as mock_config_class,
            patch("sgr_agent_core.cli.sgrsh.AgentFactory") as mock_factory,
        ):
            mock_config = Mock()
            mock_config.agents = {"test_agent": Mock()}
            mock_config_class.from_yaml.return_value = mock_config

            mock_agent = Mock()
            mock_agent.execute = AsyncMock(return_value="Test result")
            mock_agent._context = Mock()
            mock_agent._context.state = AgentStatesEnum.COMPLETED
            mock_agent.log = []
            mock_factory.create = AsyncMock(return_value=mock_agent)

            # Mock sys.argv
            original_argv = sys.argv
            sys.argv = ["sgrsh", "Test query"]

            try:
                await main()
            except SystemExit:
                pass
            finally:
                sys.argv = original_argv

            mock_factory.create.assert_called_once()

    @pytest.mark.asyncio
    async def test_main_no_config_file(self, tmp_path, monkeypatch, capsys):
        """Test main function when config file not found."""
        monkeypatch.chdir(tmp_path)

        original_argv = sys.argv
        sys.argv = ["sgrsh", "Test query"]

        try:
            await main()
        except SystemExit as e:
            assert e.code == 1
        finally:
            sys.argv = original_argv

        captured = capsys.readouterr()
        assert "Config file not found" in captured.out

    @pytest.mark.asyncio
    async def test_main_with_agent_option(self, tmp_path, monkeypatch):
        """Test main function with --agent option."""
        monkeypatch.chdir(tmp_path)
        config_file = tmp_path / "config.yaml"
        config_file.write_text(
            """
llm:
  api_key: "test-key"
  base_url: "https://api.test.com/v1"
  model: "test-model"

agents:
  agent1:
    base_class: "sgr_agent_core.agents.sgr_agent.SGRAgent"
    tools: []
  agent2:
    base_class: "sgr_agent_core.agents.sgr_agent.SGRAgent"
    tools: []
"""
        )

        with (
            patch("sgr_agent_core.cli.sgrsh.GlobalConfig") as mock_config_class,
            patch("sgr_agent_core.cli.sgrsh.AgentFactory") as mock_factory,
        ):
            mock_config = Mock()
            mock_config.agents = {
                "agent1": Mock(),
                "agent2": Mock(),
            }
            mock_config_class.from_yaml.return_value = mock_config

            mock_agent = Mock()
            mock_agent.execute = AsyncMock(return_value="Test result")
            mock_agent._context = Mock()
            mock_agent._context.state = AgentStatesEnum.COMPLETED
            mock_agent.log = []
            mock_factory.create = AsyncMock(return_value=mock_agent)

            original_argv = sys.argv
            sys.argv = ["sgrsh", "--agent", "agent2", "Test query"]

            try:
                await main()
            except SystemExit:
                pass
            finally:
                sys.argv = original_argv

                # Check that agent2 was used
                assert mock_factory.create.called
                # Check that correct agent was passed
                # AgentFactory.create is called with agent_def as first positional argument
                call_args = mock_factory.create.call_args
                # Check first positional argument (agent_def)
                if call_args.args and len(call_args.args) > 0:
                    assert call_args.args[0] == mock_config.agents["agent2"]
                elif call_args.kwargs:
                    assert call_args.kwargs.get("agent_def") == mock_config.agents["agent2"]