362 lines
12 KiB
Python
362 lines
12 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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# U6: 项目经理模式协同事件渲染测试
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPMCollaborationRendering:
|
||
"""U6: 项目经理模式协同事件渲染测试"""
|
||
|
||
def _capture_render(self, message: dict) -> str:
|
||
"""辅助：渲染 PM 事件并捕获输出。"""
|
||
from agentkit.cli.chat import _render_pm_collaboration_event
|
||
|
||
captured = io.StringIO()
|
||
console = Console(file=captured, width=120)
|
||
with patch(
|
||
"agentkit.cli.chat.rprint",
|
||
side_effect=lambda *a, **kw: console.print(*a, **kw),
|
||
):
|
||
_render_pm_collaboration_event(message)
|
||
return captured.getvalue()
|
||
|
||
def test_collaboration_contract_defined_renders_panel(self):
|
||
"""collaboration_contract_defined 事件渲染为 Panel"""
|
||
message = {
|
||
"type": "collaboration_contract_defined",
|
||
"contracts": [
|
||
{
|
||
"from_expert": "backend",
|
||
"to_expert": "frontend",
|
||
"content_description": "API 定义",
|
||
"status": "pending",
|
||
},
|
||
],
|
||
}
|
||
text = self._capture_render(message)
|
||
assert "协作契约" in text
|
||
assert "backend" in text
|
||
assert "frontend" in text
|
||
assert "API 定义" in text
|
||
|
||
def test_collaboration_contract_defined_empty_contracts(self):
|
||
"""collaboration_contract_defined 空契约列表不产生输出"""
|
||
message = {"type": "collaboration_contract_defined", "contracts": []}
|
||
text = self._capture_render(message)
|
||
assert text == ""
|
||
|
||
def test_collaboration_notice_renders_colored_text(self):
|
||
"""collaboration_notice 事件渲染为带颜色的文本"""
|
||
message = {
|
||
"type": "collaboration_notice",
|
||
"from_expert": "backend",
|
||
"to_expert": "frontend",
|
||
"content_description": "API 定义已就绪",
|
||
}
|
||
text = self._capture_render(message)
|
||
assert "backend" in text
|
||
assert "frontend" in text
|
||
assert "API 定义已就绪" in text
|
||
|
||
def test_review_result_passed_renders_green(self):
|
||
"""review_result (passed=True) 渲染为绿色"""
|
||
message = {
|
||
"type": "review_result",
|
||
"phase_name": "后端开发",
|
||
"passed": True,
|
||
"feedback": "",
|
||
"expert": "backend_engineer",
|
||
}
|
||
text = self._capture_render(message)
|
||
assert "验收通过" in text
|
||
assert "后端开发" in text
|
||
|
||
def test_review_result_failed_renders_red(self):
|
||
"""review_result (passed=False) 渲染为红色"""
|
||
message = {
|
||
"type": "review_result",
|
||
"phase_name": "后端开发",
|
||
"passed": False,
|
||
"feedback": "API 缺少错误处理",
|
||
"expert": "backend_engineer",
|
||
"rework_count": 1,
|
||
}
|
||
text = self._capture_render(message)
|
||
assert "验收未通过" in text
|
||
assert "API 缺少错误处理" in text
|
||
assert "返工次数" in text
|
||
|
||
def test_risk_flagged_renders_yellow_panel(self):
|
||
"""risk_flagged 事件渲染为黄色 Panel"""
|
||
message = {
|
||
"type": "risk_flagged",
|
||
"expert": "backend_engineer",
|
||
"risk_description": "数据库连接池可能不足",
|
||
"phase_name": "后端开发",
|
||
}
|
||
text = self._capture_render(message)
|
||
assert "风险标记" in text
|
||
assert "数据库连接池可能不足" in text
|
||
assert "backend_engineer" in text
|
||
|
||
def test_missing_data_graceful_degradation(self):
|
||
"""事件数据缺失时优雅降级"""
|
||
# collaboration_notice 缺少字段 → 回退到 "?"
|
||
text = self._capture_render({"type": "collaboration_notice"})
|
||
assert "?" in text
|
||
|
||
# review_result 缺少字段 → 仍渲染（默认 failed=红色）
|
||
text = self._capture_render({"type": "review_result"})
|
||
assert "验收" in text
|
||
|
||
# risk_flagged 缺少字段 → 仍渲染
|
||
text = self._capture_render({"type": "risk_flagged"})
|
||
assert "风险标记" in text
|
||
|
||
def test_unhandled_event_returns_false(self):
|
||
"""非 PM 事件返回 False"""
|
||
from agentkit.cli.chat import _render_pm_collaboration_event
|
||
|
||
result = _render_pm_collaboration_event({"type": "team_formed"})
|
||
assert result is False
|
||
|
||
def test_pm_event_returns_true_when_handled(self):
|
||
"""PM 事件返回 True"""
|
||
from agentkit.cli.chat import _render_pm_collaboration_event
|
||
|
||
for etype in (
|
||
"collaboration_contract_defined",
|
||
"collaboration_notice",
|
||
"review_result",
|
||
"risk_flagged",
|
||
):
|
||
result = _render_pm_collaboration_event({"type": etype})
|
||
assert result is True, f"{etype} should return True"
|
||
|
||
|
||
class TestPrintHelpPMMode:
|
||
"""_print_help 包含项目经理模式说明测试"""
|
||
|
||
def test_help_includes_pm_mode(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 "项目经理" in text
|
||
|
||
def test_help_includes_collaboration_events(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 "协作契约" in text
|
||
assert "验收结果" in text
|
||
assert "风险标记" in text
|