# ============================================================================= # 企微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