208 lines
7.6 KiB
Python
208 lines
7.6 KiB
Python
|
# =============================================================================
|
||
|
|
# 企微IT智能服务台 — 快速规则单元测试
|
||
|
|
# =============================================================================
|
||
|
|
# 用途：测试 QuickRuleService 单元逻辑
|
||
|
|
# 日期：2026-07-27
|
||
|
|
# 运行：pytest tests/test_quick_rules.py -v
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from unittest.mock import AsyncMock, MagicMock
|
||
|
|
|
||
|
|
from app.models.quick_rule import QuickRule
|
||
|
|
from app.services.quick_rule_service import QuickRuleService
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuickRuleService:
|
||
|
|
"""QuickRuleService 单元测试"""
|
||
|
|
|
||
|
|
@pytest.fixture(autouse=True)
|
||
|
|
def reset_cache(self):
|
||
|
|
"""每个测试前重置缓存"""
|
||
|
|
QuickRuleService._cache = {
|
||
|
|
'greeting': [],
|
||
|
|
'routing_prefilter': {},
|
||
|
|
'routing_target': {},
|
||
|
|
}
|
||
|
|
QuickRuleService._initialized = False
|
||
|
|
yield
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_load_all_with_greeting_rules(self):
|
||
|
|
"""测试加载打招呼关键词"""
|
||
|
|
mock_db = AsyncMock()
|
||
|
|
mock_rules = [
|
||
|
|
MagicMock(spec=QuickRule, rule_type='greeting', keyword='你好', category=None, extra_data=None, is_active=True),
|
||
|
|
MagicMock(spec=QuickRule, rule_type='greeting', keyword='您好', category=None, extra_data=None, is_active=True),
|
||
|
|
]
|
||
|
|
mock_result = MagicMock()
|
||
|
|
mock_result.scalars.return_value.all.return_value = mock_rules
|
||
|
|
mock_db.execute.return_value = mock_result
|
||
|
|
|
||
|
|
await QuickRuleService.load_all(mock_db)
|
||
|
|
|
||
|
|
keywords = QuickRuleService.get_greeting_keywords()
|
||
|
|
assert '你好' in keywords
|
||
|
|
assert '您好' in keywords
|
||
|
|
assert len(keywords) == 2
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_load_all_with_routing_keywords(self):
|
||
|
|
"""测试加载路由关键词"""
|
||
|
|
mock_db = AsyncMock()
|
||
|
|
mock_rules = [
|
||
|
|
MagicMock(spec=QuickRule, rule_type='routing_prefilter', keyword='打印机', category='IT服务', extra_data=None, is_active=True),
|
||
|
|
MagicMock(spec=QuickRule, rule_type='routing_prefilter', keyword='报销', category='财务', extra_data=None, is_active=True),
|
||
|
|
]
|
||
|
|
mock_result = MagicMock()
|
||
|
|
mock_result.scalars.return_value.all.return_value = mock_rules
|
||
|
|
mock_db.execute.return_value = mock_result
|
||
|
|
|
||
|
|
await QuickRuleService.load_all(mock_db)
|
||
|
|
|
||
|
|
keywords = QuickRuleService.get_routing_keywords()
|
||
|
|
assert keywords['打印机'] == 'IT服务'
|
||
|
|
assert keywords['报销'] == '财务'
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_load_all_with_routing_targets(self):
|
||
|
|
"""测试加载路由目标"""
|
||
|
|
mock_db = AsyncMock()
|
||
|
|
mock_rules = [
|
||
|
|
MagicMock(
|
||
|
|
spec=QuickRule,
|
||
|
|
rule_type='routing_target',
|
||
|
|
keyword='机票酒店前台',
|
||
|
|
category='行政',
|
||
|
|
extra_data={'service_name': '机票酒店前台', 'url': 'https://example.com'},
|
||
|
|
is_active=True,
|
||
|
|
),
|
||
|
|
]
|
||
|
|
mock_result = MagicMock()
|
||
|
|
mock_result.scalars.return_value.all.return_value = mock_rules
|
||
|
|
mock_db.execute.return_value = mock_result
|
||
|
|
|
||
|
|
await QuickRuleService.load_all(mock_db)
|
||
|
|
|
||
|
|
targets = QuickRuleService.get_routing_targets()
|
||
|
|
assert '行政' in targets
|
||
|
|
assert targets['行政']['service_name'] == '机票酒店前台'
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_load_all_handles_exception(self):
|
||
|
|
"""测试加载异常处理"""
|
||
|
|
mock_db = AsyncMock()
|
||
|
|
mock_db.execute.side_effect = Exception("DB error")
|
||
|
|
|
||
|
|
# 不应抛出异常
|
||
|
|
await QuickRuleService.load_all(mock_db)
|
||
|
|
|
||
|
|
# 缓存应保持空
|
||
|
|
assert QuickRuleService.get_greeting_keywords() == []
|
||
|
|
|
||
|
|
def test_check_greeting_true(self):
|
||
|
|
"""测试 greeting 检测为 true"""
|
||
|
|
QuickRuleService._cache['greeting'] = ['你好', '您好']
|
||
|
|
assert QuickRuleService.check_greeting('你好啊') is True
|
||
|
|
assert QuickRuleService.check_greeting('您好请问') is True
|
||
|
|
|
||
|
|
def test_check_greeting_false(self):
|
||
|
|
"""测试 greeting 检测为 false"""
|
||
|
|
QuickRuleService._cache['greeting'] = ['你好', '您好']
|
||
|
|
assert QuickRuleService.check_greeting('打印机坏了') is False
|
||
|
|
|
||
|
|
def test_check_greeting_empty_text(self):
|
||
|
|
"""测试空文本"""
|
||
|
|
assert QuickRuleService.check_greeting('') is False
|
||
|
|
assert QuickRuleService.check_greeting(None) is False
|
||
|
|
|
||
|
|
def test_check_routing_keyword(self):
|
||
|
|
"""测试路由关键词检测"""
|
||
|
|
QuickRuleService._cache['routing_prefilter'] = {
|
||
|
|
'打印机': 'IT服务',
|
||
|
|
'报销': '财务',
|
||
|
|
}
|
||
|
|
assert QuickRuleService.check_routing_keyword('打印机连不上') is True
|
||
|
|
assert QuickRuleService.check_routing_keyword('电脑坏了') is False
|
||
|
|
assert QuickRuleService.check_routing_keyword('报销咨询') is True
|
||
|
|
|
||
|
|
def test_get_routing_target_found(self):
|
||
|
|
"""测试获取存在的路由目标"""
|
||
|
|
QuickRuleService._cache['routing_target'] = {
|
||
|
|
'行政': {'service_name': '行政窗口', 'url': 'https://example.com'},
|
||
|
|
}
|
||
|
|
target = QuickRuleService.get_routing_target('行政')
|
||
|
|
assert target is not None
|
||
|
|
assert target['service_name'] == '行政窗口'
|
||
|
|
|
||
|
|
def test_get_routing_target_not_found(self):
|
||
|
|
"""测试获取不存在的路由目标"""
|
||
|
|
QuickRuleService._cache['routing_target'] = {
|
||
|
|
'行政': {'service_name': '行政窗口', 'url': 'https://example.com'},
|
||
|
|
}
|
||
|
|
target = QuickRuleService.get_routing_target('不存在的')
|
||
|
|
assert target is None
|
||
|
|
|
||
|
|
def test_get_categories(self):
|
||
|
|
"""测试获取所有业务分类"""
|
||
|
|
QuickRuleService._cache['routing_target'] = {
|
||
|
|
'行政': {},
|
||
|
|
'财务': {},
|
||
|
|
'人力资源': {},
|
||
|
|
}
|
||
|
|
categories = QuickRuleService.get_categories()
|
||
|
|
assert set(categories) == {'行政', '财务', '人力资源'}
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_refresh_reloads_cache(self):
|
||
|
|
"""测试 refresh 重新加载缓存"""
|
||
|
|
mock_db = AsyncMock()
|
||
|
|
mock_rules = [
|
||
|
|
MagicMock(spec=QuickRule, rule_type='greeting', keyword='刷新测试', category=None, extra_data=None, is_active=True),
|
||
|
|
]
|
||
|
|
mock_result = MagicMock()
|
||
|
|
mock_result.scalars.return_value.all.return_value = mock_rules
|
||
|
|
mock_db.execute.return_value = mock_result
|
||
|
|
|
||
|
|
await QuickRuleService.refresh(mock_db)
|
||
|
|
|
||
|
|
keywords = QuickRuleService.get_greeting_keywords()
|
||
|
|
assert '刷新测试' in keywords
|
||
|
|
|
||
|
|
|
||
|
|
class TestQuickRuleDataStructures:
|
||
|
|
"""QuickRule 数据结构测试"""
|
||
|
|
|
||
|
|
def test_quick_rule_model_attributes(self):
|
||
|
|
"""测试 QuickRule 模型属性"""
|
||
|
|
rule = QuickRule(
|
||
|
|
rule_type='greeting',
|
||
|
|
keyword='你好',
|
||
|
|
priority=10,
|
||
|
|
is_active=True,
|
||
|
|
)
|
||
|
|
assert rule.rule_type == 'greeting'
|
||
|
|
assert rule.keyword == '你好'
|
||
|
|
assert rule.priority == 10
|
||
|
|
assert rule.is_active is True
|
||
|
|
|
||
|
|
def test_to_dict_conversion(self):
|
||
|
|
"""测试 to_dict 转换"""
|
||
|
|
from datetime import datetime
|
||
|
|
rule = QuickRule(
|
||
|
|
id=1,
|
||
|
|
rule_type='greeting',
|
||
|
|
keyword='你好',
|
||
|
|
priority=10,
|
||
|
|
is_active=True,
|
||
|
|
)
|
||
|
|
rule.created_at = datetime.now()
|
||
|
|
|
||
|
|
d = rule.to_dict()
|
||
|
|
assert d['id'] == 1
|
||
|
|
assert d['rule_type'] == 'greeting'
|
||
|
|
assert d['keyword'] == '你好'
|
||
|
|
assert d['priority'] == 10
|
||
|
|
assert d['is_active'] is True
|
||
|
|
assert 'created_at' in d
|