From 54cfaa4eb58dd7965c9d5bde2646a33bc76d4d4d Mon Sep 17 00:00:00 2001 From: cfdaily Date: Sun, 17 May 2026 21:16:27 +0800 Subject: [PATCH] auto-sync: 2026-05-17 21:16:27 --- tests/test_router.py | 272 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 272 insertions(+) create mode 100644 tests/test_router.py diff --git a/tests/test_router.py b/tests/test_router.py new file mode 100644 index 0000000..a94ae59 --- /dev/null +++ b/tests/test_router.py @@ -0,0 +1,272 @@ +"""Router 单元测试 — 三种路由模式 + 校验 + fallback""" + +import pytest +from unittest.mock import MagicMock, patch + +from src.daemon.router import ( + AgentRouter, AgentProfile, LLMDriver, RouteDecision, KNOWN_CAPABILITIES, +) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def profiles(): + return { + "zhangfei-dev": AgentProfile( + agent_id="zhangfei-dev", + capabilities=["coding", "implementation", "scripting"], + can_review=False, max_concurrent=1, + ), + "simayi-challenger": AgentProfile( + agent_id="simayi-challenger", + capabilities=["review", "quality_check", "debate"], + can_review=True, max_concurrent=2, + ), + "pangtong-fujunshi": AgentProfile( + agent_id="pangtong-fujunshi", + capabilities=["planning", "coordination", "escalation", "strategy"], + can_review=True, is_fallback=True, max_concurrent=3, + ), + } + + +@pytest.fixture +def router(profiles): + return AgentRouter(agent_profiles=profiles) + + +def make_task(**overrides): + base = { + "id": "t1", "title": "Test Task", "status": "pending", + "assignee": "zhangfei-dev", "current_agent": None, + "next_capability": None, "task_type": "coding", + "description": "Write some code", + } + base.update(overrides) + return base + + +# --------------------------------------------------------------------------- +# T1: 确定性快速路径 +# --------------------------------------------------------------------------- + +class TestDeterministic: + def test_local_action(self, router): + """机械检查 → daemon""" + d = router.route(make_task(), action_type="L1_guardrail") + assert d.agent_id == "daemon" + assert d.mode == "deterministic" + + def test_format_check_local(self, router): + d = router.route(make_task(), action_type="format_check") + assert d.agent_id == "daemon" + + def test_retry_same_agent(self, router): + """retry → 原执行者""" + d = router.route(make_task(current_agent="zhangfei-dev"), + action_type="retry") + assert d.agent_id == "zhangfei-dev" + assert d.mode == "deterministic" + + def test_direct_assignee(self, router): + """有 assignee 且非生命周期流转 → 直接用""" + d = router.route(make_task(), action_type="execute") + assert d.agent_id == "zhangfei-dev" + assert d.mode == "deterministic" + + +# --------------------------------------------------------------------------- +# T2: Mode B — Agent 声明式交接 +# --------------------------------------------------------------------------- + +class TestAgentHandoff: + def test_handoff_review(self, router): + """张飞说需要 review → 匹配司马懿""" + d = router.route(make_task( + current_agent="zhangfei-dev", + next_capability="review", + )) + assert d.agent_id == "simayi-challenger" + assert d.mode == "agent_handoff" + assert "review" in d.reason + + def test_handoff_excludes_current(self, router): + """交接排除当前执行者（不能交给自己）""" + d = router.route(make_task( + current_agent="simayi-challenger", + next_capability="review", + )) + # simayi 被排除，只剩 pangtong（can_review=True 且有 review 能力） + assert d.agent_id != "simayi-challenger" + + def test_handoff_invalid_capability_ignored(self, router): + """BUG-2: 非法 next_capability 被忽略，不进路由""" + d = router.route(make_task( + current_agent="zhangfei-dev", + next_capability="nonexistent_capability", + ), action_type="execute") + # 不应该走 handoff，应该走 assignee 直派或其他 + assert d.mode != "agent_handoff" + + def test_handoff_known_but_no_match(self, router): + """合法 capability 但无匹配 Agent → 降级""" + # "escalation" 只有 pangtong 有，如果排除 pangton... + # 测试 "data" 有 zhaoyun 但 zhaoyun 不在 profiles 里 + router2 = AgentRouter(agent_profiles={ + "zhangfei-dev": AgentProfile( + agent_id="zhangfei-dev", + capabilities=["coding"], + ), + }) + d = router2.route(make_task( + current_agent="zhangfei-dev", + next_capability="review", # 合法但在 profiles 中无人匹配 + )) + # 无 LLM driver → fallback + assert d.agent_id == "pangtong-fujunshi" + assert d.mode == "fallback" + + +# --------------------------------------------------------------------------- +# T3: 生命周期流转 +# --------------------------------------------------------------------------- + +class TestLifecycle: + def test_review_action(self, router): + """review action_type → 查 review 能力，排除 current_agent""" + d = router.route(make_task( + current_agent="zhangfei-dev", + ), action_type="review") + assert d.agent_id == "simayi-challenger" + assert d.mode == "deterministic" + + def test_review_excludes_executor(self, router): + """review 排除执行者（张飞不能审张飞）""" + d = router.route(make_task( + current_agent="zhangfei-dev", + assignee="zhangfei-dev", + ), action_type="review") + assert d.agent_id != "zhangfei-dev" + assert d.agent_id == "simayi-challenger" + + +# --------------------------------------------------------------------------- +# T4: Mode A — LLM 路由 +# --------------------------------------------------------------------------- + +class TestLLMRoute: + def test_llm_returns_valid(self, router): + """LLM 返回合法 Agent""" + mock_driver = MagicMock(spec=LLMDriver) + mock_driver.route.return_value = RouteDecision( + agent_id="simayi-challenger", + reason="LLM selected reviewer", + mode="llm_route", + confidence=0.9, + model="zhipu/glm-5.1", + latency_ms=1200, + ) + router.llm_driver = mock_driver + + # 无 assignee、无 next_capability、非 lifecycle → 走 LLM + d = router.route(make_task( + assignee=None, current_agent=None, + next_capability=None, + )) + assert d.agent_id == "simayi-challenger" + assert d.mode == "llm_route" + assert d.confidence == 0.9 + + def test_llm_invalid_agent_fallback(self, router): + """LLM 返回不存在的 Agent → fallback""" + mock_driver = MagicMock(spec=LLMDriver) + mock_driver.route.return_value = RouteDecision( + agent_id="nonexistent-agent", + reason="Bad pick", + mode="llm_route", + confidence=0.8, + latency_ms=500, + ) + router.llm_driver = mock_driver + + d = router.route(make_task(assignee=None)) + assert d.agent_id == "pangtong-fujunshi" + assert d.mode == "fallback" + + def test_llm_low_confidence_fallback(self, router): + """LLM 置信度低于阈值 → fallback""" + mock_driver = MagicMock(spec=LLMDriver) + mock_driver.route.return_value = RouteDecision( + agent_id="simayi-challenger", + reason="Not sure", + mode="llm_route", + confidence=0.5, + latency_ms=300, + ) + router.llm_driver = mock_driver + + d = router.route(make_task(assignee=None)) + assert d.agent_id == "pangtong-fujunshi" + assert d.mode == "fallback" + assert d.confidence == 0.5 + + +# --------------------------------------------------------------------------- +# T5: Fallback +# --------------------------------------------------------------------------- + +class TestFallback: + def test_no_llm_no_match(self, router): + """无 LLM、无匹配 → fallback 庞统""" + d = router.route(make_task( + assignee=None, current_agent=None, + next_capability=None, task_type="unknown_type", + ), action_type="") + assert d.agent_id == "pangtong-fujunshi" + assert d.mode == "fallback" + + +# --------------------------------------------------------------------------- +# T6: Capability 校验 +# --------------------------------------------------------------------------- + +class TestCapabilityValidation: + def test_known_capabilities_populated(self, router): + """profiles 自动填充 known capabilities""" + assert "coding" in router._known_capabilities + assert "review" in router._known_capabilities + assert "escalation" in router._known_capabilities + + def test_validate_known(self, router): + assert router._validate_capability("coding") is True + assert router._validate_capability("review") is True + + def test_validate_unknown(self, router): + assert router._validate_capability("administration") is False + assert router._validate_capability("hack") is False + + def test_known_capabilities_set(self): + """KNOWN_CAPABILITIES 包含所有预期能力""" + assert "coding" in KNOWN_CAPABILITIES + assert "review" in KNOWN_CAPABILITIES + assert "escalation" in KNOWN_CAPABILITIES + + +# --------------------------------------------------------------------------- +# T7: Latency 记录 +# --------------------------------------------------------------------------- + +class TestLatency: + def test_deterministic_has_latency(self, router): + d = router.route(make_task(), action_type="L1_guardrail") + assert d.latency_ms >= 0 + + def test_handoff_has_latency(self, router): + d = router.route(make_task( + current_agent="zhangfei-dev", + next_capability="review", + )) + assert d.latency_ms >= 0