Simon
628 lines
21 KiB
Python
628 lines
21 KiB
Python
# =============================================================================
|
|
# 企微IT智能服务台 — 阶段5 自动化闭环 服务单元测试
|
|
# =============================================================================
|
|
# 说明：测试核心服务：session_manager、intent_router、executor、approval
|
|
# 创建日期: 2026-07-01
|
|
# =============================================================================
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime
|
|
|
|
from app.models.automation import (
|
|
AutoSession,
|
|
AutoAction,
|
|
ApprovalTicket,
|
|
ScenarioConfig,
|
|
)
|
|
from app.services.automation.session_manager import AutoSessionService
|
|
from app.services.automation.intent_router import IntentRouter
|
|
from app.services.automation.executor import ActionExecutor
|
|
from app.services.automation.approval import ApprovalService
|
|
from app.services.automation.exception_handler import AutomationException
|
|
from app.constants import AutomationErrorCode
|
|
|
|
|
|
# =============================================================================
|
|
# 测试 ApprovalService
|
|
# =============================================================================
|
|
class TestApprovalService:
|
|
"""审批单服务测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_ticket_idempotent(self, db_session):
|
|
"""测试 ensure_ticket 幂等性：同一动作多次调用应返回同一审批单"""
|
|
# 创建会话和动作
|
|
session = AutoSession(
|
|
conversation_id="test_conv_001",
|
|
employee_id="emp_001",
|
|
title="测试会话",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=0,
|
|
action_type="virus_scan",
|
|
risk_level="high",
|
|
title="病毒扫描",
|
|
description="扫描终端",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
await db_session.flush()
|
|
|
|
# 第一次调用 ensure_ticket
|
|
svc = ApprovalService(db_session)
|
|
ticket1 = await svc.ensure_ticket(action, channel="agent", reason="高危操作")
|
|
|
|
# 第二次调用 ensure_ticket（幂等性测试）
|
|
ticket2 = await svc.ensure_ticket(action, channel="agent", reason="高危操作")
|
|
|
|
assert ticket1.id == ticket2.id, "幂等性：同一动作应返回同一审批单"
|
|
assert ticket1.status == "pending", "审批单状态应为 pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ensure_ticket_creates_new_when_no_pending(self, db_session):
|
|
"""测试 ensure_ticket 创建新审批单"""
|
|
session = AutoSession(
|
|
conversation_id="test_conv_002",
|
|
employee_id="emp_002",
|
|
title="测试会话2",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=0,
|
|
action_type="virus_quarantine",
|
|
risk_level="high",
|
|
title="病毒隔离",
|
|
description="隔离终端",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
await db_session.flush()
|
|
|
|
svc = ApprovalService(db_session)
|
|
ticket = await svc.ensure_ticket(action, channel="h5", reason="需要员工确认")
|
|
|
|
assert ticket is not None, "应创建审批单"
|
|
assert ticket.action_id == action.id, "审批单应关联动作"
|
|
assert ticket.channel == "h5", "渠道应为 h5"
|
|
assert ticket.status == "pending", "状态应为 pending"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decide_approve(self, db_session):
|
|
"""测试审批通过"""
|
|
session = AutoSession(
|
|
conversation_id="test_conv_003",
|
|
employee_id="emp_003",
|
|
title="测试会话3",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=0,
|
|
action_type="password_reset_link",
|
|
risk_level="high",
|
|
title="密码重置",
|
|
description="重置密码",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
await db_session.flush()
|
|
|
|
svc = ApprovalService(db_session)
|
|
ticket = await svc.ensure_ticket(action, channel="agent", reason="重置密码")
|
|
|
|
# 审批通过
|
|
updated = await svc.decide(ticket.id, "approve", "同意", "agent_001")
|
|
|
|
assert updated.status == "approved", "审批单状态应为 approved"
|
|
assert updated.decision_note == "同意", "应有审批意见"
|
|
assert updated.approver_id == "agent_001", "应有审批人"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decide_reject(self, db_session):
|
|
"""测试审批驳回"""
|
|
session = AutoSession(
|
|
conversation_id="test_conv_004",
|
|
employee_id="emp_004",
|
|
title="测试会话4",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=0,
|
|
action_type="virus_quarantine",
|
|
risk_level="high",
|
|
title="病毒隔离",
|
|
description="隔离终端",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
await db_session.flush()
|
|
|
|
svc = ApprovalService(db_session)
|
|
ticket = await svc.ensure_ticket(action, channel="agent", reason="隔离终端")
|
|
|
|
# 审批驳回
|
|
updated = await svc.decide(ticket.id, "reject", "风险过高", "agent_002")
|
|
|
|
assert updated.status == "rejected", "审批单状态应为 rejected"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decide_nonexistent_raises(self, db_session):
|
|
"""测试审批单不存在时抛出异常"""
|
|
svc = ApprovalService(db_session)
|
|
|
|
with pytest.raises(ValueError, match="审批单不存在"):
|
|
await svc.decide("nonexistent_ticket_id", "approve", "同意", "agent_001")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_pending_for_action(self, db_session):
|
|
"""测试查询动作待决审批单"""
|
|
session = AutoSession(
|
|
conversation_id="test_conv_005",
|
|
employee_id="emp_005",
|
|
title="测试会话5",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=0,
|
|
action_type="terminal_locate",
|
|
risk_level="read",
|
|
title="终端定位",
|
|
description="定位终端",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
await db_session.flush()
|
|
|
|
svc = ApprovalService(db_session)
|
|
ticket = await svc.ensure_ticket(action, channel="agent", reason="查询终端")
|
|
|
|
pending = await svc.get_pending_for_action(action.id)
|
|
assert pending is not None, "应能找到待决审批单"
|
|
assert pending.id == ticket.id, "应为同一审批单"
|
|
|
|
|
|
# =============================================================================
|
|
# 测试 IntentRouter
|
|
# =============================================================================
|
|
class TestIntentRouter:
|
|
"""意图识别路由测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_with_fallback_intent(self):
|
|
"""测试关键词兜底意图识别"""
|
|
router = IntentRouter()
|
|
|
|
# 测试病毒相关关键词
|
|
result = await router.detect("电脑中病毒了", "emp_001")
|
|
|
|
assert result is not None, "应返回识别结果"
|
|
assert "scenario_key" in result, "应包含 scenario_key"
|
|
assert result.get("confidence") is not None, "应包含置信度"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_intent_password_reset(self):
|
|
"""测试密码重置关键词"""
|
|
router = IntentRouter()
|
|
|
|
result = await router.detect("我忘记密码了", "emp_002")
|
|
|
|
assert result.get("scenario_key") == "password_reset", "应识别为密码重置场景"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_intent_software_install(self):
|
|
"""测试软件安装关键词"""
|
|
router = IntentRouter()
|
|
|
|
result = await router.detect("帮我安装一个软件", "emp_003")
|
|
|
|
assert result.get("scenario_key") == "software_install", "应识别为软件安装场景"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_intent_terminal_locate(self):
|
|
"""测试终端定位关键词"""
|
|
router = IntentRouter()
|
|
|
|
result = await router.detect("我的电脑在哪", "emp_004")
|
|
|
|
assert result.get("scenario_key") == "terminal_locate", "应识别为终端定位场景"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fallback_intent_unknown(self):
|
|
"""测试未知意图返回空"""
|
|
router = IntentRouter()
|
|
|
|
result = await router.detect("今天天气不错", "emp_005")
|
|
|
|
assert result.get("scenario_key") is None, "未知意图应返回空场景"
|
|
assert result.get("confidence") == 0.0, "置信度应为0"
|
|
|
|
|
|
# =============================================================================
|
|
# 测试 AutoSessionService - 会话CRUD
|
|
# =============================================================================
|
|
class TestAutoSessionService:
|
|
"""自动化会话服务测试 - CRUD"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_session(self, db_session):
|
|
"""测试创建会话"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
session = await svc.create_session(
|
|
conversation_id="conv_001",
|
|
employee_id="emp_001",
|
|
description="测试会话",
|
|
mode="real_exec",
|
|
)
|
|
|
|
assert session is not None, "应创建会话"
|
|
assert session.employee_id == "emp_001", "员工ID应正确"
|
|
assert session.status == "created", "初始状态应为 created"
|
|
assert session.mode == "real_exec", "执行模式应正确"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session(self, db_session):
|
|
"""测试获取会话"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 先创建
|
|
created = await svc.create_session(
|
|
conversation_id="conv_002",
|
|
employee_id="emp_002",
|
|
description="测试获取会话",
|
|
mode="real_exec",
|
|
)
|
|
|
|
# 再获取
|
|
retrieved = await svc.get_session(created.id)
|
|
|
|
assert retrieved is not None, "应能获取会话"
|
|
assert retrieved.id == created.id, "会话ID应一致"
|
|
assert retrieved.employee_id == "emp_002", "员工ID应正确"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_not_found(self, db_session):
|
|
"""测试获取不存在的会话"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
result = await svc.get_session("nonexistent_id")
|
|
|
|
assert result is None, "不存在的会话应返回 None"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions(self, db_session):
|
|
"""测试列出会话"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 创建多个会话
|
|
for i in range(3):
|
|
await svc.create_session(
|
|
conversation_id=f"conv_{i}",
|
|
employee_id="emp_003",
|
|
description=f"测试会话{i}",
|
|
mode="real_exec",
|
|
)
|
|
|
|
sessions = await svc.list_sessions(employee_id="emp_003")
|
|
|
|
assert len(sessions) == 3, "应返回3个会话"
|
|
# 验证按时间倒序
|
|
assert sessions[0].created_at >= sessions[-1].created_at, "应按时间倒序"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_sessions_with_status_filter(self, db_session):
|
|
"""测试按状态过滤"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 创建会话
|
|
s1 = await svc.create_session(
|
|
conversation_id="conv_status_1",
|
|
employee_id="emp_status",
|
|
description="会话1",
|
|
mode="real_exec",
|
|
)
|
|
s2 = await svc.create_session(
|
|
conversation_id="conv_status_2",
|
|
employee_id="emp_status",
|
|
description="会话2",
|
|
mode="real_exec",
|
|
)
|
|
|
|
# 手动修改状态
|
|
s2.status = "closed"
|
|
await db_session.flush()
|
|
|
|
# 按 created 状态过滤
|
|
created_sessions = await svc.list_sessions(
|
|
employee_id="emp_status",
|
|
status="created"
|
|
)
|
|
|
|
assert len(created_sessions) == 1, "应返回1个 created 状态的会话"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_takeover(self, db_session):
|
|
"""测试转人工接管"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
session = await svc.create_session(
|
|
conversation_id="conv_takeover",
|
|
employee_id="emp_takeover",
|
|
description="测试转人工",
|
|
mode="real_exec",
|
|
)
|
|
|
|
# 转人工接管
|
|
result = await svc.takeover(session.id, agent_id="agent_001", note="测试转接")
|
|
|
|
assert result.status == "handoff", "状态应为 handoff"
|
|
assert result.agent_id == "agent_001", "接管坐席ID应正确"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_takeover_not_found(self, db_session):
|
|
"""测试转接不存在的会话"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
with pytest.raises(AutomationException) as exc:
|
|
await svc.takeover("nonexistent_id", "agent_001")
|
|
|
|
assert exc.value.code == AutomationErrorCode.SESSION_NOT_FOUND
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_feedback_satisfied(self, db_session):
|
|
"""测试员工满意反馈"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
session = await svc.create_session(
|
|
conversation_id="conv_feedback",
|
|
employee_id="emp_feedback",
|
|
description="测试反馈",
|
|
mode="real_exec",
|
|
)
|
|
session.status = "resolved" # 先设为 resolved
|
|
await db_session.flush()
|
|
|
|
result = await svc.resolve_feedback(session.id, satisfied=True, note="满意")
|
|
|
|
assert result.status == "closed", "状态应为 closed"
|
|
assert result.closed_by == "emp_feedback", "关单人应为员工"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolve_feedback_unsatisfied(self, db_session):
|
|
"""测试员工不满意反馈"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
session = await svc.create_session(
|
|
conversation_id="conv_feedback_2",
|
|
employee_id="emp_feedback_2",
|
|
description="测试反馈2",
|
|
mode="real_exec",
|
|
)
|
|
session.status = "resolved"
|
|
await db_session.flush()
|
|
|
|
result = await svc.resolve_feedback(session.id, satisfied=False, note="不满意")
|
|
|
|
assert result.status == "handoff", "状态应为 handoff"
|
|
assert result.closed_by == "employee(reject)", "关单人应为员工拒绝"
|
|
|
|
|
|
# =============================================================================
|
|
# 测试 ActionExecutor - 执行引擎
|
|
# =============================================================================
|
|
class TestActionExecutor:
|
|
"""动作执行引擎测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_load_session_and_actions(self, db_session):
|
|
"""测试加载会话和动作"""
|
|
# 创建会话
|
|
session = AutoSession(
|
|
conversation_id="exec_conv_001",
|
|
employee_id="emp_exec_001",
|
|
title="执行测试",
|
|
status="running",
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
# 创建动作
|
|
actions = []
|
|
for i in range(3):
|
|
action = AutoAction(
|
|
session_id=session.id,
|
|
action_index=i,
|
|
action_type="software_install_guide",
|
|
risk_level="low",
|
|
title=f"动作{i}",
|
|
description=f"描述{i}",
|
|
status="pending",
|
|
)
|
|
db_session.add(action)
|
|
actions.append(action)
|
|
await db_session.flush()
|
|
|
|
# 测试加载
|
|
executor = ActionExecutor(db_session)
|
|
loaded_session, loaded_actions = await executor._load(session.id)
|
|
|
|
assert loaded_session is not None, "应加载到会话"
|
|
assert len(loaded_actions) == 3, "应加载3个动作"
|
|
# 验证排序
|
|
assert loaded_actions[0].action_index == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_with_terminal_state(self, db_session):
|
|
"""测试终态会话跳过执行"""
|
|
session = AutoSession(
|
|
conversation_id="exec_conv_002",
|
|
employee_id="emp_exec_002",
|
|
title="终态测试",
|
|
status="closed", # 终态
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
executor = ActionExecutor(db_session)
|
|
await executor.run(session.id)
|
|
|
|
# 不应抛出异常，应直接返回
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_clients(self, db_session):
|
|
"""测试构建外部客户端"""
|
|
executor = ActionExecutor(db_session)
|
|
clients = await executor._build_clients()
|
|
|
|
assert isinstance(clients, dict), "应返回客户端字典"
|
|
# 验证包含必要的客户端键
|
|
assert "huorong" in clients
|
|
assert "lianruan" in clients
|
|
assert "ehr" in clients
|
|
assert "dify" in clients
|
|
|
|
|
|
# =============================================================================
|
|
# 测试场景配置管理
|
|
# =============================================================================
|
|
class TestScenarioConfig:
|
|
"""场景配置管理测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_scenario_configs(self, db_session):
|
|
"""测试列出场景配置"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 创建场景配置
|
|
config = ScenarioConfig(
|
|
scenario_key="test_scenario",
|
|
name="测试场景",
|
|
description="测试用场景",
|
|
enabled=True,
|
|
actions=[],
|
|
)
|
|
db_session.add(config)
|
|
await db_session.flush()
|
|
|
|
configs = await svc.list_scenario_configs()
|
|
|
|
assert len(configs) >= 1, "应返回场景配置"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_scenario_config_create(self, db_session):
|
|
"""测试创建场景配置"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
data = {
|
|
"name": "新场景",
|
|
"description": "新场景描述",
|
|
"enabled": True,
|
|
"actions": [
|
|
{
|
|
"action_type": "software_install_guide",
|
|
"adapter": "internal",
|
|
"risk_level": "low",
|
|
"title": "安装软件",
|
|
"description": "安装软件",
|
|
}
|
|
],
|
|
"approval_strategy": {"read": "auto", "low": "auto", "high": "approval"},
|
|
}
|
|
|
|
config = await svc.upsert_scenario_config("new_scene", data, operator="test_admin")
|
|
|
|
assert config.scenario_key == "new_scene", "场景键应正确"
|
|
assert config.name == "新场景", "名称应正确"
|
|
assert config.enabled is True, "应启用"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_scenario_config_update(self, db_session):
|
|
"""测试更新场景配置"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 先创建
|
|
await svc.upsert_scenario_config(
|
|
"update_scene",
|
|
{"name": "旧名称", "enabled": True},
|
|
operator="admin",
|
|
)
|
|
|
|
# 再更新
|
|
updated = await svc.upsert_scenario_config(
|
|
"update_scene",
|
|
{"name": "新名称"},
|
|
operator="admin",
|
|
)
|
|
|
|
assert updated.name == "新名称", "名称应更新"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_rule_versions(self, db_session):
|
|
"""测试列出规则版本"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 创建场景配置和版本
|
|
await svc.upsert_scenario_config(
|
|
"version_test",
|
|
{"name": "版本测试", "actions": []},
|
|
operator="admin",
|
|
)
|
|
|
|
versions = await svc.list_rule_versions("version_test")
|
|
|
|
assert len(versions) >= 1, "应返回规则版本"
|
|
|
|
|
|
# =============================================================================
|
|
# 测试指标汇总
|
|
# =============================================================================
|
|
class TestMetrics:
|
|
"""指标汇总测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_metrics(self, db_session):
|
|
"""测试指标计算"""
|
|
svc = AutoSessionService(db_session)
|
|
|
|
# 创建不同状态的会话
|
|
for status in ["resolved", "resolved", "handoff", "error"]:
|
|
session = AutoSession(
|
|
conversation_id=f"metric_conv_{status}",
|
|
employee_id="emp_metric",
|
|
title=f"指标测试_{status}",
|
|
status=status,
|
|
)
|
|
db_session.add(session)
|
|
await db_session.flush()
|
|
|
|
metrics = await svc.metrics()
|
|
|
|
assert "total_sessions" in metrics, "应包含总会话数"
|
|
assert "resolved_sessions" in metrics, "应包含已解决数"
|
|
assert "handoff_sessions" in metrics, "应包含转人工数"
|
|
assert "error_sessions" in metrics, "应包含错误数"
|
|
assert metrics["total_sessions"] == 4, "应有4个会话"
|
|
assert metrics["resolved_sessions"] == 2, "应有2个resolved"
|