fischer-agentkit/tests/unit/tools/test_pty_session.py

"""PTYSession 单元测试

测试场景：
- PTY 会话启动和关闭
- 交互式命令执行
- 自动应答提示
- 超时处理
- 自定义应答规则
"""

from __future__ import annotations

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

import pytest

from agentkit.tools.pty_session import PTYSession, PTYOutput
from agentkit.tools.shell import ShellTool


class TestPTYSessionConstruction:
    """测试 PTYSession 构造"""

    def test_default_construction(self):
        pty = PTYSession()
        assert pty.is_running is False
        assert pty._auto_respond is False
        assert pty._default_timeout == 30.0

    def test_custom_construction(self):
        pty = PTYSession(
            auto_respond=False,
            custom_rules=[(r"confirm\?", "yes")],
            default_timeout=60.0,
        )
        assert pty._auto_respond is False
        assert len(pty._respond_rules) > len(pty._respond_rules) - 1
        assert pty._default_timeout == 60.0


class TestPTYSessionLifecycle:
    """测试 PTYSession 生命周期"""

    @pytest.mark.asyncio
    async def test_start_and_close(self):
        """启动和关闭 PTY 会话"""
        pty = PTYSession()
        await pty.start()
        assert pty.is_running is True
        await pty.close()
        assert pty.is_running is False

    @pytest.mark.asyncio
    async def test_start_idempotent(self):
        """重复启动不报错"""
        pty = PTYSession()
        await pty.start()
        await pty.start()  # 不应抛出异常
        assert pty.is_running is True
        await pty.close()

    @pytest.mark.asyncio
    async def test_close_without_start(self):
        """未启动时关闭不报错"""
        pty = PTYSession()
        await pty.close()  # 不应抛出异常


class TestPTYSessionExecution:
    """测试 PTYSession 命令执行"""

    @pytest.mark.asyncio
    async def test_run_simple_command(self):
        """执行简单命令"""
        pty = PTYSession(default_timeout=10.0)
        try:
            await pty.start()
            result = await pty.run_command("echo hello_pty")
            assert "hello_pty" in result.output
            assert result.exit_code == 0
            assert result.timed_out is False
        finally:
            await pty.close()

    @pytest.mark.asyncio
    async def test_run_command_with_cwd(self):
        """指定工作目录执行命令"""
        pty = PTYSession(default_timeout=10.0)
        try:
            await pty.start()
            result = await pty.run_command("pwd", cwd="/tmp")
            assert "/tmp" in result.output
        finally:
            await pty.close()

    @pytest.mark.asyncio
    async def test_run_command_with_env(self):
        """指定环境变量执行命令"""
        pty = PTYSession(default_timeout=10.0)
        try:
            await pty.start()
            result = await pty.run_command(
                "echo $PTY_TEST_VAR",
                env={"PTY_TEST_VAR": "pty_value"},
            )
            assert "pty_value" in result.output
        finally:
            await pty.close()

    @pytest.mark.asyncio
    async def test_run_failing_command(self):
        """执行失败命令"""
        pty = PTYSession(default_timeout=10.0)
        try:
            await pty.start()
            result = await pty.run_command("ls /nonexistent_dir_xyz_12345")
            assert result.exit_code != 0
        finally:
            await pty.close()

    @pytest.mark.asyncio
    async def test_run_command_timeout(self):
        """命令超时"""
        pty = PTYSession(default_timeout=10.0)
        try:
            await pty.start()
            result = await pty.run_command("sleep 30", timeout=0.5)
            assert result.timed_out is True
            assert result.exit_code == -1
        finally:
            await pty.close()


class TestPTYSessionAutoRespond:
    """测试 PTYSession 自动应答"""

    @pytest.mark.asyncio
    async def test_auto_respond_yes_no(self):
        """自动应答 [y/N] 提示"""
        # 使用 echo 模拟包含提示的输出，然后验证自动应答规则存在
        pty = PTYSession(auto_respond=True)
        # 验证规则已加载
        rule_patterns = [r[0] for r in pty._respond_rules]
        assert any("y/N" in p or "Y/n" in p for p in rule_patterns)

    @pytest.mark.asyncio
    async def test_auto_respond_disabled(self):
        """禁用自动应答"""
        pty = PTYSession(auto_respond=False)
        assert pty._auto_respond is False

    @pytest.mark.asyncio
    async def test_custom_respond_rules(self):
        """自定义应答规则"""
        pty = PTYSession(
            auto_respond=True,
            custom_rules=[(r"continue\?\s*$", "yes")],
        )
        rule_patterns = [r[0] for r in pty._respond_rules]
        assert r"continue\?\s*$" in rule_patterns


class TestPTYSessionSendAndRead:
    """测试 PTYSession 发送和读取"""

    @pytest.mark.asyncio
    async def test_send_without_start(self):
        """未启动时发送不报错"""
        pty = PTYSession()
        await pty.send("test")  # 不应抛出异常

    @pytest.mark.asyncio
    async def test_read_output_without_start(self):
        """未启动时读取返回空"""
        pty = PTYSession()
        output = await pty.read_output()
        assert output == ""


class TestPTYOutput:
    """测试 PTYOutput 数据类"""

    def test_default_values(self):
        output = PTYOutput(output="test")
        assert output.output == "test"
        assert output.exit_code == -1
        assert output.timed_out is False

    def test_custom_values(self):
        output = PTYOutput(output="error", exit_code=1, timed_out=True)
        assert output.exit_code == 1
        assert output.timed_out is True


class TestShellToolInteractiveMode:
    """测试 ShellTool 交互式模式"""

    @pytest.mark.asyncio
    async def test_interactive_mode(self):
        """ShellTool interactive 模式执行命令"""
        tool = ShellTool()
        result = await tool.execute(command="echo interactive_test", interactive=True)
        assert result["exit_code"] == 0
        assert "interactive_test" in result["output"]

    @pytest.mark.asyncio
    async def test_interactive_mode_with_session(self):
        """ShellTool 会话模式 + 交互式"""
        tool = ShellTool()
        result = await tool.execute(
            command="echo session_interactive",
            session_id="int-session",
            interactive=True,
        )
        assert result["exit_code"] == 0
        assert "session_interactive" in result["output"]