"""TerminalSession 和 ShellTool 单元测试

测试场景：
- 跨命令保持 cwd → cd 后执行 pwd 返回正确目录
- 跨命令保持 env → export 后执行 echo 返回正确值
- 危险命令需确认 → rm 命令触发确认回调
- 输出解析 → 错误输出结构化为错误类型+建议
- 无 session_id 时保持现有行为
- 会话管理器功能
"""

from __future__ import annotations

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

import pytest

from agentkit.tools.terminal_session import TerminalSession, TerminalSessionManager, CommandRecord
from agentkit.tools.shell import ShellTool
from agentkit.tools.output_parser import OutputParser, ParsedOutput, ErrorType


# ============================================================
# OutputParser 测试
# ============================================================


class TestOutputParser:
    """测试 OutputParser 结构化解析"""

    def setup_method(self):
        self.parser = OutputParser()

    def test_parse_success_output(self):
        """成功输出解析"""
        result = self.parser.parse("hello world", 0)
        assert result.exit_code == 0
        assert result.is_error is False
        assert result.error_type == ErrorType.NONE
        assert result.message == "hello world"
        assert result.suggestions == []

    def test_parse_empty_output(self):
        """空输出解析"""
        result = self.parser.parse("", 0)
        assert result.exit_code == 0
        assert result.is_error is False
        assert result.message == ""

    def test_parse_permission_denied(self):
        """权限不足错误解析"""
        result = self.parser.parse("permission denied: /root/secret", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.PERMISSION_DENIED
        assert len(result.suggestions) > 0
        assert any("sudo" in s for s in result.suggestions)

    def test_parse_not_found(self):
        """文件不存在错误解析"""
        result = self.parser.parse("No such file or directory: /tmp/missing", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.NOT_FOUND

    def test_parse_timeout(self):
        """超时错误解析"""
        result = self.parser.parse("Connection timed out", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.TIMEOUT

    def test_parse_syntax_error(self):
        """语法错误解析"""
        result = self.parser.parse("syntax error near unexpected token", 2)
        assert result.is_error is True
        assert result.error_type == ErrorType.SYNTAX_ERROR

    def test_parse_connection_refused(self):
        """连接被拒绝解析"""
        result = self.parser.parse("Connection refused on port 8080", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.CONNECTION_REFUSED

    def test_parse_out_of_memory(self):
        """内存不足解析"""
        result = self.parser.parse("Out of memory: cannot allocate", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.OUT_OF_MEMORY

    def test_parse_disk_full(self):
        """磁盘满解析"""
        result = self.parser.parse("No space left on device", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.DISK_FULL

    def test_parse_already_exists(self):
        """已存在解析"""
        result = self.parser.parse("File already exists: /tmp/test", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.ALREADY_EXISTS

    def test_parse_invalid_argument(self):
        """无效参数解析"""
        result = self.parser.parse("invalid argument: --unknown-flag", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.INVALID_ARGUMENT

    def test_parse_network_error(self):
        """网络错误解析"""
        result = self.parser.parse("Network is unreachable", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.NETWORK_ERROR

    def test_parse_exit_code_126(self):
        """退出码 126 → 权限不足"""
        result = self.parser.parse("some unknown error", 126)
        assert result.is_error is True
        assert result.error_type == ErrorType.PERMISSION_DENIED

    def test_parse_exit_code_127(self):
        """退出码 127 → 命令未找到"""
        result = self.parser.parse("some unknown error", 127)
        assert result.is_error is True
        assert result.error_type == ErrorType.NOT_FOUND

    def test_parse_exit_code_130(self):
        """退出码 130 → 被中断"""
        result = self.parser.parse("some unknown error", 130)
        assert result.is_error is True
        assert result.error_type == ErrorType.TIMEOUT

    def test_parse_unknown_error(self):
        """未知错误"""
        result = self.parser.parse("something went wrong", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.UNKNOWN

    def test_parse_long_message_truncated(self):
        """长消息截断"""
        long_output = "x" * 300
        result = self.parser.parse(long_output, 0)
        assert len(result.message) <= 203  # 200 + "..."

    def test_parsed_output_to_dict(self):
        """ParsedOutput.to_dict()"""
        result = self.parser.parse("permission denied", 1)
        d = result.to_dict()
        assert d["exit_code"] == 1
        assert d["is_error"] is True
        assert d["error_type"] == "permission_denied"
        assert isinstance(d["suggestions"], list)

    def test_parse_chinese_error_messages(self):
        """中文错误消息解析"""
        result = self.parser.parse("权限不足: 无法访问", 1)
        assert result.is_error is True
        assert result.error_type == ErrorType.PERMISSION_DENIED

    def test_parse_multiline_output_message_is_last_line(self):
        """多行输出取最后一行作为消息"""
        output = "line1\nline2\nline3"
        result = self.parser.parse(output, 0)
        assert result.message == "line3"


# ============================================================
# TerminalSession 测试
# ============================================================


class TestTerminalSession:
    """测试 TerminalSession 会话状态管理"""

    def test_construction_default(self):
        """默认构造"""
        session = TerminalSession(session_id="test")
        assert session.session_id == "test"
        assert session.cwd == os.getcwd()
        assert isinstance(session.env, dict)
        assert session.history == []

    def test_construction_custom_cwd(self):
        """自定义工作目录"""
        session = TerminalSession(session_id="test", cwd="/tmp")
        assert session.cwd == "/tmp"

    def test_construction_custom_env(self):
        """自定义环境变量"""
        session = TerminalSession(session_id="test", env={"FOO": "bar"})
        assert session.env.get("FOO") == "bar"

    def test_set_cwd(self):
        """手动设置 cwd"""
        session = TerminalSession(session_id="test")
        session.set_cwd("/usr/local")
        assert session.cwd == "/usr/local"

    def test_set_env(self):
        """手动设置环境变量"""
        session = TerminalSession(session_id="test")
        session.set_env("MY_VAR", "hello")
        assert session.env.get("MY_VAR") == "hello"

    def test_update_env(self):
        """批量更新环境变量"""
        session = TerminalSession(session_id="test")
        session.update_env({"A": "1", "B": "2"})
        assert session.env.get("A") == "1"
        assert session.env.get("B") == "2"

    def test_get_env_returns_copy(self):
        """get_env 返回副本，修改不影响原数据"""
        session = TerminalSession(session_id="test")
        env = session.get_env()
        env["HACKED"] = "yes"
        assert "HACKED" not in session.env

    def test_get_history_returns_copy(self):
        """get_history 返回副本"""
        session = TerminalSession(session_id="test")
        history = session.get_history()
        assert history is not session._history

    @pytest.mark.asyncio
    async def test_execute_simple_command(self):
        """执行简单命令"""
        session = TerminalSession(session_id="test")
        result = await session.execute("echo hello")
        assert result.exit_code == 0
        assert "hello" in result.raw_output

    @pytest.mark.asyncio
    async def test_execute_records_history(self):
        """执行命令记录历史"""
        session = TerminalSession(session_id="test")
        await session.execute("echo first")
        await session.execute("echo second")
        assert len(session.history) == 2
        assert session.history[0].command == "echo first"
        assert session.history[1].command == "echo second"

    @pytest.mark.asyncio
    async def test_cross_command_cwd(self):
        """跨命令保持 cwd：cd 后 pwd 返回正确目录"""
        session = TerminalSession(session_id="test")
        await session.execute("cd /tmp")
        assert session.cwd == "/tmp"

        result = await session.execute("pwd")
        assert "/tmp" in result.raw_output

    @pytest.mark.asyncio
    async def test_cross_command_env(self):
        """跨命令保持 env：export 后 echo 返回正确值"""
        session = TerminalSession(session_id="test")
        await session.execute("export MY_TEST_VAR=hello123")
        assert session.env.get("MY_TEST_VAR") == "hello123"

        result = await session.execute("echo $MY_TEST_VAR")
        assert "hello123" in result.raw_output

    @pytest.mark.asyncio
    async def test_cd_relative_path(self):
        """cd 相对路径（目录存在时更新 cwd）"""
        # 使用 /usr 作为基础目录，cd local（/usr/local 存在）
        session = TerminalSession(session_id="test", cwd="/usr")
        await session.execute("cd local")
        assert session.cwd == "/usr/local"

    @pytest.mark.asyncio
    async def test_cd_absolute_path(self):
        """cd 绝对路径"""
        session = TerminalSession(session_id="test")
        await session.execute("cd /usr")
        assert session.cwd == "/usr"

    @pytest.mark.asyncio
    async def test_failed_command_no_state_update(self):
        """失败命令不更新状态"""
        session = TerminalSession(session_id="test", cwd="/tmp")
        await session.execute("cd /nonexistent_dir_xyz")
        # cd 失败，cwd 不应更新
        assert session.cwd == "/tmp"

    @pytest.mark.asyncio
    async def test_timeout(self):
        """命令超时"""
        session = TerminalSession(session_id="test")
        result = await session.execute("sleep 10", timeout=0.5)
        assert result.exit_code == -1
        assert result.is_error is True

    @pytest.mark.asyncio
    async def test_max_history(self):
        """历史记录上限"""
        session = TerminalSession(session_id="test", max_history=3)
        for i in range(5):
            await session.execute(f"echo {i}")
        assert len(session.history) == 3
        assert session.history[0].command == "echo 2"

    def test_close(self):
        """关闭会话"""
        session = TerminalSession(session_id="test")
        session.close()  # 不应抛出异常


# ============================================================
# TerminalSessionManager 测试
# ============================================================


class TestTerminalSessionManager:
    """测试 TerminalSessionManager 会话管理"""

    def test_get_or_create_new(self):
        """创建新会话"""
        manager = TerminalSessionManager()
        session = manager.get_or_create("s1")
        assert session.session_id == "s1"

    def test_get_or_create_existing(self):
        """获取已有会话"""
        manager = TerminalSessionManager()
        s1 = manager.get_or_create("s1")
        s1.set_cwd("/tmp")
        s2 = manager.get_or_create("s1")
        assert s2.cwd == "/tmp"

    def test_get_existing(self):
        """get 获取已有会话"""
        manager = TerminalSessionManager()
        manager.get_or_create("s1")
        session = manager.get("s1")
        assert session is not None

    def test_get_nonexistent(self):
        """get 不存在的会话返回 None"""
        manager = TerminalSessionManager()
        assert manager.get("nonexistent") is None

    def test_remove(self):
        """移除会话"""
        manager = TerminalSessionManager()
        manager.get_or_create("s1")
        manager.remove("s1")
        assert manager.get("s1") is None

    def test_list_sessions(self):
        """列出会话"""
        manager = TerminalSessionManager()
        manager.get_or_create("s1")
        manager.get_or_create("s2")
        assert sorted(manager.list_sessions()) == ["s1", "s2"]

    def test_has_session(self):
        """检查会话是否存在"""
        manager = TerminalSessionManager()
        manager.get_or_create("s1")
        assert manager.has_session("s1") is True
        assert manager.has_session("s2") is False

    def test_max_sessions_eviction(self):
        """超过最大会话数时移除最旧会话"""
        manager = TerminalSessionManager(max_sessions=2)
        manager.get_or_create("s1")
        manager.get_or_create("s2")
        manager.get_or_create("s3")  # 应该移除 s1
        assert not manager.has_session("s1")
        assert manager.has_session("s2")
        assert manager.has_session("s3")

    def test_close_all(self):
        """关闭所有会话"""
        manager = TerminalSessionManager()
        manager.get_or_create("s1")
        manager.get_or_create("s2")
        manager.close_all()
        assert manager.list_sessions() == []


# ============================================================
# ShellTool 测试
# ============================================================


class TestShellToolConstruction:
    """测试 ShellTool 构造"""

    def test_default_construction(self):
        tool = ShellTool()
        assert tool.name == "shell"
        assert tool.input_schema is not None
        assert "command" in tool.input_schema["properties"]
        assert "session_id" in tool.input_schema["properties"]
        assert tool.input_schema["required"] == ["command"]

    def test_custom_construction(self):
        tool = ShellTool(name="my_shell", version="2.0.0")
        assert tool.name == "my_shell"
        assert tool.version == "2.0.0"

    def test_to_dict(self):
        tool = ShellTool()
        d = tool.to_dict()
        assert d["name"] == "shell"
        assert "input_schema" in d

    def test_repr(self):
        tool = ShellTool()
        r = repr(tool)
        assert "ShellTool" in r
        assert "shell" in r


class TestShellToolExecution:
    """测试 ShellTool 命令执行"""

    @pytest.mark.asyncio
    async def test_execute_simple_command(self):
        """执行简单命令（无会话模式）"""
        tool = ShellTool()
        result = await tool.execute(command="echo hello")
        assert result["exit_code"] == 0
        assert "hello" in result["output"]
        assert result["is_error"] is False
        assert result["session_id"] is None

    @pytest.mark.asyncio
    async def test_execute_missing_command(self):
        """缺少 command 参数"""
        tool = ShellTool()
        result = await tool.execute()
        assert result["is_error"] is True
        assert result["exit_code"] == 1

    @pytest.mark.asyncio
    async def test_execute_with_working_dir(self):
        """指定工作目录"""
        tool = ShellTool()
        result = await tool.execute(command="pwd", working_dir="/tmp")
        assert result["exit_code"] == 0
        assert "/tmp" in result["output"]

    @pytest.mark.asyncio
    async def test_execute_with_session(self):
        """会话模式执行命令"""
        tool = ShellTool()
        result = await tool.execute(command="echo session_test", session_id="s1")
        assert result["exit_code"] == 0
        assert "session_test" in result["output"]
        assert result["session_id"] == "s1"

    @pytest.mark.asyncio
    async def test_session_preserves_cwd(self):
        """会话模式保持 cwd"""
        tool = ShellTool()
        await tool.execute(command="cd /tmp", session_id="cwd-test")
        result = await tool.execute(command="pwd", session_id="cwd-test")
        assert "/tmp" in result["output"]

    @pytest.mark.asyncio
    async def test_session_preserves_env(self):
        """会话模式保持 env"""
        tool = ShellTool()
        await tool.execute(
            command="export SHELL_TEST_VAR=world", session_id="env-test"
        )
        result = await tool.execute(
            command="echo $SHELL_TEST_VAR", session_id="env-test"
        )
        assert "world" in result["output"]

    @pytest.mark.asyncio
    async def test_no_session_id_backward_compatible(self):
        """无 session_id 时保持现有行为"""
        tool = ShellTool()
        result = await tool.execute(command="echo no_session")
        assert result["exit_code"] == 0
        assert "no_session" in result["output"]
        assert result["session_id"] is None

    @pytest.mark.asyncio
    async def test_different_sessions_independent(self):
        """不同会话互不影响"""
        tool = ShellTool()
        await tool.execute(command="cd /tmp", session_id="s1")
        await tool.execute(command="cd /usr", session_id="s2")

        r1 = await tool.execute(command="pwd", session_id="s1")
        r2 = await tool.execute(command="pwd", session_id="s2")

        assert "/tmp" in r1["output"]
        assert "/usr" in r2["output"]


class TestShellToolSecurity:
    """测试 ShellTool 安全控制"""

    @pytest.mark.asyncio
    async def test_safe_command_allowed(self):
        """安全命令直接执行"""
        tool = ShellTool()
        result = await tool.execute(command="ls /tmp")
        assert result["exit_code"] == 0

    @pytest.mark.asyncio
    async def test_dangerous_command_blocked_without_callback(self):
        """危险命令无确认回调时被拒绝"""
        tool = ShellTool()
        result = await tool.execute(command="rm -rf /tmp/test")
        assert result["is_error"] is True
        assert result["exit_code"] == 126

    @pytest.mark.asyncio
    async def test_dangerous_command_confirmed(self):
        """危险命令通过确认回调允许执行"""
        confirm = AsyncMock(return_value=True)
        tool = ShellTool(confirm_callback=confirm)
        result = await tool.execute(command="rm -rf /tmp/nonexistent_test_dir")
        assert confirm.called
        # 命令本身可能失败（目录不存在），但不应被安全机制拒绝
        assert result["exit_code"] != 126 or not result["is_error"]

    @pytest.mark.asyncio
    async def test_dangerous_command_rejected_by_callback(self):
        """确认回调拒绝危险命令"""
        confirm = AsyncMock(return_value=False)
        tool = ShellTool(confirm_callback=confirm)
        result = await tool.execute(command="rm -rf /tmp/test")
        assert result["is_error"] is True
        assert result["exit_code"] == 126

    @pytest.mark.asyncio
    async def test_audit_log_recorded(self):
        """审计日志记录"""
        tool = ShellTool()
        await tool.execute(command="echo audit_test")
        assert len(tool.audit_log) > 0
        assert tool.audit_log[0]["command"] == "echo audit_test"

    @pytest.mark.asyncio
    async def test_blocked_command_in_audit_log(self):
        """被阻止的命令记录在审计日志"""
        tool = ShellTool()
        await tool.execute(command="rm -rf /tmp/test")
        blocked_entries = [e for e in tool.audit_log if e.get("blocked")]
        assert len(blocked_entries) > 0

    @pytest.mark.asyncio
    async def test_git_push_force_is_dangerous(self):
        """git push --force 是危险命令"""
        tool = ShellTool()
        result = await tool.execute(command="git push --force origin main")
        assert result["is_error"] is True
        assert result["exit_code"] == 126

    @pytest.mark.asyncio
    async def test_git_status_is_safe(self):
        """git status 是安全命令"""
        tool = ShellTool()
        result = await tool.execute(command="git status")
        # git status 可能在非 git 目录失败，但不应被安全机制拒绝
        assert result["exit_code"] != 126


class TestShellToolOutputParsing:
    """测试 ShellTool 输出解析集成"""

    @pytest.mark.asyncio
    async def test_error_output_structured(self):
        """错误输出结构化"""
        tool = ShellTool()
        result = await tool.execute(command="ls /nonexistent_dir_xyz_12345")
        assert result["is_error"] is True
        assert result["error_type"] in ("not_found", "unknown")
        assert isinstance(result["suggestions"], list)

    @pytest.mark.asyncio
    async def test_success_output_not_error(self):
        """成功输出不标记为错误"""
        tool = ShellTool()
        result = await tool.execute(command="echo success")
        assert result["is_error"] is False
        assert result["error_type"] == "none"


class TestShellToolSessionManager:
    """测试 ShellTool 会话管理器访问"""

    def test_session_manager_accessible(self):
        tool = ShellTool()
        assert tool.session_manager is not None

    @pytest.mark.asyncio
    async def test_session_created_on_first_use(self):
        """首次使用 session_id 时创建会话"""
        tool = ShellTool()
        assert not tool.session_manager.has_session("new-session")
        await tool.execute(command="echo test", session_id="new-session")
        assert tool.session_manager.has_session("new-session")
