# ============================================================================= # 阶段5 自动化 - 意图识别路由单元测试 # ============================================================================= # 测试范围:intent_router.py - 意图识别 + 场景路由(Dify+RAGFlow) # ============================================================================= import pytest from unittest.mock import AsyncMock, patch, MagicMock from app.services.automation.intent_router import IntentRouter class TestIntentRouter: """意图识别路由器测试用例""" @pytest.mark.asyncio async def test_detect_with_dify_client_success(self, db_session): """测试 detect Dify 客户端可用时正常识别""" # Arrange router = IntentRouter(db_session) mock_client = AsyncMock() mock_client.detect_intent = AsyncMock(return_value={ "scenario_key": "virus_dispose", "confidence": 0.95, "raw": {"intent": "virus_dispose"} }) # Mock build_dify_client 返回 mock 客户端 with patch("app.services.automation.intent_router.build_dify_client", return_value=mock_client): # Act result = await router.detect("电脑中毒了", "emp_001") # Assert assert result["scenario_key"] == "virus_dispose" assert result["confidence"] == 0.95 @pytest.mark.asyncio async def test_detect_fallback_when_dify_not_configured(self, db_session): """测试 detect Dify 未配置时走关键词兜底""" # Arrange router = IntentRouter(db_session) # Mock build_dify_client 返回 None(Dify 未配置) with patch("app.services.automation.intent_router.build_dify_client", return_value=None): # Act result = await router.detect("帮我重置密码", "emp_001") # Assert - 应该走关键词兜底 assert result["error"] == "dify_not_configured" assert "scenario_key" in result assert "confidence" in result @pytest.mark.asyncio async def test_detect_fallback_when_dify_exception(self, db_session): """测试 detect Dify 调用异常时走关键词兜底""" # Arrange router = IntentRouter(db_session) mock_client = AsyncMock() mock_client.detect_intent = AsyncMock(side_effect=Exception("Dify API error")) with patch("app.services.automation.intent_router.build_dify_client", return_value=mock_client): # Act result = await router.detect("我的电脑很卡", "emp_001") # Assert - 应该捕获异常并走兜底 assert "error" in result assert result["error"] != "dify_not_configured" # 是实际错误信息 assert "scenario_key" in result @pytest.mark.asyncio async def test_fallback_intent_password_reset(self): """测试关键词兜底 - 密码重置场景""" # Arrange - 直接测试 DifyClient 的静态方法 from app.integrations.dify import DifyClient # Act result = DifyClient._fallback_intent("我想重置密码") # Assert assert result["scenario_key"] == "password_reset" assert result["confidence"] == 0.6 # 兜底置信度 @pytest.mark.asyncio async def test_fallback_intent_virus_dispose(self): """测试关键词兜底 - 病毒查杀场景""" from app.integrations.dify import DifyClient # Act result = DifyClient._fallback_intent("电脑中病毒了") # Assert assert result["scenario_key"] == "virus_dispose" assert result["confidence"] == 0.5 @pytest.mark.asyncio async def test_fallback_intent_terminal_locate(self): """测试关键词兜底 - 终端定位场景""" from app.integrations.dify import DifyClient # Act result = DifyClient._fallback_intent("我的电脑在哪里") # Assert assert result["scenario_key"] == "terminal_locate" assert result["confidence"] == 0.5 @pytest.mark.asyncio async def test_fallback_intent_software_install(self): """测试关键词兜底 - 软件安装场景""" from app.integrations.dify import DifyClient # Act result = DifyClient._fallback_intent("帮我安装一个软件") # Assert assert result["scenario_key"] == "software_install" assert result["confidence"] == 0.5 @pytest.mark.asyncio async def test_fallback_intent_unknown(self): """测试关键词兜底 - 未知场景""" from app.integrations.dify import DifyClient # Act result = DifyClient._fallback_intent("今天天气真好") # Assert assert result["scenario_key"] is None assert result["confidence"] == 0.0