"""Unit tests for the SDK migration (secretary model).

Tests cover:
- SDKSession lifecycle, message buffering, get_progress, approval
- sdk_hooks audit + deny
- SessionManager new methods (send_message, send_and_wait, get_progress, interrupt, approve)
- audit.py new functions (log_tool_use, log_permission_decision)
- bot/commands.py new commands (//stop, //progress, //perm auto)
- orchestrator/tools.py new tools (SessionProgressTool, InterruptConversationTool)
- bot/handler.py text approval fallback
"""

from __future__ import annotations

import asyncio
import json
import time
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch, PropertyMock

import pytest

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def tmp_audit_dir(tmp_path):
    """Redirect audit logs to a temp directory."""
    import agent.audit as audit_mod
    original = audit_mod.AUDIT_DIR
    audit_mod.AUDIT_DIR = tmp_path / "audit"
    yield tmp_path / "audit"
    audit_mod.AUDIT_DIR = original


@pytest.fixture
def mock_sdk_client():
    """Create a mock ClaudeSDKClient that yields controllable messages."""
    client = AsyncMock()
    client.connect = AsyncMock()
    client.disconnect = AsyncMock()
    client.query = AsyncMock()
    client.interrupt = AsyncMock()
    client.set_permission_mode = AsyncMock()
    return client


@pytest.fixture
def mock_feishu():
    """Mock all Feishu send functions."""
    captured = {"texts": [], "cards": [], "markdowns": []}

    async def _send_text(rid, rtype, text):
        captured["texts"].append(text)

    async def _send_card(rid, rtype, card):
        captured["cards"].append(card)

    async def _send_markdown(rid, rtype, content):
        captured["markdowns"].append(content)

    with patch("bot.feishu.send_text", side_effect=_send_text), \
         patch("bot.feishu.send_card", side_effect=_send_card), \
         patch("bot.feishu.send_markdown", side_effect=_send_markdown), \
         patch("bot.handler.send_text", side_effect=_send_text), \
         patch("bot.handler.send_markdown", side_effect=_send_markdown):
        yield captured


# ===========================================================================
# 1. SDKSession unit tests
# ===========================================================================


class TestSDKSessionProgress:
    """Test SDKSession.get_progress() with buffered messages."""

    def test_initial_progress_is_idle(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        p = s.get_progress()
        assert p.busy is False
        assert p.current_prompt == ""
        assert p.text_messages == []
        assert p.tool_calls == []

    def test_progress_after_manual_state_change(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        s._busy = True
        s._current_prompt = "write hello.py"
        s._started_at = time.time() - 5
        s._text_buffer = ["I'll create the file", "Done"]
        s._tool_buffer = ["Write(hello.py)", "Read(hello.py)"]
        p = s.get_progress()
        assert p.busy is True
        assert p.current_prompt == "write hello.py"
        assert p.elapsed_seconds >= 4
        assert len(p.text_messages) == 2
        assert len(p.tool_calls) == 2

    def test_buffer_limits(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        for i in range(30):
            s._text_buffer.append(f"text-{i}")
            if len(s._text_buffer) > s.MAX_BUFFER_TEXTS:
                s._text_buffer.pop(0)
        assert len(s._text_buffer) == s.MAX_BUFFER_TEXTS
        assert s._text_buffer[0] == "text-10"

    def test_progress_pending_approval(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        s._pending_approval_desc = "Bash: `rm -rf /tmp/test`"
        p = s.get_progress()
        assert p.pending_approval == "Bash: `rm -rf /tmp/test`"


class TestSDKSessionApproval:
    """Test the approval mechanism."""

    @pytest.mark.asyncio
    async def test_approve_resolves_future(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        loop = asyncio.get_running_loop()
        s._pending_approval = loop.create_future()
        await s.approve(True)
        assert s._pending_approval.result() is True

    @pytest.mark.asyncio
    async def test_approve_deny(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        loop = asyncio.get_running_loop()
        s._pending_approval = loop.create_future()
        await s.approve(False)
        assert s._pending_approval.result() is False

    @pytest.mark.asyncio
    async def test_approve_no_pending_is_noop(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        # Should not raise
        await s.approve(True)


class TestSDKSessionClose:
    """Test clean shutdown."""

    @pytest.mark.asyncio
    async def test_close_without_start(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        # Should not raise
        await s.close()
        assert s.client is None

    @pytest.mark.asyncio
    async def test_close_disconnects_client(self, mock_sdk_client):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        s.client = mock_sdk_client
        s._message_loop_task = None
        await s.close()
        mock_sdk_client.disconnect.assert_awaited_once()
        assert s.client is None


class TestSDKSessionSend:
    """Test send() and send_and_wait() with mocked client."""

    @pytest.mark.asyncio
    async def test_send_returns_immediately(self):
        from agent.sdk_session import SDKSession

        s = SDKSession("c1", "/tmp", "u1", chat_id="chat1")

        # Provide a mock client that has receive_messages yielding nothing
        mock_client = AsyncMock()
        mock_client.query = AsyncMock()

        async def _empty_messages():
            return
            yield  # make it an async generator

        mock_client.receive_messages = _empty_messages
        s.client = mock_client

        result = await s.send("hello")
        assert "已开始执行" in result
        assert s._busy is True
        assert s._current_prompt == "hello"

        # Cleanup
        if s._message_loop_task:
            s._message_loop_task.cancel()
            try:
                await s._message_loop_task
            except asyncio.CancelledError:
                pass


class TestSDKSessionFormatSummary:
    """Test _format_tool_summary."""

    def test_bash_summary(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        result = s._format_tool_summary("Bash", {"command": "ls -la"})
        assert "`ls -la`" in result

    def test_edit_summary(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        result = s._format_tool_summary("Edit", {"file_path": "/tmp/test.py"})
        assert "test.py" in result

    def test_other_summary_truncated(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        result = s._format_tool_summary("CustomTool", {"key": "x" * 500})
        assert len(result) <= 200


# ===========================================================================
# 2. sdk_hooks tests
# ===========================================================================


class TestSDKHooks:

    @pytest.mark.asyncio
    async def test_audit_hook_logs(self, tmp_audit_dir):
        from agent.sdk_hooks import audit_hook

        input_data = {
            "session_id": "test-session",
            "tool_name": "Bash",
            "tool_input": {"command": "echo hello"},
            "tool_response": "hello\n",
        }
        result = await audit_hook(input_data, "tu-1", {"signal": None})
        assert result == {}

        # Check JSONL was written
        log_file = tmp_audit_dir / "test-session.jsonl"
        assert log_file.exists()
        entry = json.loads(log_file.read_text().strip())
        assert entry["type"] == "tool_use"
        assert entry["tool_name"] == "Bash"

    @pytest.mark.asyncio
    async def test_deny_dangerous_rm_rf(self):
        from agent.sdk_hooks import deny_dangerous_hook

        input_data = {
            "tool_name": "Bash",
            "tool_input": {"command": "rm -rf /"},
        }
        result = await deny_dangerous_hook(input_data, None, {"signal": None})
        assert result.get("hookSpecificOutput", {}).get("permissionDecision") == "deny"

    @pytest.mark.asyncio
    async def test_deny_allows_safe_commands(self):
        from agent.sdk_hooks import deny_dangerous_hook

        input_data = {
            "tool_name": "Bash",
            "tool_input": {"command": "ls -la /tmp"},
        }
        result = await deny_dangerous_hook(input_data, None, {"signal": None})
        assert result == {}

    @pytest.mark.asyncio
    async def test_deny_ignores_non_bash(self):
        from agent.sdk_hooks import deny_dangerous_hook

        input_data = {
            "tool_name": "Edit",
            "tool_input": {"file_path": "/etc/passwd"},
        }
        result = await deny_dangerous_hook(input_data, None, {"signal": None})
        assert result == {}

    def test_build_hooks_returns_expected_structure(self):
        from agent.sdk_hooks import build_hooks

        hooks = build_hooks("test-conv")
        assert "PostToolUse" in hooks
        assert "PreToolUse" in hooks
        assert len(hooks["PostToolUse"]) == 1
        assert len(hooks["PreToolUse"]) == 1


# ===========================================================================
# 3. manager tests (new methods)
# ===========================================================================


class TestSessionManagerNew:

    @pytest.fixture(autouse=True)
    def _reset(self):
        from agent.manager import manager
        manager._sessions.clear()
        yield
        manager._sessions.clear()

    @pytest.mark.asyncio
    async def test_create_session_no_cc_timeout(self):
        from agent.manager import manager, Session
        s = await manager.create("c1", "/tmp/test", owner_id="u1", chat_id="chat1")
        assert s.conv_id == "c1"
        assert s.chat_id == "chat1"
        assert not hasattr(s, "cc_timeout") or "cc_timeout" not in s.to_dict()

    @pytest.mark.asyncio
    async def test_get_progress_no_session(self):
        from agent.manager import manager
        result = manager.get_progress("nonexistent")
        assert result is None

    @pytest.mark.asyncio
    async def test_get_progress_no_sdk_session(self):
        from agent.manager import manager
        await manager.create("c1", "/tmp/test", owner_id="u1")
        p = manager.get_progress("c1", user_id="u1")
        assert p is not None
        assert p.busy is False

    @pytest.mark.asyncio
    async def test_interrupt_no_sdk_session(self):
        from agent.manager import manager
        await manager.create("c1", "/tmp/test", owner_id="u1")
        result = await manager.interrupt("c1", user_id="u1")
        assert result is False

    @pytest.mark.asyncio
    async def test_approve_no_sdk_session(self):
        from agent.manager import manager
        await manager.create("c1", "/tmp/test", owner_id="u1")
        # Should not raise
        await manager.approve("c1", True)

    @pytest.mark.asyncio
    async def test_close_with_sdk_session(self):
        from agent.manager import manager
        from agent.sdk_session import SDKSession
        await manager.create("c1", "/tmp/test", owner_id="u1")
        mock_sdk = MagicMock(spec=SDKSession)
        mock_sdk.close = AsyncMock()
        manager._sessions["c1"].sdk_session = mock_sdk

        result = await manager.close("c1", user_id="u1")
        assert result is True
        mock_sdk.close.assert_awaited_once()

    @pytest.mark.asyncio
    async def test_set_permission_mode_with_sdk_session(self):
        from agent.manager import manager
        from agent.sdk_session import SDKSession
        await manager.create("c1", "/tmp/test", owner_id="u1")

        mock_sdk = MagicMock(spec=SDKSession)
        mock_sdk.set_permission_mode = AsyncMock()
        manager._sessions["c1"].sdk_session = mock_sdk

        manager.set_permission_mode("c1", "acceptEdits", user_id="u1")
        assert manager._sessions["c1"].permission_mode == "acceptEdits"

    def test_list_sessions_includes_busy(self):
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        mock_sdk = MagicMock(spec=SDKSession)
        mock_sdk._busy = True
        session.sdk_session = mock_sdk
        manager._sessions["c1"] = session
        result = manager.list_sessions()
        assert result[0]["busy"] is True

    def test_session_from_dict_strips_old_fields(self):
        from agent.manager import Session
        old_data = {
            "conv_id": "c1",
            "cwd": "/tmp",
            "owner_id": "u1",
            "cc_session_id": "old-uuid",
            "started": True,
            "cc_timeout": 300.0,
            "last_activity": 0.0,
            "idle_timeout": 1800,
            "permission_mode": "default",
        }
        s = Session.from_dict(old_data)
        assert s.conv_id == "c1"
        assert not hasattr(s, "cc_session_id")

    def test_session_to_dict_excludes_sdk_session(self):
        from agent.manager import Session
        s = Session(conv_id="c1", cwd="/tmp")
        s.sdk_session = MagicMock()
        d = s.to_dict()
        assert "sdk_session" not in d


# ===========================================================================
# 4. audit tests (new functions)
# ===========================================================================


class TestAuditNewFunctions:

    def test_log_tool_use(self, tmp_audit_dir):
        from agent.audit import log_tool_use

        log_tool_use(
            session_id="s1",
            tool_name="Bash",
            tool_input={"command": "echo hello"},
            tool_response="hello\n",
        )

        log_file = tmp_audit_dir / "s1.jsonl"
        assert log_file.exists()
        entry = json.loads(log_file.read_text().strip())
        assert entry["type"] == "tool_use"
        assert entry["tool_name"] == "Bash"

    def test_log_permission_decision_approved(self, tmp_audit_dir):
        from agent.audit import log_permission_decision

        log_permission_decision(
            conv_id="c1",
            tool_name="Bash",
            tool_input={"command": "rm test.txt"},
            approved=True,
        )

        log_file = tmp_audit_dir / "c1.jsonl"
        assert log_file.exists()
        entry = json.loads(log_file.read_text().strip())
        assert entry["type"] == "permission_decision"
        assert entry["approved"] is True

    def test_log_permission_decision_denied(self, tmp_audit_dir):
        from agent.audit import log_permission_decision

        log_permission_decision(
            conv_id="c1",
            tool_name="Write",
            tool_input={"file_path": "/etc/passwd"},
            approved=False,
        )

        log_file = tmp_audit_dir / "c1.jsonl"
        entry = json.loads(log_file.read_text().strip())
        assert entry["approved"] is False


# ===========================================================================
# 5. bot/commands.py new commands
# ===========================================================================


class TestNewCommands:

    @pytest.fixture(autouse=True)
    def _reset(self):
        from agent.manager import manager
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user, set_current_chat
        manager._sessions.clear()
        agent._active_conv.clear()
        agent._passthrough.clear()
        set_current_user(None)
        set_current_chat(None)
        yield
        manager._sessions.clear()
        agent._active_conv.clear()

    @pytest.mark.asyncio
    async def test_stop_no_active_session(self):
        from bot.commands import handle_command
        result = await handle_command("u1", "//stop")
        assert "No active session" in result

    @pytest.mark.asyncio
    async def test_stop_with_session_no_sdk(self):
        from bot.commands import handle_command
        from agent.manager import manager, Session
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user

        set_current_user("u1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        result = await handle_command("u1", "//stop")
        assert "No active task" in result

    @pytest.mark.asyncio
    async def test_progress_no_session(self):
        from bot.commands import handle_command
        result = await handle_command("u1", "//progress")
        assert "No active session" in result

    @pytest.mark.asyncio
    async def test_progress_idle_session(self):
        from bot.commands import handle_command
        from agent.manager import manager, Session
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user

        set_current_user("u1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        result = await handle_command("u1", "//progress")
        assert "空闲" in result

    @pytest.mark.asyncio
    async def test_progress_busy_session(self):
        from bot.commands import handle_command
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user

        set_current_user("u1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        sdk = SDKSession("c1", "/tmp", "u1")
        sdk._busy = True
        sdk._started_at = time.time() - 10
        sdk._tool_buffer = ["Bash(echo hello)", "Read(test.py)"]
        session.sdk_session = sdk
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        result = await handle_command("u1", "//progress")
        assert "执行中" in result
        assert "Bash" in result

    @pytest.mark.asyncio
    async def test_progress_with_pending_approval(self):
        from bot.commands import handle_command
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user

        set_current_user("u1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        sdk = SDKSession("c1", "/tmp", "u1")
        sdk._busy = True
        sdk._started_at = time.time() - 5
        sdk._pending_approval_desc = "Bash: `rm test`"
        session.sdk_session = sdk
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        result = await handle_command("u1", "//progress")
        assert "审批" in result

    @pytest.mark.asyncio
    async def test_perm_auto_alias(self):
        from bot.commands import handle_command
        from agent.manager import manager, Session
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_user

        set_current_user("u1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        result = await handle_command("u1", "//perm auto")
        assert "auto" in result
        assert manager._sessions["c1"].permission_mode == "dontAsk"


# ===========================================================================
# 6. orchestrator/tools.py new tools
# ===========================================================================


class TestNewTools:

    @pytest.fixture(autouse=True)
    def _reset(self):
        from agent.manager import manager
        from orchestrator.tools import set_current_user, set_current_chat
        manager._sessions.clear()
        set_current_user("u1")
        set_current_chat("chat1")
        yield
        manager._sessions.clear()
        set_current_user(None)
        set_current_chat(None)

    @pytest.mark.asyncio
    async def test_session_progress_not_found(self):
        from orchestrator.tools import SessionProgressTool
        tool = SessionProgressTool()
        result = await tool._arun("nonexistent")
        data = json.loads(result)
        assert "error" in data

    @pytest.mark.asyncio
    async def test_session_progress_idle(self):
        from orchestrator.tools import SessionProgressTool
        from agent.manager import manager, Session
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session

        tool = SessionProgressTool()
        result = await tool._arun("c1")
        data = json.loads(result)
        assert data["busy"] is False

    @pytest.mark.asyncio
    async def test_session_progress_busy(self):
        from orchestrator.tools import SessionProgressTool
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession

        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        sdk = SDKSession("c1", "/tmp", "u1")
        sdk._busy = True
        sdk._started_at = time.time() - 30
        sdk._current_prompt = "fix the bug"
        sdk._tool_buffer = ["Read(main.py)", "Edit(main.py)"]
        session.sdk_session = sdk
        manager._sessions["c1"] = session

        tool = SessionProgressTool()
        result = await tool._arun("c1")
        data = json.loads(result)
        assert data["busy"] is True
        assert data["elapsed_seconds"] >= 29
        assert "Edit" in str(data["recent_tools"])

    @pytest.mark.asyncio
    async def test_interrupt_not_found(self):
        from orchestrator.tools import InterruptConversationTool
        tool = InterruptConversationTool()
        result = await tool._arun("nonexistent")
        assert "not found" in result

    @pytest.mark.asyncio
    async def test_interrupt_no_active_task(self):
        from orchestrator.tools import InterruptConversationTool
        from agent.manager import manager, Session
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session

        tool = InterruptConversationTool()
        result = await tool._arun("c1")
        assert "No active task" in result


# ===========================================================================
# 7. bot/handler.py text approval fallback
# ===========================================================================


class TestTextApprovalFallback:

    @pytest.fixture(autouse=True)
    def _reset(self):
        from agent.manager import manager
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_chat
        manager._sessions.clear()
        agent._active_conv.clear()
        set_current_chat(None)
        yield
        manager._sessions.clear()
        agent._active_conv.clear()

    @pytest.mark.asyncio
    async def test_y_resolves_pending_approval(self, mock_feishu):
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_chat

        set_current_chat("chat1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        sdk = SDKSession("c1", "/tmp", "u1", chat_id="chat1")
        loop = asyncio.get_running_loop()
        sdk._pending_approval = loop.create_future()
        session.sdk_session = sdk
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        from bot.handler import _process_message
        await _process_message("u1", "chat1", "y")

        assert sdk._pending_approval.done()
        assert sdk._pending_approval.result() is True
        # handler calls send_text which is mocked separately from send_markdown
        assert any("批准" in t for t in mock_feishu["texts"])

    @pytest.mark.asyncio
    async def test_n_denies_pending_approval(self, mock_feishu):
        from agent.manager import manager, Session
        from agent.sdk_session import SDKSession
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_chat

        set_current_chat("chat1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        sdk = SDKSession("c1", "/tmp", "u1", chat_id="chat1")
        loop = asyncio.get_running_loop()
        sdk._pending_approval = loop.create_future()
        session.sdk_session = sdk
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        from bot.handler import _process_message
        await _process_message("u1", "chat1", "n")

        assert sdk._pending_approval.done()
        assert sdk._pending_approval.result() is False
        assert any("拒绝" in t for t in mock_feishu["texts"])

    @pytest.mark.asyncio
    async def test_y_without_pending_falls_through(self, mock_feishu):
        """If there's no pending approval, 'y' should not be consumed."""
        from agent.manager import manager, Session
        from orchestrator.agent import agent
        from orchestrator.tools import set_current_chat

        set_current_chat("chat1")
        session = Session(conv_id="c1", cwd="/tmp", owner_id="u1")
        manager._sessions["c1"] = session
        agent._active_conv["u1"] = "c1"

        from bot.handler import _process_message
        # Patch agent.run to avoid actual LLM call
        with patch("orchestrator.agent.agent.run", new_callable=AsyncMock, return_value="ok"):
            await _process_message("u1", "chat1", "y")

        # Should not have consumed as approval
        assert not any("批准" in t for t in mock_feishu["markdowns"])


# ===========================================================================
# 8. bot/feishu.py build_approval_card
# ===========================================================================


class TestBuildApprovalCard:

    def test_card_structure(self):
        from bot.feishu import build_approval_card
        card = build_approval_card("c1", "Bash", "`echo hello`", timeout=60)

        assert card["schema"] == "2.0"
        assert "权限审批" in card["header"]["title"]["content"]

        body_elements = card["body"]["elements"]
        tags = [e["tag"] for e in body_elements]
        # JSON 2.0: no "action" wrapper, no "note" — buttons are direct elements
        assert "action" not in tags
        assert "note" not in tags
        assert "markdown" in tags
        assert tags.count("button") == 2

        # Buttons carry approve/deny values
        buttons = [e for e in body_elements if e["tag"] == "button"]
        assert buttons[0]["value"]["action"] == "approve"
        assert buttons[0]["value"]["conv_id"] == "c1"
        assert buttons[1]["value"]["action"] == "deny"


# ===========================================================================
# 8b. bot/handler.py card callback response
# ===========================================================================


class TestCardCallbackResponse:
    """Test _handle_card_action returns proper P2CardActionTriggerResponse per
    docs/feishu/card_callback_communication.md."""

    def _make_trigger(self, action: str, conv_id: str, **extra) -> "P2CardActionTrigger":
        from lark_oapi.event.callback.model.p2_card_action_trigger import (
            CallBackAction, CallBackOperator, P2CardActionTrigger, P2CardActionTriggerData,
        )
        value = {"action": action, "conv_id": conv_id, **extra}
        act = CallBackAction()
        act.value = value
        act.tag = "button"
        op = CallBackOperator()
        op.open_id = "ou_test_user"
        data = P2CardActionTriggerData()
        data.action = act
        data.operator = op
        trigger = P2CardActionTrigger()
        trigger.event = data
        return trigger

    def test_approve_returns_toast_and_card(self):
        from bot.handler import _handle_card_action
        trigger = self._make_trigger("approve", "c1")
        with patch("bot.handler._main_loop", new=MagicMock()):
            resp = _handle_card_action(trigger)
        assert resp.toast is not None
        assert resp.toast.type == "success"
        assert "批准" in resp.toast.content
        assert resp.card is not None
        assert resp.card.type == "raw"
        assert resp.card.data["header"]["template"] == "green"

    def test_deny_returns_warning_toast(self):
        from bot.handler import _handle_card_action
        trigger = self._make_trigger("deny", "c1")
        with patch("bot.handler._main_loop", new=MagicMock()):
            resp = _handle_card_action(trigger)
        assert resp.toast.type == "warning"
        assert "拒绝" in resp.toast.content
        assert resp.card.data["header"]["template"] == "red"

    def test_missing_value_returns_empty_response(self):
        from bot.handler import _handle_card_action
        from lark_oapi.event.callback.model.p2_card_action_trigger import (
            CallBackAction, P2CardActionTrigger, P2CardActionTriggerData,
        )
        act = CallBackAction()
        act.value = {}  # no action/conv_id
        data = P2CardActionTriggerData()
        data.action = act
        trigger = P2CardActionTrigger()
        trigger.event = data
        resp = _handle_card_action(trigger)
        assert resp.toast is None
        assert resp.card is None

    def test_answer_question_returns_success_toast(self):
        from bot.handler import _handle_card_action
        trigger = self._make_trigger(
            "answer_question", "c1",
            question="Which lang?", answer="Python"
        )
        with patch("bot.handler._main_loop", new=MagicMock()):
            resp = _handle_card_action(trigger)
        assert resp.toast.type == "success"
        assert "Python" in resp.toast.content
        assert "Python" in resp.card.data["body"]["elements"][0]["content"]


# ===========================================================================
# 8c. AskUserQuestion flow
# ===========================================================================


class TestAskUserQuestion:

    def test_build_question_card(self):
        from bot.feishu import build_question_card
        questions = [
            {
                "question": "Which language?",
                "header": "Language",
                "options": [
                    {"label": "Python", "description": "Great for AI"},
                    {"label": "TypeScript", "description": "Great for web"},
                ],
                "multiSelect": False,
            }
        ]
        card = build_question_card("c1", questions)
        assert card["schema"] == "2.0"
        assert "提问" in card["header"]["title"]["content"]

        elements = card["body"]["elements"]
        buttons = [e for e in elements if e["tag"] == "button"]
        assert len(buttons) == 2
        assert buttons[0]["value"]["action"] == "answer_question"
        assert buttons[0]["value"]["question"] == "Which language?"
        assert buttons[0]["value"]["answer"] == "Python"
        assert buttons[1]["value"]["answer"] == "TypeScript"

    @pytest.mark.asyncio
    async def test_answer_question_resolves_future(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        loop = asyncio.get_running_loop()
        s._pending_question = loop.create_future()
        await s.answer_question({"Which language?": "Python"})
        assert s._pending_question.done()
        assert s._pending_question.result() == {"Which language?": "Python"}

    def test_card_callback_answer_question(self):
        from bot.handler import _handle_card_action
        from lark_oapi.event.callback.model.p2_card_action_trigger import (
            CallBackAction, CallBackOperator, P2CardActionTrigger, P2CardActionTriggerData,
        )
        act = CallBackAction()
        act.value = {"action": "answer_question", "conv_id": "c1",
                     "question": "Which lang?", "answer": "Python"}
        act.tag = "button"
        op = CallBackOperator()
        op.open_id = "ou_test"
        data = P2CardActionTriggerData()
        data.action = act
        data.operator = op
        trigger = P2CardActionTrigger()
        trigger.event = data

        with patch("bot.handler._main_loop", new=MagicMock()):
            resp = _handle_card_action(trigger)

        assert resp.toast.type == "success"
        assert "Python" in resp.toast.content
        assert "Python" in resp.card.data["body"]["elements"][0]["content"]

    def test_progress_shows_pending_question(self):
        from agent.sdk_session import SDKSession
        s = SDKSession("c1", "/tmp", "u1")
        s._busy = True
        s._started_at = time.time()
        s._pending_question_data = {
            "questions": [{"question": "Pick a color?", "options": [{"label": "Red"}, {"label": "Blue"}]}],
            "conv_id": "c1",
        }
        p = s.get_progress()
        assert p.pending_question is not None
        assert p.pending_question["questions"][0]["question"] == "Pick a color?"


# ===========================================================================
# 9. Permission mode constants
# ===========================================================================


class TestPermissionModes:

    def test_valid_modes_includes_dontask(self):
        from agent.sdk_session import VALID_PERMISSION_MODES
        assert "dontAsk" in VALID_PERMISSION_MODES

    def test_perm_aliases_has_auto(self):
        from bot.commands import _PERM_ALIASES
        assert _PERM_ALIASES["auto"] == "dontAsk"

    def test_perm_labels_has_dontask(self):
        from bot.commands import _PERM_LABELS
        assert _PERM_LABELS["dontAsk"] == "auto"


# ===========================================================================
# 10. Config
# ===========================================================================


class TestConfig:

    def test_sdk_approval_timeout_exists(self):
        from config import SDK_APPROVAL_TIMEOUT
        assert isinstance(SDK_APPROVAL_TIMEOUT, int)
        assert SDK_APPROVAL_TIMEOUT > 0