eac90941ef
- Implement SDK session with secretary model for tool approval flow - Add audit logging for tool usage and permission decisions - Support Feishu card interactions for approval requests - Add new commands for task interruption and progress checking - Remove old test files and update documentation
385 lines
15 KiB
Python
385 lines
15 KiB
Python
"""Tests for bot slash commands (replaces BDD feature tests).
|
|
|
|
Covers: //help, //new, //close, //switch, //status, //perm,
|
|
//direct, //smart, //shell, //remind, //tasks, //stop, //progress, //nodes
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from agent.manager import manager, Session
|
|
from orchestrator.agent import agent
|
|
from orchestrator.tools import set_current_user, set_current_chat
|
|
|
|
|
|
def _setup_user(user_id="user_abc123", chat_id=None):
|
|
set_current_user(user_id)
|
|
if chat_id:
|
|
set_current_chat(chat_id)
|
|
|
|
|
|
def _add_session(conv_id, cwd="/tmp/proj", user_id="user_abc123", activate=False):
|
|
session = Session(conv_id=conv_id, cwd=cwd, owner_id=user_id)
|
|
manager._sessions[conv_id] = session
|
|
if activate:
|
|
agent._active_conv[user_id] = conv_id
|
|
return session
|
|
|
|
|
|
# ── //help ──────────────────────────────────────────────────────────────────
|
|
|
|
class TestHelp:
|
|
@pytest.mark.asyncio
|
|
async def test_help_lists_commands(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//help")
|
|
for cmd in ("//new", "//status", "//close", "//switch", "//perm",
|
|
"//stop", "//progress", "//direct", "//smart", "//shell"):
|
|
assert cmd in reply, f"Missing {cmd} in help"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_h_alias(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//h")
|
|
assert "//new" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_command_returns_none(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//unknown_xyz")
|
|
assert reply is None
|
|
|
|
|
|
# ── //new ───────────────────────────────────────────────────────────────────
|
|
|
|
class TestNew:
|
|
@pytest.mark.asyncio
|
|
async def test_no_args_shows_usage(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//new")
|
|
assert "Usage" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_creates_session(self, tmp_working_dir):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//new myproject")
|
|
assert "myproject" in reply
|
|
assert len(manager.list_sessions(user_id="user_abc123")) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_traversal_blocked(self, tmp_working_dir):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//new ../../etc")
|
|
assert "Error" in reply
|
|
assert len(manager.list_sessions(user_id="user_abc123")) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_with_perm_flag(self, tmp_working_dir):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//new myproject --perm plan")
|
|
sessions = manager.list_sessions(user_id="user_abc123")
|
|
assert len(sessions) == 1
|
|
assert sessions[0]["permission_mode"] == "plan"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sends_card_when_chat_set(self, tmp_working_dir, feishu_calls):
|
|
from bot.commands import handle_command
|
|
_setup_user(chat_id="chat1")
|
|
reply = await handle_command("user_abc123", "//new myproject")
|
|
assert reply == "" # card was sent instead
|
|
assert len(feishu_calls["cards"]) >= 1
|
|
|
|
|
|
# ── //close ─────────────────────────────────────────────────────────────────
|
|
|
|
class TestClose:
|
|
@pytest.mark.asyncio
|
|
async def test_no_sessions(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//close")
|
|
assert "No sessions" in reply or "No active" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_active(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//close")
|
|
assert "Closed" in reply
|
|
assert len(manager.list_sessions(user_id="user_abc123")) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_by_number(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
_add_session("s2")
|
|
reply = await handle_command("user_abc123", "//close 1")
|
|
assert "Closed" in reply
|
|
assert len(manager.list_sessions(user_id="user_abc123")) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_number(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
reply = await handle_command("user_abc123", "//close 9")
|
|
assert "Invalid" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cannot_close_other_user(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", user_id="other_user")
|
|
reply = await handle_command("user_abc123", "//close s1")
|
|
assert "another user" in reply or "not found" in reply.lower()
|
|
|
|
|
|
# ── //switch ────────────────────────────────────────────────────────────────
|
|
|
|
class TestSwitch:
|
|
@pytest.mark.asyncio
|
|
async def test_no_sessions(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//switch 1")
|
|
assert "No sessions" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_switch(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
_add_session("s2")
|
|
reply = await handle_command("user_abc123", "//switch 2")
|
|
assert "Switched" in reply
|
|
assert agent._active_conv["user_abc123"] == "s2"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_out_of_range(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
reply = await handle_command("user_abc123", "//switch 5")
|
|
assert "Invalid" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_numeric(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
reply = await handle_command("user_abc123", "//switch abc")
|
|
assert "Invalid" in reply
|
|
|
|
|
|
# ── //status ────────────────────────────────────────────────────────────────
|
|
|
|
class TestStatus:
|
|
@pytest.mark.asyncio
|
|
async def test_no_sessions(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//status")
|
|
assert "No active sessions" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_sessions(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1")
|
|
_add_session("s2")
|
|
reply = await handle_command("user_abc123", "//status")
|
|
assert "s1" in reply
|
|
assert "s2" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_active_marker(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//status")
|
|
assert "→" in reply
|
|
|
|
|
|
# ── //perm ──────────────────────────────────────────────────────────────────
|
|
|
|
class TestPerm:
|
|
@pytest.mark.asyncio
|
|
async def test_no_args_shows_usage(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//perm")
|
|
assert "Usage" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_edit(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//perm edit")
|
|
assert "edit" in reply
|
|
assert manager._sessions["s1"].permission_mode == "acceptEdits"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_plan(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//perm plan")
|
|
assert "plan" in reply
|
|
assert manager._sessions["s1"].permission_mode == "plan"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_auto(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//perm auto")
|
|
assert "auto" in reply
|
|
assert manager._sessions["s1"].permission_mode == "dontAsk"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unknown_mode(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//perm xyz")
|
|
assert "Unknown" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_active_session(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//perm edit")
|
|
assert "No active session" in reply
|
|
|
|
|
|
# ── //direct + //smart ──────────────────────────────────────────────────────
|
|
|
|
class TestDirectSmart:
|
|
@pytest.mark.asyncio
|
|
async def test_direct_requires_session(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//direct")
|
|
assert "No active session" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_direct_enables_passthrough(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
reply = await handle_command("user_abc123", "//direct")
|
|
assert "Direct mode ON" in reply
|
|
assert agent._passthrough["user_abc123"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_smart_disables_passthrough(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
_add_session("s1", activate=True)
|
|
agent._passthrough["user_abc123"] = True
|
|
reply = await handle_command("user_abc123", "//smart")
|
|
assert "Smart mode ON" in reply
|
|
assert agent._passthrough["user_abc123"] is False
|
|
|
|
|
|
# ── //shell ─────────────────────────────────────────────────────────────────
|
|
|
|
class TestShell:
|
|
@pytest.mark.asyncio
|
|
async def test_no_args_shows_usage(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//shell")
|
|
assert "Usage" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_echo(self, tmp_working_dir):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//shell echo hello")
|
|
assert "hello" in reply
|
|
assert "exit code: 0" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_blocked_dangerous(self, tmp_working_dir):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//shell rm -rf /")
|
|
assert "Blocked" in reply
|
|
|
|
|
|
# ── //remind ────────────────────────────────────────────────────────────────
|
|
|
|
class TestRemind:
|
|
@pytest.mark.asyncio
|
|
async def test_no_args(self):
|
|
from bot.commands import handle_command
|
|
_setup_user(chat_id="chat1")
|
|
reply = await handle_command("user_abc123", "//remind")
|
|
assert "Usage" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_message(self):
|
|
from bot.commands import handle_command
|
|
_setup_user(chat_id="chat1")
|
|
reply = await handle_command("user_abc123", "//remind 10m")
|
|
assert "Usage" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_reminder(self):
|
|
from bot.commands import handle_command
|
|
from agent.scheduler import scheduler
|
|
_setup_user(chat_id="chat1")
|
|
reply = await handle_command("user_abc123", "//remind 30s check build")
|
|
assert "Reminder" in reply
|
|
assert len(scheduler._jobs) == 1
|
|
|
|
|
|
# ── //tasks ─────────────────────────────────────────────────────────────────
|
|
|
|
class TestTasks:
|
|
@pytest.mark.asyncio
|
|
async def test_no_tasks(self):
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//tasks")
|
|
assert "No background tasks" in reply
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_shows_running_task(self):
|
|
from bot.commands import handle_command
|
|
from agent.task_runner import task_runner, BackgroundTask, TaskStatus
|
|
_setup_user()
|
|
task_runner._tasks["t1"] = BackgroundTask(
|
|
task_id="t1", description="fix bug", started_at=time.time(),
|
|
status=TaskStatus.RUNNING,
|
|
)
|
|
reply = await handle_command("user_abc123", "//tasks")
|
|
assert "t1" in reply
|
|
assert "⏳" in reply
|
|
|
|
|
|
# ── //nodes ─────────────────────────────────────────────────────────────────
|
|
|
|
class TestNodes:
|
|
@pytest.mark.asyncio
|
|
async def test_nodes_outside_router_mode(self, monkeypatch):
|
|
import config
|
|
monkeypatch.setattr(config, "ROUTER_MODE", False)
|
|
from bot.commands import handle_command
|
|
_setup_user()
|
|
reply = await handle_command("user_abc123", "//nodes")
|
|
assert "Not in router mode" in reply
|