265 lines
9.0 KiB
Python
265 lines
9.0 KiB
Python
"""
|
||
Neural Sync - Role Dispatcher Unit Tests
|
||
"""
|
||
|
||
import pytest
|
||
from backend.app.dispatcher import (
|
||
Role,
|
||
RoutingRule,
|
||
RoleRouter,
|
||
RoutingResult,
|
||
route_signal,
|
||
get_router,
|
||
)
|
||
|
||
|
||
class TestRole:
|
||
"""测试 Role 枚举"""
|
||
|
||
def test_role_values(self):
|
||
"""测试角色值"""
|
||
assert Role.BOSS.value == "boss"
|
||
assert Role.AI_ENGINEER.value == "ai_engineer"
|
||
assert Role.BACKEND.value == "backend"
|
||
assert Role.FRONTEND.value == "frontend"
|
||
|
||
|
||
class TestRoleRouter:
|
||
"""测试 RoleRouter"""
|
||
|
||
def test_router_initialization(self):
|
||
"""测试路由器初始化"""
|
||
router = RoleRouter()
|
||
assert len(router.ROLE_RULES) == 4
|
||
assert router._compiled_patterns is not None
|
||
|
||
def test_get_role_info(self):
|
||
"""测试获取角色信息"""
|
||
router = RoleRouter()
|
||
boss_info = router.get_role_info(Role.BOSS)
|
||
assert boss_info["role"] == "boss"
|
||
assert boss_info["name"] == "战略决策"
|
||
assert "keywords" in boss_info
|
||
|
||
def test_get_all_roles(self):
|
||
"""测试获取所有角色"""
|
||
router = RoleRouter()
|
||
roles = router.get_all_roles()
|
||
assert len(roles) == 4
|
||
# 应该按优先级降序排列
|
||
assert roles[0]["priority"] >= roles[1]["priority"]
|
||
|
||
def test_route_to_boss_by_cost(self):
|
||
"""测试根据成本路由到老板"""
|
||
router = RoleRouter()
|
||
analysis = {"resource_cost": {"dev_days": 10}}
|
||
role, confidence, details = router.route("需要重构系统架构", analysis)
|
||
assert role == Role.BOSS
|
||
assert confidence == 1.0
|
||
assert "threshold_check" in details.get("matched_rules", [])
|
||
|
||
def test_route_to_boss_by_risk(self):
|
||
"""测试根据风险路由到老板"""
|
||
router = RoleRouter()
|
||
analysis = {"risk_level": "CRITICAL"}
|
||
role, confidence, details = router.route("修复安全漏洞", analysis)
|
||
assert role == Role.BOSS
|
||
assert confidence == 1.0
|
||
|
||
def test_route_to_ai_engineer(self):
|
||
"""测试路由到 AI 工程师"""
|
||
router = RoleRouter()
|
||
role, confidence, details = router.route("优化 prompt 提高模型准确率", {})
|
||
assert role == Role.AI_ENGINEER
|
||
assert confidence > 0
|
||
|
||
def test_route_to_backend(self):
|
||
"""测试路由到后端开发"""
|
||
router = RoleRouter()
|
||
role, confidence, details = router.route("使用 FastAPI 重构 API 接口", {})
|
||
assert role == Role.BACKEND
|
||
assert confidence > 0
|
||
|
||
def test_route_to_frontend(self):
|
||
"""测试路由到前端开发"""
|
||
router = RoleRouter()
|
||
role, confidence, details = router.route("使用 Vue 3 重构仪表盘组件", {})
|
||
assert role == Role.FRONTEND
|
||
assert confidence > 0
|
||
|
||
def test_route_with_multiple_keywords(self):
|
||
"""测试多个关键词匹配"""
|
||
router = RoleRouter()
|
||
content = "使用 FastAPI 和 PostgreSQL 优化数据库查询性能"
|
||
role, confidence, details = router.route(content, {})
|
||
assert role == Role.BACKEND
|
||
# 多个关键词应该提高置信度
|
||
assert confidence > 0.5
|
||
|
||
def test_route_with_pattern_matching(self):
|
||
"""测试正则模式匹配"""
|
||
router = RoleRouter()
|
||
content = "创建新的 component.tsx 文件"
|
||
role, confidence, details = router.route(content, {})
|
||
# 应该匹配到前端（tsx 模式）
|
||
assert role == Role.FRONTEND
|
||
|
||
def test_route_default_to_ai_engineer(self):
|
||
"""测试默认路由到 AI 工程师"""
|
||
router = RoleRouter()
|
||
# 没有明显关键词的内容
|
||
role, confidence, details = router.route("这是一些普通的描述文字", {})
|
||
assert role == Role.AI_ENGINEER
|
||
|
||
def test_route_with_low_confidence(self):
|
||
"""测试低置信度路由"""
|
||
router = RoleRouter()
|
||
role, confidence, details = router.route("处理用户反馈", {})
|
||
assert confidence < 0.5 # 应该是较低的置信度
|
||
|
||
def test_route_tie_breaker_by_priority(self):
|
||
"""测试优先级平局决胜"""
|
||
router = RoleRouter()
|
||
# 同时匹配前端和后端关键词，但后端优先级应该更高
|
||
content = "修复 API 接口并更新 UI 组件"
|
||
role, confidence, details = router.route(content, {})
|
||
# 应该根据优先级选择
|
||
assert role in [Role.BACKEND, Role.FRONTEND]
|
||
|
||
def test_route_details(self):
|
||
"""测试路由详情"""
|
||
router = RoleRouter()
|
||
role, confidence, details = router.route("使用 FastAPI 创建 REST API", {})
|
||
|
||
assert "matched_rules" in details
|
||
assert "scores" in details
|
||
assert "reasons" in details
|
||
assert "final_scores" in details
|
||
|
||
def test_should_escalate_to_boss(self):
|
||
"""测试是否应该升级到老板"""
|
||
router = RoleRouter()
|
||
|
||
# 测试成本阈值
|
||
assert router._should_escalate_to_boss({"resource_cost": {"dev_days": 5}})
|
||
assert router._should_escalate_to_boss({"resource_cost": {"dev_days": 10}})
|
||
|
||
# 测试风险阈值
|
||
assert router._should_escalate_to_boss({"risk_level": "CRITICAL"})
|
||
assert router._should_escalate_to_boss({"risk_level": "HIGH"})
|
||
assert not router._should_escalate_to_boss({"risk_level": "MEDIUM"})
|
||
|
||
# 测试架构变更
|
||
assert router._should_escalate_to_boss({"is_architecture_change": True})
|
||
|
||
# 测试不需要升级
|
||
assert not router._should_escalate_to_boss({})
|
||
|
||
|
||
class TestRoutingResult:
|
||
"""测试 RoutingResult"""
|
||
|
||
def test_routing_result_creation(self):
|
||
"""测试创建路由结果"""
|
||
result = RoutingResult(
|
||
role="boss",
|
||
confidence=0.95,
|
||
details={"matched_rules": ["threshold_check"]},
|
||
signal_id="test-123",
|
||
)
|
||
assert result.role == "boss"
|
||
assert result.confidence == 0.95
|
||
assert result.signal_id == "test-123"
|
||
assert result.timestamp is not None
|
||
|
||
|
||
class TestGlobalRouter:
|
||
"""测试全局路由器"""
|
||
|
||
def test_get_router_singleton(self):
|
||
"""测试获取单例"""
|
||
router1 = get_router()
|
||
router2 = get_router()
|
||
assert router1 is router2
|
||
|
||
|
||
class TestConvenienceFunctions:
|
||
"""测试便捷函数"""
|
||
|
||
def test_route_signal(self):
|
||
"""测试路由信号便捷函数"""
|
||
result = route_signal("优化 prompt 提高准确率", signal_id="signal-1")
|
||
|
||
assert isinstance(result, RoutingResult)
|
||
assert result.signal_id == "signal-1"
|
||
assert result.role in ["boss", "ai_engineer", "backend", "frontend"]
|
||
assert 0 <= result.confidence <= 1
|
||
|
||
|
||
class TestRealWorldScenarios:
|
||
"""真实场景测试"""
|
||
|
||
def test_security_issue(self):
|
||
"""测试安全问题 - 应该路由到老板"""
|
||
result = route_signal("发现严重的 SQL 注入漏洞，需要紧急修复", {"risk_level": "CRITICAL"})
|
||
assert result.role == "boss"
|
||
|
||
def test_large_feature_request(self):
|
||
"""测试大功能需求 - 应该路由到老板"""
|
||
result = route_signal(
|
||
"重写整个用户认证系统，支持 OAuth2 和 SAML",
|
||
{"resource_cost": {"dev_days": 15}},
|
||
)
|
||
assert result.role == "boss"
|
||
|
||
def test_llm_optimization(self):
|
||
"""测试 LLM 优化 - 应该路由到 AI 工程师"""
|
||
result = route_signal("优化 embedding 模型提高检索准确率")
|
||
assert result.role == "ai_engineer"
|
||
|
||
def test_api_development(self):
|
||
"""测试 API 开发 - 应该路由到后端"""
|
||
result = route_signal("创建新的 REST API 端点用于用户管理")
|
||
assert result.role == "backend"
|
||
|
||
def test_frontend_component(self):
|
||
"""测试前端组件 - 应该路由到前端"""
|
||
result = route_signal("使用 Vue 3 创建新的数据可视化组件")
|
||
assert result.role == "frontend"
|
||
|
||
def test_ambiguous_request(self):
|
||
"""测试模糊需求 - 应该有默认行为"""
|
||
result = route_signal("优化系统性能")
|
||
# 应该有一个合理的默认路由
|
||
assert result.role in ["boss", "ai_engineer", "backend", "frontend"]
|
||
# 置信度应该较低
|
||
assert result.confidence < 0.7
|
||
|
||
|
||
class TestEdgeCases:
|
||
"""边缘情况测试"""
|
||
|
||
def test_empty_content(self):
|
||
"""测试空内容"""
|
||
result = route_signal("")
|
||
assert result.role in ["boss", "ai_engineer", "backend", "frontend"]
|
||
|
||
def test_chinese_content(self):
|
||
"""测试中文内容"""
|
||
result = route_signal("使用 Vue 3 开发前端组件")
|
||
assert result.role == "frontend"
|
||
|
||
def test_mixed_language_content(self):
|
||
"""测试混合语言内容"""
|
||
result = route_signal("创建新的 FastAPI endpoint 用于数据处理")
|
||
assert result.role == "backend"
|
||
|
||
def test_technical_jargon(self):
|
||
"""测试技术术语"""
|
||
result = route_signal("Implement RAG pipeline with vector database")
|
||
assert result.role == "ai_engineer"
|
||
|
||
|
||
if __name__ == "__main__":
|
||
pytest.main([__file__, "-v"])
|