821 lines
28 KiB
Python
821 lines
28 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""代答排除模块回归测试 — T04
|
|||
|
|
|
|||
|
|
测试范围:
|
|||
|
|
Matcher单元测试（12个）: keyword(5) + regex(3) + intent(2) + category(2)
|
|||
|
|
ExclusionService测试（5个）: no_rules, priority_order, first_hit_stops,
|
|||
|
|
logs_hit, test_match_no_logging
|
|||
|
|
API测试（8个）: create, duplicate_name, get_detail, update, delete,
|
|||
|
|
toggle, test_match, get_stats
|
|||
|
|
ai_handler集成测试（3个）: exclusion_hit, exclusion_miss, exclusion_error
|
|||
|
|
|
|||
|
|
测试依赖: conftest.py 提供的 client / db_session fixtures
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import uuid
|
|||
|
|
from contextlib import ExitStack
|
|||
|
|
from datetime import datetime
|
|||
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from httpx import AsyncClient
|
|||
|
|
from sqlalchemy import select
|
|||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
|
|
|||
|
|
from app.models.exclusion_rule import ExclusionRule
|
|||
|
|
from app.models.exclusion_log import ExclusionLog
|
|||
|
|
from app.models.triage_session import TriageSession
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================================
|
|||
|
|
# 辅助函数
|
|||
|
|
# ============================================================================
|
|||
|
|
|
|||
|
|
async def _login_admin(client: AsyncClient, db_session: AsyncSession) -> str:
|
|||
|
|
"""创建 admin 角色用户并返回 Bearer token。"""
|
|||
|
|
from app.models.role import Role
|
|||
|
|
from app.models.user_role import UserRole
|
|||
|
|
|
|||
|
|
admin_id = f"test_admin_{uuid.uuid4().hex[:8]}"
|
|||
|
|
|
|||
|
|
# 确保 admin 角色存在
|
|||
|
|
stmt = select(Role).where(Role.name == "admin")
|
|||
|
|
result = await db_session.execute(stmt)
|
|||
|
|
admin_role = result.scalars().first()
|
|||
|
|
if not admin_role:
|
|||
|
|
admin_role = Role(
|
|||
|
|
name="admin", display_name="管理员",
|
|||
|
|
description="系统管理员", permissions=[],
|
|||
|
|
)
|
|||
|
|
db_session.add(admin_role)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 创建 UserRole 关联
|
|||
|
|
db_session.add(UserRole(
|
|||
|
|
employee_id=admin_id, role_id=admin_role.id,
|
|||
|
|
source="manual", assigned_by="test_fixture",
|
|||
|
|
))
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 登录
|
|||
|
|
resp = await client.post("/agents/login", json={
|
|||
|
|
"user_id": admin_id, "name": "测试管理员",
|
|||
|
|
})
|
|||
|
|
return resp.json()["data"]["token"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def _create_exclusion_rule(db_session: AsyncSession, **kwargs) -> ExclusionRule:
|
|||
|
|
"""在数据库中创建排除规则。
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
db_session: 数据库会话
|
|||
|
|
**kwargs: 覆盖默认字段值
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
ExclusionRule: 创建的规则对象
|
|||
|
|
"""
|
|||
|
|
defaults = {
|
|||
|
|
"rule_name": f"规则-{uuid.uuid4().hex[:8]}",
|
|||
|
|
"rule_description": "测试规则",
|
|||
|
|
"priority": "P2",
|
|||
|
|
"match_type": "keyword",
|
|||
|
|
"match_condition": "密码",
|
|||
|
|
"match_scope": [],
|
|||
|
|
"action_type": "transfer_human",
|
|||
|
|
"transfer_message": "已为您转接人工坐席",
|
|||
|
|
"status": "enabled",
|
|||
|
|
"hit_count": 0,
|
|||
|
|
"created_by": "test_admin",
|
|||
|
|
}
|
|||
|
|
defaults.update(kwargs)
|
|||
|
|
rule = ExclusionRule(**defaults)
|
|||
|
|
db_session.add(rule)
|
|||
|
|
await db_session.flush()
|
|||
|
|
return rule
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================================
|
|||
|
|
# Section A — Matcher 单元测试（12个）
|
|||
|
|
# ============================================================================
|
|||
|
|
|
|||
|
|
class TestKeywordMatcher:
|
|||
|
|
"""关键词匹配器测试。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_keyword_match_hit(self):
|
|||
|
|
"""关键词命中。"""
|
|||
|
|
from app.services.matchers.keyword_matcher import KeywordMatcher
|
|||
|
|
|
|||
|
|
matcher = KeywordMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="我的密码过期了怎么办",
|
|||
|
|
condition="密码过期,账号锁定",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "密码过期" in result.matched_detail
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_keyword_match_miss(self):
|
|||
|
|
"""关键词未命中。"""
|
|||
|
|
from app.services.matchers.keyword_matcher import KeywordMatcher
|
|||
|
|
|
|||
|
|
matcher = KeywordMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="今天天气真好",
|
|||
|
|
condition="密码过期,账号锁定",
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_keyword_match_case_insensitive(self):
|
|||
|
|
"""大小写不敏感匹配。"""
|
|||
|
|
from app.services.matchers.keyword_matcher import KeywordMatcher
|
|||
|
|
|
|||
|
|
matcher = KeywordMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="VPN connection failed",
|
|||
|
|
condition="vpn,password",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "vpn" in result.matched_detail.lower()
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_keyword_match_multiple_keywords(self):
|
|||
|
|
"""多关键词逗号分隔，任一命中即匹配。"""
|
|||
|
|
from app.services.matchers.keyword_matcher import KeywordMatcher
|
|||
|
|
|
|||
|
|
matcher = KeywordMatcher()
|
|||
|
|
# 第三个关键词命中
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="打印机卡纸了",
|
|||
|
|
condition="密码过期,账号锁定,打印机",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "打印机" in result.matched_detail
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_keyword_match_empty_message(self):
|
|||
|
|
"""空消息返回未命中。"""
|
|||
|
|
from app.services.matchers.keyword_matcher import KeywordMatcher
|
|||
|
|
|
|||
|
|
matcher = KeywordMatcher()
|
|||
|
|
result = await matcher.match(message="", condition="密码")
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
# 空条件也应返回未命中
|
|||
|
|
result = await matcher.match(message="密码过期", condition="")
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestRegexMatcher:
|
|||
|
|
"""正则匹配器测试。
|
|||
|
|
|
|||
|
|
注意: RegexMatcher 使用 signal.SIGALRM 做 ReDoS 超时保护，
|
|||
|
|
Windows 不支持 SIGALRM。测试通过 fixture patch signal 模块
|
|||
|
|
使正则匹配在 Windows 上正常工作。
|
|||
|
|
源码 regex_matcher.py 在 Windows 上存在兼容性问题
|
|||
|
|
（SIGALRM 不可用时直接返回未命中，应降级为无超时匹配）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
@pytest.fixture(autouse=True)
|
|||
|
|
def _patch_signal(self):
|
|||
|
|
"""Windows 兼容: patch signal.SIGALRM 使正则匹配器正常工作。"""
|
|||
|
|
import signal as sig
|
|||
|
|
if not hasattr(sig, "SIGALRM"):
|
|||
|
|
with ExitStack() as stack:
|
|||
|
|
stack.enter_context(patch.object(sig, "SIGALRM", 14, create=True))
|
|||
|
|
stack.enter_context(patch.object(sig, "ITIMER_REAL", 0, create=True))
|
|||
|
|
if not hasattr(sig, "setitimer"):
|
|||
|
|
stack.enter_context(
|
|||
|
|
patch.object(sig, "setitimer", create=True, return_value=None)
|
|||
|
|
)
|
|||
|
|
stack.enter_context(
|
|||
|
|
patch.object(sig, "signal", return_value=sig.SIG_DFL)
|
|||
|
|
)
|
|||
|
|
yield
|
|||
|
|
else:
|
|||
|
|
yield
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_regex_match_hit(self):
|
|||
|
|
"""正则命中。"""
|
|||
|
|
from app.services.matchers.regex_matcher import RegexMatcher
|
|||
|
|
|
|||
|
|
matcher = RegexMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="我的密码好像过期了",
|
|||
|
|
condition="密码.*过期",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "密码" in result.matched_detail
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_regex_match_miss(self):
|
|||
|
|
"""正则未命中。"""
|
|||
|
|
from app.services.matchers.regex_matcher import RegexMatcher
|
|||
|
|
|
|||
|
|
matcher = RegexMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="今天天气真好",
|
|||
|
|
condition="密码.*过期",
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_regex_match_invalid_pattern(self):
|
|||
|
|
"""无效正则返回未命中（不抛异常）。"""
|
|||
|
|
from app.services.matchers.regex_matcher import RegexMatcher
|
|||
|
|
|
|||
|
|
matcher = RegexMatcher()
|
|||
|
|
# 无效正则括号不匹配
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="测试消息",
|
|||
|
|
condition="[unclosed",
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestIntentMatcher:
|
|||
|
|
"""意图匹配器测试。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_intent_match_dify_unavailable(self):
|
|||
|
|
"""Dify 不可用时降级返回未命中。"""
|
|||
|
|
from app.services.matchers.intent_matcher import IntentMatcher
|
|||
|
|
|
|||
|
|
matcher = IntentMatcher()
|
|||
|
|
# Mock _recognize_intent 返回 None（Dify 不可用）
|
|||
|
|
with patch.object(matcher, "_recognize_intent", return_value=None):
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="我的密码忘了",
|
|||
|
|
condition="password_reset,account_unlock",
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_intent_match_hit(self):
|
|||
|
|
"""意图命中（mock Dify 返回匹配的意图）。"""
|
|||
|
|
from app.services.matchers.intent_matcher import IntentMatcher
|
|||
|
|
|
|||
|
|
matcher = IntentMatcher()
|
|||
|
|
# Mock _recognize_intent 返回 password_reset
|
|||
|
|
with patch.object(matcher, "_recognize_intent", return_value="password_reset"):
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="我的密码忘了，帮我重置一下",
|
|||
|
|
condition="password_reset,account_unlock",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "password_reset" in result.matched_detail
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestCategoryMatcher:
|
|||
|
|
"""分类匹配器测试。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_category_match_hit(self, db_session: AsyncSession):
|
|||
|
|
"""分类命中 — 分诊记录的 problem_category 在排除列表中。"""
|
|||
|
|
from app.services.matchers.category_matcher import CategoryMatcher
|
|||
|
|
|
|||
|
|
conv_id = f"conv-cat-{uuid.uuid4().hex[:8]}"
|
|||
|
|
# 创建分诊记录，problem_category = "Outlook"
|
|||
|
|
triage = TriageSession(
|
|||
|
|
conversation_id=conv_id,
|
|||
|
|
user_id="test_user",
|
|||
|
|
user_name="测试",
|
|||
|
|
request_title="测试",
|
|||
|
|
request_content="测试内容",
|
|||
|
|
source="wecom_h5",
|
|||
|
|
status="routed",
|
|||
|
|
urgency="low",
|
|||
|
|
problem_category="Outlook",
|
|||
|
|
)
|
|||
|
|
db_session.add(triage)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
matcher = CategoryMatcher()
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="测试消息",
|
|||
|
|
condition="Outlook,VPN,打印机",
|
|||
|
|
context={"conversation_id": conv_id, "db": db_session},
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert "Outlook" in result.matched_detail
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_category_match_no_triage(self, db_session: AsyncSession):
|
|||
|
|
"""无分诊记录返回未命中（软依赖）。"""
|
|||
|
|
from app.services.matchers.category_matcher import CategoryMatcher
|
|||
|
|
|
|||
|
|
matcher = CategoryMatcher()
|
|||
|
|
# 使用不存在的 conversation_id
|
|||
|
|
result = await matcher.match(
|
|||
|
|
message="测试消息",
|
|||
|
|
condition="Outlook,VPN",
|
|||
|
|
context={"conversation_id": "non-existent-conv", "db": db_session},
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================================
|
|||
|
|
# Section B — ExclusionService 测试（5个）
|
|||
|
|
# ============================================================================
|
|||
|
|
|
|||
|
|
class TestExclusionService:
|
|||
|
|
"""ExclusionService 责任链匹配引擎测试。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_check_exclusions_no_rules(self, db_session: AsyncSession):
|
|||
|
|
"""无规则时返回未命中。"""
|
|||
|
|
from app.services.exclusion_service import ExclusionService
|
|||
|
|
|
|||
|
|
service = ExclusionService()
|
|||
|
|
result = await service.check_exclusions(
|
|||
|
|
db=db_session,
|
|||
|
|
message="测试消息",
|
|||
|
|
conversation_id="conv-001",
|
|||
|
|
user_id="user-001",
|
|||
|
|
)
|
|||
|
|
assert result.matched is False
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_check_exclusions_priority_order(self, db_session: AsyncSession):
|
|||
|
|
"""优先级排序 P0 > P1 — P0 规则先匹配。"""
|
|||
|
|
from app.services.exclusion_service import ExclusionService
|
|||
|
|
|
|||
|
|
# 创建两条规则，P1 和 P0，消息同时包含两个关键词
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="P1规则-密码",
|
|||
|
|
priority="P1",
|
|||
|
|
match_condition="密码",
|
|||
|
|
)
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="P0规则-宕机",
|
|||
|
|
priority="P0",
|
|||
|
|
match_condition="宕机",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
service = ExclusionService()
|
|||
|
|
result = await service.check_exclusions(
|
|||
|
|
db=db_session,
|
|||
|
|
message="系统宕机了，密码也忘了",
|
|||
|
|
conversation_id="conv-002",
|
|||
|
|
user_id="user-002",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
# P0 规则应先匹配
|
|||
|
|
assert result.rule_name == "P0规则-宕机"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_check_exclusions_first_hit_stops(self, db_session: AsyncSession):
|
|||
|
|
"""命中即停止 — 只记录一条日志。"""
|
|||
|
|
from app.services.exclusion_service import ExclusionService
|
|||
|
|
|
|||
|
|
# 创建两条都能匹配的规则
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="规则A-密码",
|
|||
|
|
priority="P0",
|
|||
|
|
match_condition="密码",
|
|||
|
|
)
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="规则B-密码",
|
|||
|
|
priority="P1",
|
|||
|
|
match_condition="密码",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
service = ExclusionService()
|
|||
|
|
result = await service.check_exclusions(
|
|||
|
|
db=db_session,
|
|||
|
|
message="密码过期了",
|
|||
|
|
conversation_id="conv-003",
|
|||
|
|
user_id="user-003",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
# 只有 P0 规则应命中
|
|||
|
|
assert result.rule_name == "规则A-密码"
|
|||
|
|
|
|||
|
|
# 验证只创建了一条日志
|
|||
|
|
log_result = await db_session.execute(select(ExclusionLog))
|
|||
|
|
logs = log_result.scalars().all()
|
|||
|
|
assert len(logs) == 1
|
|||
|
|
assert logs[0].rule_name == "规则A-密码"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_check_exclusions_logs_hit(self, db_session: AsyncSession):
|
|||
|
|
"""命中记录日志 + 更新 hit_count。"""
|
|||
|
|
from app.services.exclusion_service import ExclusionService
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="日志测试规则",
|
|||
|
|
match_condition="密码过期",
|
|||
|
|
hit_count=0,
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
service = ExclusionService()
|
|||
|
|
result = await service.check_exclusions(
|
|||
|
|
db=db_session,
|
|||
|
|
message="我的密码过期了",
|
|||
|
|
conversation_id="conv-004",
|
|||
|
|
user_id="user-004",
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
assert result.rule_name == "日志测试规则"
|
|||
|
|
|
|||
|
|
# 验证 hit_count 已更新
|
|||
|
|
await db_session.refresh(rule)
|
|||
|
|
assert rule.hit_count == 1
|
|||
|
|
|
|||
|
|
# 验证日志已创建
|
|||
|
|
log_result = await db_session.execute(
|
|||
|
|
select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
|
|||
|
|
)
|
|||
|
|
logs = log_result.scalars().all()
|
|||
|
|
assert len(logs) == 1
|
|||
|
|
assert logs[0].action_type == "transfer_human"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_test_match_no_logging(self, db_session: AsyncSession):
|
|||
|
|
"""test_match 不记录日志、不更新 hit_count。"""
|
|||
|
|
from app.services.exclusion_service import ExclusionService
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="测试匹配规则",
|
|||
|
|
match_condition="密码过期",
|
|||
|
|
hit_count=0,
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
service = ExclusionService()
|
|||
|
|
result = await service.test_match(
|
|||
|
|
db=db_session,
|
|||
|
|
message="我的密码过期了",
|
|||
|
|
rule_id=rule.id,
|
|||
|
|
)
|
|||
|
|
assert result.matched is True
|
|||
|
|
|
|||
|
|
# 验证 hit_count 未更新
|
|||
|
|
await db_session.refresh(rule)
|
|||
|
|
assert rule.hit_count == 0
|
|||
|
|
|
|||
|
|
# 验证无日志创建
|
|||
|
|
log_result = await db_session.execute(
|
|||
|
|
select(ExclusionLog).where(ExclusionLog.rule_id == rule.id)
|
|||
|
|
)
|
|||
|
|
logs = log_result.scalars().all()
|
|||
|
|
assert len(logs) == 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================================
|
|||
|
|
# Section C — API 测试（8个）
|
|||
|
|
# ============================================================================
|
|||
|
|
|
|||
|
|
class TestExclusionAPI:
|
|||
|
|
"""代答排除管理 API 测试。"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_create_rule_success(self, client, db_session):
|
|||
|
|
"""新建规则成功。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
resp = await client.post(
|
|||
|
|
"/admin/exclusion-rules",
|
|||
|
|
json={
|
|||
|
|
"rule_name": "测试规则-新建",
|
|||
|
|
"rule_description": "测试描述",
|
|||
|
|
"priority": "P1",
|
|||
|
|
"match_type": "keyword",
|
|||
|
|
"match_condition": "密码,账号",
|
|||
|
|
"match_scope": [],
|
|||
|
|
"action_type": "transfer_human",
|
|||
|
|
"transfer_message": "已转人工",
|
|||
|
|
},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
rule = data["data"]
|
|||
|
|
assert rule["rule_name"] == "测试规则-新建"
|
|||
|
|
assert rule["status"] == "enabled"
|
|||
|
|
assert rule["hit_count"] == 0
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_create_rule_duplicate_name(self, client, db_session):
|
|||
|
|
"""规则名重复返回 400。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
# 先创建一条规则
|
|||
|
|
await _create_exclusion_rule(db_session, rule_name="重复规则名")
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 再用同名创建
|
|||
|
|
resp = await client.post(
|
|||
|
|
"/admin/exclusion-rules",
|
|||
|
|
json={
|
|||
|
|
"rule_name": "重复规则名",
|
|||
|
|
"priority": "P2",
|
|||
|
|
"match_type": "keyword",
|
|||
|
|
"match_condition": "测试",
|
|||
|
|
"action_type": "transfer_human",
|
|||
|
|
},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 400
|
|||
|
|
assert "已存在" in data["message"]
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_get_rule_detail(self, client, db_session):
|
|||
|
|
"""规则详情。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="详情测试规则",
|
|||
|
|
match_type="keyword",
|
|||
|
|
match_condition="密码",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
resp = await client.get(
|
|||
|
|
f"/admin/exclusion-rules/{rule.id}",
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["id"] == rule.id
|
|||
|
|
assert data["data"]["rule_name"] == "详情测试规则"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_update_rule(self, client, db_session):
|
|||
|
|
"""编辑规则。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="编辑前规则",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
resp = await client.put(
|
|||
|
|
f"/admin/exclusion-rules/{rule.id}",
|
|||
|
|
json={
|
|||
|
|
"rule_name": "编辑后规则",
|
|||
|
|
"match_condition": "VPN,网络",
|
|||
|
|
},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["rule_name"] == "编辑后规则"
|
|||
|
|
assert data["data"]["match_condition"] == "VPN,网络"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_delete_rule(self, client, db_session):
|
|||
|
|
"""删除规则。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="待删除规则",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
rule_id = rule.id
|
|||
|
|
|
|||
|
|
resp = await client.delete(
|
|||
|
|
f"/admin/exclusion-rules/{rule_id}",
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
|
|||
|
|
# 验证已删除
|
|||
|
|
resp = await client.get(
|
|||
|
|
f"/admin/exclusion-rules/{rule_id}",
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
assert resp.json()["code"] == 404
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_toggle_rule(self, client, db_session):
|
|||
|
|
"""启用/停用规则。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
rule = await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="切换状态规则",
|
|||
|
|
status="enabled",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 停用
|
|||
|
|
resp = await client.post(
|
|||
|
|
f"/admin/exclusion-rules/{rule.id}/toggle",
|
|||
|
|
json={"status": "disabled"},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["status"] == "disabled"
|
|||
|
|
|
|||
|
|
# 再启用
|
|||
|
|
resp = await client.post(
|
|||
|
|
f"/admin/exclusion-rules/{rule.id}/toggle",
|
|||
|
|
json={"status": "enabled"},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
assert resp.json()["data"]["status"] == "enabled"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_test_match_api(self, client, db_session):
|
|||
|
|
"""测试匹配 API。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="API测试匹配规则",
|
|||
|
|
match_type="keyword",
|
|||
|
|
match_condition="密码过期",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
resp = await client.post(
|
|||
|
|
"/admin/exclusion-rules/test",
|
|||
|
|
json={"message": "我的密码过期了"},
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
assert data["data"]["matched"] is True
|
|||
|
|
assert data["data"]["rule_name"] == "API测试匹配规则"
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_get_stats_api(self, client, db_session):
|
|||
|
|
"""统计概要 API。"""
|
|||
|
|
token = await _login_admin(client, db_session)
|
|||
|
|
|
|||
|
|
# 创建规则
|
|||
|
|
await _create_exclusion_rule(db_session, status="enabled")
|
|||
|
|
await _create_exclusion_rule(db_session, status="disabled")
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
resp = await client.get(
|
|||
|
|
"/admin/exclusion-rules/stats",
|
|||
|
|
headers={"Authorization": f"Bearer {token}"},
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
data = resp.json()
|
|||
|
|
assert data["code"] == 0
|
|||
|
|
stats = data["data"]
|
|||
|
|
assert "enabled_count" in stats
|
|||
|
|
assert "disabled_count" in stats
|
|||
|
|
assert "monthly_hits" in stats
|
|||
|
|
assert "monthly_transfers" in stats
|
|||
|
|
assert stats["enabled_count"] == 1
|
|||
|
|
assert stats["disabled_count"] == 1
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ============================================================================
|
|||
|
|
# Section D — ai_handler 集成测试（3个）
|
|||
|
|
# ============================================================================
|
|||
|
|
|
|||
|
|
class TestAIHandlerExclusion:
|
|||
|
|
"""AI 回复处理器与代答排除的集成测试。
|
|||
|
|
|
|||
|
|
验证 ai_handler.handle_message 在 AI 回复前检查排除规则：
|
|||
|
|
1. 命中排除规则 → 拦截 AI 回复，返回排除结果
|
|||
|
|
2. 未命中 → 正常调用 AI
|
|||
|
|
3. 排除检查异常 → 降级继续 AI 回复（不阻断主流程）
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_ai_handler_exclusion_hit(self, db_session: AsyncSession):
|
|||
|
|
"""命中排除规则 — AI 回复前拦截，不调用 AI 服务。"""
|
|||
|
|
from app.services.ai_handler import AIHandler
|
|||
|
|
|
|||
|
|
# 创建排除规则（关键词 "密码" → transfer_human）
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="密码排除规则",
|
|||
|
|
match_type="keyword",
|
|||
|
|
match_condition="密码",
|
|||
|
|
action_type="transfer_human",
|
|||
|
|
transfer_message="此问题需转人工处理",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 创建 AIHandler with mock AIService
|
|||
|
|
mock_ai_service = AsyncMock()
|
|||
|
|
handler = AIHandler(ai_service=mock_ai_service)
|
|||
|
|
|
|||
|
|
result = await handler.handle_message(
|
|||
|
|
content="我的密码过期了怎么办",
|
|||
|
|
conversation_id="conv-ai-001",
|
|||
|
|
user_id="user-001",
|
|||
|
|
db=db_session,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 验证：命中排除规则，返回 excluded 类型
|
|||
|
|
assert result.reply_type == "excluded"
|
|||
|
|
assert result.should_transfer is True
|
|||
|
|
assert result.should_count is False
|
|||
|
|
# AI 服务不应被调用
|
|||
|
|
mock_ai_service.get_reply.assert_not_called()
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_ai_handler_exclusion_miss(self, db_session: AsyncSession):
|
|||
|
|
"""未命中排除规则 — 正常调用 AI 服务。"""
|
|||
|
|
from app.services.ai_handler import AIHandler
|
|||
|
|
|
|||
|
|
# 创建排除规则（关键词 "密码"）
|
|||
|
|
await _create_exclusion_rule(
|
|||
|
|
db_session,
|
|||
|
|
rule_name="密码排除规则",
|
|||
|
|
match_type="keyword",
|
|||
|
|
match_condition="密码",
|
|||
|
|
)
|
|||
|
|
await db_session.flush()
|
|||
|
|
|
|||
|
|
# 创建 AIHandler with mock AIService
|
|||
|
|
mock_ai_service = AsyncMock()
|
|||
|
|
mock_ai_service.get_reply.return_value = {
|
|||
|
|
"hit": True,
|
|||
|
|
"content": "建议您重启电脑试试",
|
|||
|
|
"conversation_id": "dify-conv-001",
|
|||
|
|
}
|
|||
|
|
handler = AIHandler(ai_service=mock_ai_service)
|
|||
|
|
|
|||
|
|
# 消息不含 "密码" → 不命中排除规则 → 正常 AI 回复
|
|||
|
|
result = await handler.handle_message(
|
|||
|
|
content="打印机怎么连接",
|
|||
|
|
conversation_id="conv-ai-002",
|
|||
|
|
user_id="user-002",
|
|||
|
|
db=db_session,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 验证：正常 AI 回复
|
|||
|
|
assert result.reply_type == "ai_hit"
|
|||
|
|
assert result.should_count is True
|
|||
|
|
assert "重启电脑" in result.content
|
|||
|
|
# AI 服务应被调用
|
|||
|
|
mock_ai_service.get_reply.assert_called_once()
|
|||
|
|
|
|||
|
|
@pytest.mark.asyncio
|
|||
|
|
async def test_ai_handler_exclusion_error(self, db_session: AsyncSession):
|
|||
|
|
"""排除检查异常不阻断主流程 — 降级继续 AI 回复。"""
|
|||
|
|
from app.services.ai_handler import AIHandler
|
|||
|
|
|
|||
|
|
# 创建 AIHandler with mock AIService
|
|||
|
|
mock_ai_service = AsyncMock()
|
|||
|
|
mock_ai_service.get_reply.return_value = {
|
|||
|
|
"hit": True,
|
|||
|
|
"content": "AI降级回复",
|
|||
|
|
"conversation_id": "dify-conv-002",
|
|||
|
|
}
|
|||
|
|
handler = AIHandler(ai_service=mock_ai_service)
|
|||
|
|
|
|||
|
|
# Patch check_exclusions 抛出异常
|
|||
|
|
with patch(
|
|||
|
|
"app.services.exclusion_service.get_exclusion_service"
|
|||
|
|
) as mock_get_service:
|
|||
|
|
mock_service = AsyncMock()
|
|||
|
|
mock_service.check_exclusions.side_effect = Exception("DB connection error")
|
|||
|
|
mock_get_service.return_value = mock_service
|
|||
|
|
|
|||
|
|
result = await handler.handle_message(
|
|||
|
|
content="打印机问题",
|
|||
|
|
conversation_id="conv-ai-003",
|
|||
|
|
user_id="user-003",
|
|||
|
|
db=db_session,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# 验证：排除检查异常后降级继续 AI 回复
|
|||
|
|
assert result.reply_type == "ai_hit"
|
|||
|
|
assert result.should_count is True
|
|||
|
|
# AI 服务应被调用（降级不阻断）
|
|||
|
|
mock_ai_service.get_reply.assert_called_once()
|