338 lines
11 KiB
Python
338 lines
11 KiB
Python
"""
|
|
Tests for Mid Platform services.
|
|
[AC-IDMP-05/07/08/09/18/20] 中台服务测试
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime
|
|
|
|
from app.services.mid import (
|
|
HighRiskHandler,
|
|
HighRiskMatchResult,
|
|
PolicyRouter,
|
|
PolicyRouteResult,
|
|
TraceLogger,
|
|
MetricsCollector,
|
|
MetricsRecord,
|
|
SessionModeService,
|
|
DEFAULT_HIGH_RISK_SCENARIOS,
|
|
)
|
|
from app.models.entities import (
|
|
HighRiskPolicy,
|
|
HighRiskScenarioType,
|
|
SessionModeRecord,
|
|
MidAuditLog,
|
|
)
|
|
from app.models.mid import Mode, SessionMode
|
|
|
|
|
|
class TestHighRiskHandler:
|
|
"""[AC-IDMP-05/20] 高风险场景处理器测试"""
|
|
|
|
@pytest.fixture
|
|
def mock_session(self):
|
|
session = AsyncMock()
|
|
return session
|
|
|
|
@pytest.fixture
|
|
def handler(self, mock_session):
|
|
return HighRiskHandler(mock_session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_active_scenario_set_returns_default_when_empty(self, handler, mock_session):
|
|
"""[AC-IDMP-20] 空集保护：数据库无配置时返回默认最小集"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
scenarios = await handler.get_active_scenario_set("test_tenant")
|
|
|
|
assert scenarios == DEFAULT_HIGH_RISK_SCENARIOS
|
|
assert "refund" in scenarios
|
|
assert "complaint_escalation" in scenarios
|
|
assert "privacy_sensitive_promise" in scenarios
|
|
assert "transfer" in scenarios
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_refund_keyword(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 检测退款场景"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "我要退款")
|
|
|
|
assert result.matched is True
|
|
assert result.scenario == HighRiskScenarioType.REFUND.value
|
|
assert result.handler_mode == "micro_flow"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_transfer_keyword(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 检测转人工场景"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "转人工")
|
|
|
|
assert result.matched is True
|
|
assert result.scenario == HighRiskScenarioType.TRANSFER.value
|
|
assert result.handler_mode == "transfer"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_complaint_keyword(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 检测投诉升级场景"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "我要投诉你们")
|
|
|
|
assert result.matched is True
|
|
assert result.scenario == HighRiskScenarioType.COMPLAINT_ESCALATION.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_privacy_keyword(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 检测隐私敏感承诺场景"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "你能保证我的隐私安全吗")
|
|
|
|
assert result.matched is True
|
|
assert result.scenario == HighRiskScenarioType.PRIVACY_SENSITIVE_PROMISE.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_no_match(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 正常消息不触发高风险"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = []
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "你好，请问有什么可以帮助我的")
|
|
|
|
assert result.matched is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_detect_high_risk_with_policy(self, handler, mock_session):
|
|
"""[AC-IDMP-05] 使用数据库策略检测高风险"""
|
|
policy = HighRiskPolicy(
|
|
tenant_id="test_tenant",
|
|
scenario=HighRiskScenarioType.REFUND.value,
|
|
handler_mode="micro_flow",
|
|
keywords=["退货退款"],
|
|
priority=100,
|
|
is_enabled=True,
|
|
)
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.all.return_value = [policy]
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await handler.detect_high_risk("test_tenant", "我要退货退款")
|
|
|
|
assert result.matched is True
|
|
assert result.scenario == HighRiskScenarioType.REFUND.value
|
|
assert result.matched_keyword == "退货退款"
|
|
|
|
|
|
class TestPolicyRouter:
|
|
"""[AC-IDMP-05] 策略路由器测试"""
|
|
|
|
@pytest.fixture
|
|
def mock_high_risk_handler(self):
|
|
return AsyncMock(spec=HighRiskHandler)
|
|
|
|
@pytest.fixture
|
|
def router(self, mock_high_risk_handler):
|
|
return PolicyRouter(mock_high_risk_handler)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_human_mode_returns_transfer(self, router, mock_high_risk_handler):
|
|
"""[AC-IDMP-05] 人工模式返回 transfer"""
|
|
mock_high_risk_handler.detect_high_risk.return_value = HighRiskMatchResult(matched=False)
|
|
|
|
result = await router.route(
|
|
tenant_id="test_tenant",
|
|
user_message="你好",
|
|
session_mode="HUMAN_ACTIVE",
|
|
)
|
|
|
|
assert result.mode == "transfer"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_high_risk_returns_micro_flow(self, router, mock_high_risk_handler):
|
|
"""[AC-IDMP-05] 高风险场景返回 micro_flow 或 transfer"""
|
|
mock_high_risk_handler.detect_high_risk.return_value = HighRiskMatchResult(
|
|
matched=True,
|
|
scenario="refund",
|
|
handler_mode="micro_flow",
|
|
)
|
|
|
|
result = await router.route(
|
|
tenant_id="test_tenant",
|
|
user_message="我要退款",
|
|
session_mode="BOT_ACTIVE",
|
|
)
|
|
|
|
assert result.mode == "micro_flow"
|
|
assert result.high_risk_scenario == "refund"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_low_confidence_returns_fixed(self, router, mock_high_risk_handler):
|
|
"""[AC-IDMP-05] 低置信度返回 fixed"""
|
|
mock_high_risk_handler.detect_high_risk.return_value = HighRiskMatchResult(matched=False)
|
|
|
|
result = await router.route(
|
|
tenant_id="test_tenant",
|
|
user_message="你好",
|
|
session_mode="BOT_ACTIVE",
|
|
confidence=0.2,
|
|
)
|
|
|
|
assert result.mode == "fixed"
|
|
assert result.fallback_reason_code == "low_confidence"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_route_normal_returns_agent(self, router, mock_high_risk_handler):
|
|
"""[AC-IDMP-05] 正常场景返回 agent"""
|
|
mock_high_risk_handler.detect_high_risk.return_value = HighRiskMatchResult(matched=False)
|
|
|
|
result = await router.route(
|
|
tenant_id="test_tenant",
|
|
user_message="你好",
|
|
session_mode="BOT_ACTIVE",
|
|
confidence=0.8,
|
|
)
|
|
|
|
assert result.mode == "agent"
|
|
|
|
|
|
class TestMetricsCollector:
|
|
"""[AC-IDMP-18] 指标采集器测试"""
|
|
|
|
@pytest.fixture
|
|
def collector(self):
|
|
return MetricsCollector()
|
|
|
|
def test_record_and_get_metrics(self, collector):
|
|
"""[AC-IDMP-18] 记录并获取指标"""
|
|
record1 = MetricsRecord(
|
|
tenant_id="tenant1",
|
|
session_id="session1",
|
|
request_id="req1",
|
|
task_completed=True,
|
|
slots_filled=3,
|
|
slots_total=5,
|
|
was_transferred=False,
|
|
had_recall=True,
|
|
latency_ms=100,
|
|
)
|
|
|
|
record2 = MetricsRecord(
|
|
tenant_id="tenant1",
|
|
session_id="session2",
|
|
request_id="req2",
|
|
task_completed=False,
|
|
slots_filled=2,
|
|
slots_total=5,
|
|
was_transferred=True,
|
|
had_recall=False,
|
|
latency_ms=200,
|
|
)
|
|
|
|
collector.record(record1)
|
|
collector.record(record2)
|
|
|
|
metrics = collector.get_metrics_snapshot("tenant1")
|
|
|
|
assert metrics["task_completion_rate"] == 0.5
|
|
assert metrics["slot_completion_rate"] == 0.5
|
|
assert metrics["wrong_transfer_rate"] == 0.5
|
|
assert metrics["no_recall_rate"] == 0.5
|
|
assert metrics["avg_latency_ms"] == 150.0
|
|
|
|
def test_get_metrics_empty_tenant(self, collector):
|
|
"""[AC-IDMP-18] 空租户返回零值指标"""
|
|
metrics = collector.get_metrics_snapshot("unknown_tenant")
|
|
|
|
assert metrics["task_completion_rate"] == 0.0
|
|
assert metrics["slot_completion_rate"] == 0.0
|
|
assert metrics["wrong_transfer_rate"] == 0.0
|
|
assert metrics["no_recall_rate"] == 0.0
|
|
assert metrics["avg_latency_ms"] == 0.0
|
|
|
|
|
|
class TestSessionModeService:
|
|
"""[AC-IDMP-09] 会话模式服务测试"""
|
|
|
|
@pytest.fixture
|
|
def mock_session(self):
|
|
return AsyncMock()
|
|
|
|
@pytest.fixture
|
|
def service(self, mock_session):
|
|
return SessionModeService(mock_session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_mode_returns_default(self, service, mock_session):
|
|
"""[AC-IDMP-09] 获取默认模式"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
mode = await service.get_mode("tenant1", "session1")
|
|
|
|
assert mode == "BOT_ACTIVE"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_switch_mode_creates_new(self, service, mock_session):
|
|
"""[AC-IDMP-09] 切换模式创建新记录"""
|
|
mock_result = MagicMock()
|
|
mock_result.scalar_one_or_none.return_value = None
|
|
mock_session.execute.return_value = mock_result
|
|
|
|
result = await service.switch_mode(
|
|
tenant_id="tenant1",
|
|
session_id="session1",
|
|
mode="HUMAN_ACTIVE",
|
|
reason="user_request",
|
|
)
|
|
|
|
assert result.mode == "HUMAN_ACTIVE"
|
|
assert result.reason == "user_request"
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
|
|
class TestTraceLogger:
|
|
"""[AC-IDMP-07] Trace 日志服务测试"""
|
|
|
|
@pytest.fixture
|
|
def mock_session(self):
|
|
return AsyncMock()
|
|
|
|
@pytest.fixture
|
|
def logger(self, mock_session):
|
|
return TraceLogger(mock_session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_creates_audit_record(self, logger, mock_session):
|
|
"""[AC-IDMP-07] 记录审计日志"""
|
|
result = await logger.log(
|
|
tenant_id="tenant1",
|
|
session_id="session1",
|
|
request_id="req1",
|
|
generation_id="gen1",
|
|
mode="agent",
|
|
intent="greeting",
|
|
tool_calls=[{"tool": "search", "duration_ms": 100}],
|
|
guardrail_triggered=False,
|
|
latency_ms=500,
|
|
)
|
|
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|