193 lines
6.4 KiB
Python
193 lines
6.4 KiB
Python
"""CLI 多 Agent 入口 + 辩论支持单元测试 (U6)"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from rich.console import Console
|
||
|
||
from agentkit.experts.router import ExpertTeamRouter
|
||
from agentkit.experts.team import ExpertTeam
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# @team 前缀路由测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestTeamPrefixRouting:
|
||
"""@team 前缀路由测试"""
|
||
|
||
def test_team_prefix_matched(self):
|
||
"""@team 前缀被 ExpertTeamRouter 识别"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team 开发用户登录功能")
|
||
assert result.matched is True
|
||
assert result.task_content == "开发用户登录功能"
|
||
|
||
def test_team_prefix_with_template(self):
|
||
"""@team:dev_team 模板被识别"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team:dev_team 开发API")
|
||
assert result.matched is True
|
||
assert result.task_content == "开发API"
|
||
|
||
def test_non_team_input_not_matched(self):
|
||
"""非 @team 输入不被匹配"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("你好")
|
||
assert result.matched is False
|
||
|
||
def test_team_prefix_alone_matched(self):
|
||
"""@team 单独出现也被匹配（task_content 回退为完整输入）"""
|
||
router = ExpertTeamRouter()
|
||
result = router.resolve("@team")
|
||
assert result.matched is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _print_help 文档测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPrintHelp:
|
||
"""_print_help 包含 @team 文档测试"""
|
||
|
||
def test_help_includes_team_docs(self):
|
||
"""帮助文本包含 @team 说明"""
|
||
from agentkit.cli.chat import _print_help
|
||
|
||
captured = io.StringIO()
|
||
console = Console(file=captured, width=120)
|
||
with patch(
|
||
"agentkit.cli.chat.rprint",
|
||
side_effect=lambda *a, **kw: console.print(*a, **kw),
|
||
):
|
||
_print_help()
|
||
text = captured.getvalue()
|
||
assert "@team" in text
|
||
assert "/debate" in text
|
||
assert "/stop" in text
|
||
assert "专家团" in text
|
||
|
||
def test_help_includes_intervention_section(self):
|
||
"""帮助文本包含干预说明"""
|
||
from agentkit.cli.chat import _print_help
|
||
|
||
captured = io.StringIO()
|
||
console = Console(file=captured, width=120)
|
||
with patch(
|
||
"agentkit.cli.chat.rprint",
|
||
side_effect=lambda *a, **kw: console.print(*a, **kw),
|
||
):
|
||
_print_help()
|
||
text = captured.getvalue()
|
||
assert "Interventions" in text or "干预" in text
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# _execute_team_cli 函数测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestExecuteTeamCli:
|
||
"""_execute_team_cli 函数测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_false_for_non_team_input(self):
|
||
"""非 @team 输入返回 False"""
|
||
from agentkit.cli.chat import _execute_team_cli
|
||
|
||
gateway = MagicMock()
|
||
pool = MagicMock()
|
||
registry = MagicMock()
|
||
|
||
result = await _execute_team_cli("你好", gateway, pool, registry)
|
||
assert result is False
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_true_for_team_without_task(self):
|
||
"""@team 无任务描述返回 True（已处理，提示用法）"""
|
||
from agentkit.cli.chat import _execute_team_cli
|
||
|
||
gateway = MagicMock()
|
||
pool = MagicMock()
|
||
registry = MagicMock()
|
||
|
||
with patch.object(ExpertTeamRouter, "resolve") as mock_resolve:
|
||
mock_result = MagicMock()
|
||
mock_result.matched = True
|
||
mock_result.task_content = ""
|
||
mock_resolve.return_value = mock_result
|
||
|
||
result = await _execute_team_cli("@team", gateway, pool, registry)
|
||
assert result is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_returns_true_when_experts_unresolvable(self):
|
||
"""@team 有任务但无法解析专家时返回 True（错误提示）"""
|
||
from agentkit.cli.chat import _execute_team_cli
|
||
|
||
gateway = MagicMock()
|
||
pool = MagicMock()
|
||
registry = MagicMock()
|
||
|
||
with (
|
||
patch.object(ExpertTeamRouter, "resolve") as mock_resolve,
|
||
patch.object(ExpertTeamRouter, "resolve_expert_configs") as mock_configs,
|
||
):
|
||
mock_result = MagicMock()
|
||
mock_result.matched = True
|
||
mock_result.task_content = "开发功能"
|
||
mock_result.specified_experts = ["nonexistent"]
|
||
mock_resolve.return_value = mock_result
|
||
mock_configs.return_value = []
|
||
|
||
result = await _execute_team_cli("@team:nonexistent 开发功能", gateway, pool, registry)
|
||
assert result is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 干预命令支持测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestInterventionSupport:
|
||
"""干预命令基础设施测试"""
|
||
|
||
def test_team_has_broadcast_user_message(self):
|
||
"""ExpertTeam 有 broadcast_user_message 方法（干预广播基础）"""
|
||
assert hasattr(ExpertTeam, "broadcast_user_message")
|
||
|
||
def test_help_lists_debate_command(self):
|
||
"""帮助文本列出 /debate 命令"""
|
||
from agentkit.cli.chat import _print_help
|
||
|
||
captured = io.StringIO()
|
||
console = Console(file=captured, width=120)
|
||
with patch(
|
||
"agentkit.cli.chat.rprint",
|
||
side_effect=lambda *a, **kw: console.print(*a, **kw),
|
||
):
|
||
_print_help()
|
||
text = captured.getvalue()
|
||
assert "/debate" in text
|
||
assert "辩论" in text
|
||
|
||
def test_help_lists_stop_command(self):
|
||
"""帮助文本列出 /stop 命令"""
|
||
from agentkit.cli.chat import _print_help
|
||
|
||
captured = io.StringIO()
|
||
console = Console(file=captured, width=120)
|
||
with patch(
|
||
"agentkit.cli.chat.rprint",
|
||
side_effect=lambda *a, **kw: console.print(*a, **kw),
|
||
):
|
||
_print_help()
|
||
text = captured.getvalue()
|
||
assert "/stop" in text
|
||
assert "终止" in text
|