645 lines
23 KiB
Python
645 lines
23 KiB
Python
"""ExpertTeam 容器单元测试 (hub-and-spoke 模式)"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from agentkit.core.config_driven import ConfigDrivenAgent
|
||
from agentkit.core.handoff_transport import InProcessHandoffTransport
|
||
from agentkit.core.shared_workspace import SharedWorkspace
|
||
from agentkit.experts.config import ExpertConfig, ExpertTemplate
|
||
from agentkit.experts.expert import Expert
|
||
from agentkit.experts.registry import ExpertTemplateRegistry
|
||
from agentkit.experts.team import ExpertTeam, TeamStatus
|
||
|
||
|
||
# ── 辅助函数 ──────────────────────────────────────────────
|
||
|
||
|
||
def _make_expert_config(
|
||
name: str = "test_expert",
|
||
agent_type: str = "expert",
|
||
persona: str = "测试专家",
|
||
thinking_style: str = "逻辑推理",
|
||
bound_skills: list[str] | None = None,
|
||
is_lead: bool = False,
|
||
**kwargs,
|
||
) -> ExpertConfig:
|
||
"""创建测试用 ExpertConfig 实例"""
|
||
return ExpertConfig(
|
||
name=name,
|
||
agent_type=agent_type,
|
||
persona=persona,
|
||
thinking_style=thinking_style,
|
||
bound_skills=bound_skills or ["skill_a"],
|
||
is_lead=is_lead,
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "测试"},
|
||
**kwargs,
|
||
)
|
||
|
||
|
||
def _make_mock_agent() -> MagicMock:
|
||
"""创建 mock ConfigDrivenAgent"""
|
||
agent = MagicMock(spec=ConfigDrivenAgent)
|
||
agent.name = "test_expert"
|
||
agent._prompt_template = None
|
||
return agent
|
||
|
||
|
||
def _make_mock_pool() -> AsyncMock:
|
||
"""创建 mock AgentPool"""
|
||
pool = AsyncMock()
|
||
pool.create_agent = AsyncMock(return_value=_make_mock_agent())
|
||
pool.remove_agent = AsyncMock()
|
||
return pool
|
||
|
||
|
||
def _make_mock_expert(
|
||
name: str = "test_expert",
|
||
is_lead: bool = False,
|
||
is_active: bool = True,
|
||
) -> MagicMock:
|
||
"""创建 mock Expert"""
|
||
config = _make_expert_config(name=name, is_lead=is_lead)
|
||
expert = MagicMock(spec=Expert)
|
||
expert.config = config
|
||
expert.is_active = is_active
|
||
expert.team_id = None
|
||
expert.get_capabilities_summary.return_value = {
|
||
"name": name,
|
||
"persona": config.persona,
|
||
"thinking_style": config.thinking_style,
|
||
"bound_skills": config.bound_skills,
|
||
"is_lead": is_lead,
|
||
}
|
||
expert.destroy = AsyncMock()
|
||
return expert
|
||
|
||
|
||
# ── ExpertTeam 创建测试 ───────────────────────────────────
|
||
|
||
|
||
class TestExpertTeamCreation:
|
||
"""ExpertTeam 初始化与默认值测试"""
|
||
|
||
def test_default_values(self):
|
||
"""默认值：自动生成 team_id，FORMING 状态"""
|
||
team = ExpertTeam()
|
||
|
||
assert team.team_id is not None
|
||
assert len(team.team_id) > 0
|
||
assert team.status == TeamStatus.FORMING
|
||
assert team.lead_expert is None
|
||
assert team.experts == []
|
||
assert team.active_experts == []
|
||
|
||
def test_custom_team_id(self):
|
||
"""自定义 team_id"""
|
||
team = ExpertTeam(team_id="my_team")
|
||
|
||
assert team.team_id == "my_team"
|
||
|
||
def test_custom_workspace(self):
|
||
"""自定义 SharedWorkspace"""
|
||
workspace = SharedWorkspace()
|
||
team = ExpertTeam(workspace=workspace)
|
||
|
||
assert team._workspace is workspace
|
||
|
||
def test_custom_pool(self):
|
||
"""自定义 AgentPool"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
assert team._pool is pool
|
||
|
||
def test_custom_template_registry(self):
|
||
"""自定义 ExpertTemplateRegistry"""
|
||
registry = ExpertTemplateRegistry()
|
||
team = ExpertTeam(template_registry=registry)
|
||
|
||
assert team._template_registry is registry
|
||
|
||
|
||
# ── ExpertTeam.create_team 测试 ────────────────────────────
|
||
|
||
|
||
class TestExpertTeamCreateTeam:
|
||
"""ExpertTeam.create_team 团队创建测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_team_with_lead_only(self):
|
||
"""仅创建 Lead Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
mock_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = mock_expert
|
||
|
||
await team.create_team(lead_config)
|
||
|
||
assert team._lead_expert_name == "lead"
|
||
assert team.lead_expert is mock_expert
|
||
assert team.status == TeamStatus.PLANNING
|
||
assert mock_expert.team_id == team.team_id
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_team_with_lead_and_members(self):
|
||
"""创建 Lead Expert 和成员 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1", is_lead=False)
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1", is_lead=False)
|
||
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
assert len(team.experts) == 2
|
||
assert team._lead_expert_name == "lead"
|
||
assert team.status == TeamStatus.PLANNING
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_team_without_pool_raises(self):
|
||
"""没有 AgentPool 时 create_team 抛出 RuntimeError"""
|
||
team = ExpertTeam(pool=None)
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
|
||
with pytest.raises(RuntimeError, match="AgentPool not configured"):
|
||
await team.create_team(lead_config)
|
||
|
||
|
||
# ── ExpertTeam.add_expert 测试 ─────────────────────────────
|
||
|
||
|
||
class TestExpertTeamAddExpert:
|
||
"""ExpertTeam.add_expert 动态添加 Expert 测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_expert_with_config(self):
|
||
"""通过 ExpertConfig 添加 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
# 先创建团队
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
# 添加新成员
|
||
new_config = _make_expert_config(name="new_member")
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
new_expert = _make_mock_expert(name="new_member")
|
||
mock_create.return_value = new_expert
|
||
|
||
result = await team.add_expert(new_config)
|
||
|
||
assert result is new_expert
|
||
assert "new_member" in team._experts
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_expert_with_template_name(self):
|
||
"""通过模板名称添加 Expert"""
|
||
pool = _make_mock_pool()
|
||
registry = ExpertTemplateRegistry()
|
||
template_config = _make_expert_config(name="analyst", persona="分析师")
|
||
registry.register(ExpertTemplate(name="analyst", config=template_config))
|
||
|
||
team = ExpertTeam(pool=pool, template_registry=registry)
|
||
|
||
# 先创建团队
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
# 通过模板名称添加
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
analyst_expert = _make_mock_expert(name="analyst")
|
||
mock_create.return_value = analyst_expert
|
||
|
||
result = await team.add_expert("analyst")
|
||
|
||
assert result is analyst_expert
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_expert_with_nonexistent_template_raises(self):
|
||
"""使用不存在的模板名称添加 Expert 抛出 ValueError"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
# 先创建团队
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
with pytest.raises(ValueError, match="ExpertTemplate 'nonexistent' not found"):
|
||
await team.add_expert("nonexistent")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_expert_broadcasts_joined_message(self):
|
||
"""添加 Expert 时广播 expert_joined 消息"""
|
||
pool = _make_mock_pool()
|
||
transport = AsyncMock(spec=InProcessHandoffTransport)
|
||
team = ExpertTeam(pool=pool)
|
||
team._handoff_transport = transport
|
||
|
||
# 先创建团队
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
# 添加新成员
|
||
new_config = _make_expert_config(name="new_member")
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
new_expert = _make_mock_expert(name="new_member")
|
||
mock_create.return_value = new_expert
|
||
|
||
await team.add_expert(new_config)
|
||
|
||
# 验证 broadcast 消息（create_team 也会调用 send，所以检查最后一次）
|
||
calls = transport.send.call_args_list
|
||
joined_calls = [
|
||
c for c in calls if c[0][1].get("type") == "expert_joined"
|
||
]
|
||
assert len(joined_calls) >= 1
|
||
last_joined = joined_calls[-1]
|
||
assert last_joined[0][1]["expert_name"] == "new_member"
|
||
|
||
|
||
# ── ExpertTeam.remove_expert 测试 ──────────────────────────
|
||
|
||
|
||
class TestExpertTeamRemoveExpert:
|
||
"""ExpertTeam.remove_expert 移除 Expert 测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_expert(self):
|
||
"""移除普通 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1")
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
await team.remove_expert("member1")
|
||
|
||
assert "member1" not in team._experts
|
||
member_expert.destroy.assert_awaited_once_with(pool)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_lead_expert_reassigns(self):
|
||
"""移除 Lead Expert 时重新分配给下一个活跃 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1")
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
await team.remove_expert("lead")
|
||
|
||
assert team._lead_expert_name == "member1"
|
||
assert member_expert.config.is_lead is True
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_lead_expert_no_active_members(self):
|
||
"""移除 Lead Expert 且无其他活跃成员时 lead_expert_name 为 None"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
await team.remove_expert("lead")
|
||
|
||
assert team._lead_expert_name is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_nonexistent_expert_no_error(self):
|
||
"""移除不存在的 Expert 不报错"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
# 不应抛出异常
|
||
await team.remove_expert("nonexistent")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_remove_expert_broadcasts_left_message(self):
|
||
"""移除 Expert 时广播 expert_left 消息"""
|
||
pool = _make_mock_pool()
|
||
transport = AsyncMock(spec=InProcessHandoffTransport)
|
||
team = ExpertTeam(pool=pool)
|
||
team._handoff_transport = transport
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1")
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
await team.remove_expert("member1")
|
||
|
||
# 验证 expert_left 消息
|
||
calls = transport.send.call_args_list
|
||
left_calls = [c for c in calls if c[0][1].get("type") == "expert_left"]
|
||
assert len(left_calls) >= 1
|
||
last_left = left_calls[-1]
|
||
assert last_left[0][1]["expert_name"] == "member1"
|
||
|
||
|
||
# ── ExpertTeam.broadcast_user_message 测试 ─────────────────
|
||
|
||
|
||
class TestExpertTeamBroadcastUserMessage:
|
||
"""ExpertTeam.broadcast_user_message 用户干预消息广播测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_broadcast_user_message(self):
|
||
"""广播用户干预消息到团队频道"""
|
||
team = ExpertTeam()
|
||
transport = AsyncMock(spec=InProcessHandoffTransport)
|
||
team._handoff_transport = transport
|
||
|
||
await team.broadcast_user_message("请暂停执行")
|
||
|
||
transport.send.assert_awaited_once()
|
||
call_args = transport.send.call_args
|
||
assert call_args[0][0] == team._team_channel
|
||
message = call_args[0][1]
|
||
assert message["type"] == "user_intervention"
|
||
assert message["content"] == "请暂停执行"
|
||
assert "timestamp" in message
|
||
|
||
|
||
# ── ExpertTeam.get_shared_context 测试 ────────────────────
|
||
|
||
|
||
class TestExpertTeamGetSharedContext:
|
||
"""ExpertTeam.get_shared_context 共享上下文读取测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_shared_context_reads_team_keys(self):
|
||
"""读取团队范围的共享上下文"""
|
||
workspace = AsyncMock(spec=SharedWorkspace)
|
||
workspace.list_keys = AsyncMock(
|
||
return_value=[
|
||
"team:abc:output1",
|
||
"team:abc:output2",
|
||
"other:key",
|
||
]
|
||
)
|
||
workspace.read = AsyncMock(
|
||
side_effect=lambda key: {"value": f"data_{key}"} if key.startswith("team:abc") else None
|
||
)
|
||
|
||
team = ExpertTeam(team_id="abc", workspace=workspace)
|
||
|
||
context = await team.get_shared_context()
|
||
|
||
assert "team:abc:output1" in context
|
||
assert "team:abc:output2" in context
|
||
assert "other:key" not in context
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_shared_context_empty(self):
|
||
"""没有团队范围的键时返回空字典"""
|
||
workspace = AsyncMock(spec=SharedWorkspace)
|
||
workspace.list_keys = AsyncMock(return_value=[])
|
||
|
||
team = ExpertTeam(team_id="abc", workspace=workspace)
|
||
|
||
context = await team.get_shared_context()
|
||
|
||
assert context == {}
|
||
|
||
|
||
# ── ExpertTeam.dissolve 测试 ───────────────────────────────
|
||
|
||
|
||
class TestExpertTeamDissolve:
|
||
"""ExpertTeam.dissolve 团队解散测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dissolve_recycles_experts(self):
|
||
"""解散团队时回收所有 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1")
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
await team.dissolve()
|
||
|
||
assert team.status == TeamStatus.DISSOLVED
|
||
assert team.experts == []
|
||
assert team._lead_expert_name is None
|
||
lead_expert.destroy.assert_awaited_once_with(pool)
|
||
member_expert.destroy.assert_awaited_once_with(pool)
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dissolve_preserves_outputs_in_workspace(self):
|
||
"""解散团队后 SharedWorkspace 中的输出仍保留"""
|
||
workspace = SharedWorkspace()
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool, workspace=workspace)
|
||
|
||
# 写入一些数据到 workspace
|
||
await workspace.write("team:abc:result", "重要输出", "lead")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
await team.create_team(lead_config)
|
||
|
||
await team.dissolve()
|
||
|
||
# workspace 数据仍然存在
|
||
data = await workspace.read("team:abc:result")
|
||
assert data is not None
|
||
assert data["value"] == "重要输出"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dissolve_closes_handoff_transport(self):
|
||
"""解散团队时关闭 handoff_transport"""
|
||
pool = _make_mock_pool()
|
||
transport = MagicMock(spec=InProcessHandoffTransport)
|
||
transport.close = MagicMock()
|
||
team = ExpertTeam(pool=pool)
|
||
team._handoff_transport = transport
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
await team.create_team(lead_config)
|
||
|
||
await team.dissolve()
|
||
|
||
transport.close.assert_called_once()
|
||
|
||
|
||
# ── ExpertTeam 操作已解散团队测试 ──────────────────────────
|
||
|
||
|
||
class TestExpertTeamDissolvedOperations:
|
||
"""解散后的团队操作应报错"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_team_on_dissolved_raises(self):
|
||
"""在已解散的团队上 create_team 应报错（因为 pool 可能已被清理）"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
await team.create_team(lead_config)
|
||
|
||
await team.dissolve()
|
||
|
||
# 解散后状态为 DISSOLVED
|
||
assert team.status == TeamStatus.DISSOLVED
|
||
|
||
|
||
# ── ExpertTeam.lead_expert 属性测试 ────────────────────────
|
||
|
||
|
||
class TestExpertTeamLeadExpert:
|
||
"""ExpertTeam.lead_expert 属性测试"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_lead_expert_returns_lead(self):
|
||
"""lead_expert 返回 Lead Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
mock_create.return_value = lead_expert
|
||
await team.create_team(lead_config)
|
||
|
||
assert team.lead_expert is lead_expert
|
||
|
||
def test_lead_expert_none_when_no_lead(self):
|
||
"""没有 Lead Expert 时返回 None"""
|
||
team = ExpertTeam()
|
||
|
||
assert team.lead_expert is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_active_experts_filters_inactive(self):
|
||
"""active_experts 只返回活跃的 Expert"""
|
||
pool = _make_mock_pool()
|
||
team = ExpertTeam(pool=pool)
|
||
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
member_config = _make_expert_config(name="member1")
|
||
|
||
with patch.object(Expert, "create", new_callable=AsyncMock) as mock_create:
|
||
lead_expert = _make_mock_expert(name="lead", is_lead=True)
|
||
member_expert = _make_mock_expert(name="member1")
|
||
mock_create.side_effect = [lead_expert, member_expert]
|
||
await team.create_team(lead_config, [member_config])
|
||
|
||
# 标记 member1 为非活跃
|
||
member_expert.is_active = False
|
||
|
||
active = team.active_experts
|
||
assert len(active) == 1
|
||
assert active[0] is lead_expert
|
||
|
||
|
||
# ── ExpertTeam._build_team_context 测试 ────────────────────
|
||
|
||
|
||
class TestExpertTeamBuildContext:
|
||
"""ExpertTeam._build_team_context 团队上下文构建测试"""
|
||
|
||
def test_build_team_context_with_lead_and_members(self):
|
||
"""构建包含 Lead 和成员的团队上下文"""
|
||
team = ExpertTeam()
|
||
|
||
lead_config = _make_expert_config(
|
||
name="lead", persona="领导者", is_lead=True
|
||
)
|
||
member_config = _make_expert_config(
|
||
name="analyst", persona="分析师", bound_skills=["data_query"]
|
||
)
|
||
|
||
context = team._build_team_context(lead_config, [member_config])
|
||
|
||
assert "hub-and-spoke mode" in context
|
||
assert "Lead Expert (hub): lead (领导者)" in context
|
||
assert "Team Member (spoke): analyst (分析师)" in context
|
||
assert "data_query" in context
|
||
assert "depth=1" in context
|
||
|
||
def test_build_team_context_no_lead(self):
|
||
"""没有 Lead Expert 时构建上下文"""
|
||
team = ExpertTeam()
|
||
member_config = _make_expert_config(name="analyst")
|
||
|
||
context = team._build_team_context(None, [member_config])
|
||
|
||
# 不应出现具体的 Lead Expert (hub): name 行
|
||
assert "Lead Expert (hub):" not in context
|
||
assert "Team Member (spoke): analyst" in context
|
||
|
||
def test_build_team_context_skips_lead_in_members(self):
|
||
"""成员列表中包含 Lead 时跳过"""
|
||
team = ExpertTeam()
|
||
lead_config = _make_expert_config(name="lead", is_lead=True)
|
||
|
||
context = team._build_team_context(lead_config, [lead_config])
|
||
|
||
# Lead 不应出现在 Team Member 行
|
||
assert "Team Member (spoke): lead" not in context
|