285 lines
11 KiB
Python
285 lines
11 KiB
Python
|
# =============================================================================
|
||
|
|
# 企微IT智能服务台 — 阶段5 自动化闭环 服务单元测试（简化版）
|
||
|
|
# =============================================================================
|
||
|
|
# 说明：简化版测试，直接导入服务类进行测试，避免 conftest 导入问题
|
||
|
|
# 创建日期: 2026-07-01
|
||
|
|
# =============================================================================
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
|
||
|
|
# 设置测试环境变量（在导入 app 之前）
|
||
|
|
os.environ.setdefault("DEV_MODE", "true")
|
||
|
|
os.environ.setdefault("WECOM_SCO_CALLBACK_BASE", "https://test.example.com")
|
||
|
|
|
||
|
|
# 测试意图路由 - 这些不需要数据库
|
||
|
|
class TestIntentRouterSimple:
|
||
|
|
"""简化版意图识别测试"""
|
||
|
|
|
||
|
|
def test_fallback_intent_virus_keywords(self):
|
||
|
|
"""测试病毒关键词识别"""
|
||
|
|
from app.services.automation.intent_router import IntentRouter
|
||
|
|
|
||
|
|
router = IntentRouter()
|
||
|
|
|
||
|
|
# 模拟 DifyClient._fallback_intent 方法的行为
|
||
|
|
from app.integrations.dify import DifyClient
|
||
|
|
|
||
|
|
result = DifyClient._fallback_intent("电脑中病毒了")
|
||
|
|
|
||
|
|
assert result is not None
|
||
|
|
assert "scenario_key" in result
|
||
|
|
|
||
|
|
def test_fallback_intent_password_keywords(self):
|
||
|
|
"""测试密码重置关键词识别"""
|
||
|
|
from app.integrations.dify import DifyClient
|
||
|
|
|
||
|
|
result = DifyClient._fallback_intent("我忘记密码了")
|
||
|
|
|
||
|
|
assert result.get("scenario_key") == "password_reset"
|
||
|
|
|
||
|
|
def test_fallback_intent_software_keywords(self):
|
||
|
|
"""测试软件安装关键词识别"""
|
||
|
|
from app.integrations.dify import DifyClient
|
||
|
|
|
||
|
|
result = DifyClient._fallback_intent("帮我安装一个软件")
|
||
|
|
|
||
|
|
assert result.get("scenario_key") == "software_install"
|
||
|
|
|
||
|
|
def test_fallback_intent_terminal_keywords(self):
|
||
|
|
"""测试终端定位关键词识别"""
|
||
|
|
from app.integrations.dify import DifyClient
|
||
|
|
|
||
|
|
result = DifyClient._fallback_intent("我的电脑在哪里")
|
||
|
|
|
||
|
|
assert result.get("scenario_key") == "terminal_locate"
|
||
|
|
|
||
|
|
def test_fallback_intent_unknown(self):
|
||
|
|
"""测试未知意图"""
|
||
|
|
from app.integrations.dify import DifyClient
|
||
|
|
|
||
|
|
result = DifyClient._fallback_intent("今天天气不错")
|
||
|
|
|
||
|
|
assert result.get("scenario_key") is None
|
||
|
|
assert result.get("confidence") == 0.0
|
||
|
|
|
||
|
|
|
||
|
|
class TestConstants:
|
||
|
|
"""测试常量定义"""
|
||
|
|
|
||
|
|
def test_error_codes(self):
|
||
|
|
"""测试错误码定义"""
|
||
|
|
from app.constants import AutomationErrorCode
|
||
|
|
|
||
|
|
assert AutomationErrorCode.SCENARIO_NOT_FOUND == 4001
|
||
|
|
assert AutomationErrorCode.INTENT_FAILED == 4002
|
||
|
|
assert AutomationErrorCode.EXTERNAL_CALL_FAILED == 4003
|
||
|
|
assert AutomationErrorCode.ACTION_REJECTED == 4004
|
||
|
|
assert AutomationErrorCode.SESSION_NOT_FOUND == 4005
|
||
|
|
assert AutomationErrorCode.APPROVAL_REJECTED == 4006
|
||
|
|
assert AutomationErrorCode.TIMEOUT_HANDOFF == 4007
|
||
|
|
assert AutomationErrorCode.EMPLOYEE_DECLINED == 4008
|
||
|
|
assert AutomationErrorCode.CONFIG_ERROR == 4009
|
||
|
|
assert AutomationErrorCode.MAPPING_FAILED == 4010
|
||
|
|
assert AutomationErrorCode.INVALID_MODE == 4011
|
||
|
|
assert AutomationErrorCode.ROLLBACK_FAILED == 4012
|
||
|
|
|
||
|
|
def test_risk_levels(self):
|
||
|
|
"""测试风险等级"""
|
||
|
|
from app.constants import (
|
||
|
|
AUTOMATION_RISK_READ,
|
||
|
|
AUTOMATION_RISK_LOW,
|
||
|
|
AUTOMATION_RISK_HIGH,
|
||
|
|
AUTOMATION_AUTO_EXECUTABLE_RISKS,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert AUTOMATION_RISK_READ == "read"
|
||
|
|
assert AUTOMATION_RISK_LOW == "low"
|
||
|
|
assert AUTOMATION_RISK_HIGH == "high"
|
||
|
|
assert AUTOMATION_RISK_READ in AUTOMATION_AUTO_EXECUTABLE_RISKS
|
||
|
|
assert AUTOMATION_RISK_LOW in AUTOMATION_AUTO_EXECUTABLE_RISKS
|
||
|
|
assert AUTOMATION_RISK_HIGH not in AUTOMATION_AUTO_EXECUTABLE_RISKS
|
||
|
|
|
||
|
|
def test_session_states(self):
|
||
|
|
"""测试会话状态"""
|
||
|
|
from app.constants import (
|
||
|
|
AUTOMATION_SESSION_CREATED,
|
||
|
|
AUTOMATION_SESSION_RUNNING,
|
||
|
|
AUTOMATION_SESSION_PAUSED,
|
||
|
|
AUTOMATION_SESSION_RESOLVED,
|
||
|
|
AUTOMATION_SESSION_CLOSED,
|
||
|
|
AUTOMATION_SESSION_HANDOFF,
|
||
|
|
AUTOMATION_SESSION_ERROR,
|
||
|
|
AUTOMATION_SESSION_TERMINAL_STATES,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert AUTOMATION_SESSION_CREATED == "created"
|
||
|
|
assert AUTOMATION_SESSION_RUNNING == "running"
|
||
|
|
assert AUTOMATION_SESSION_PAUSED == "paused"
|
||
|
|
assert AUTOMATION_SESSION_RESOLVED == "resolved"
|
||
|
|
assert AUTOMATION_SESSION_CLOSED == "closed"
|
||
|
|
assert AUTOMATION_SESSION_HANDOFF == "handoff"
|
||
|
|
assert AUTOMATION_SESSION_ERROR == "error"
|
||
|
|
|
||
|
|
assert "closed" in AUTOMATION_SESSION_TERMINAL_STATES
|
||
|
|
assert "handoff" in AUTOMATION_SESSION_TERMINAL_STATES
|
||
|
|
assert "error" in AUTOMATION_SESSION_TERMINAL_STATES
|
||
|
|
|
||
|
|
def test_action_states(self):
|
||
|
|
"""测试动作状态"""
|
||
|
|
from app.constants import (
|
||
|
|
AUTOMATION_ACTION_PENDING,
|
||
|
|
AUTOMATION_ACTION_RUNNING,
|
||
|
|
AUTOMATION_ACTION_SUCCESS,
|
||
|
|
AUTOMATION_ACTION_FAILED,
|
||
|
|
AUTOMATION_ACTION_SKIPPED,
|
||
|
|
AUTOMATION_ACTION_AWAIT_APPROVAL,
|
||
|
|
AUTOMATION_ACTION_APPROVED,
|
||
|
|
AUTOMATION_ACTION_REJECTED,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert AUTOMATION_ACTION_PENDING == "pending"
|
||
|
|
assert AUTOMATION_ACTION_RUNNING == "running"
|
||
|
|
assert AUTOMATION_ACTION_SUCCESS == "success"
|
||
|
|
assert AUTOMATION_ACTION_FAILED == "failed"
|
||
|
|
assert AUTOMATION_ACTION_SKIPPED == "skipped"
|
||
|
|
assert AUTOMATION_ACTION_AWAIT_APPROVAL == "await_approval"
|
||
|
|
assert AUTOMATION_ACTION_APPROVED == "approved"
|
||
|
|
assert AUTOMATION_ACTION_REJECTED == "rejected"
|
||
|
|
|
||
|
|
|
||
|
|
class TestActionRegistry:
|
||
|
|
"""测试动作注册表"""
|
||
|
|
|
||
|
|
def test_handler_registry_populated(self):
|
||
|
|
"""测试处理器注册表已填充"""
|
||
|
|
from app.services.automation.action_registry import (
|
||
|
|
HANDLER_REGISTRY,
|
||
|
|
register_builtin_handlers,
|
||
|
|
)
|
||
|
|
|
||
|
|
# 确保内置处理器已注册
|
||
|
|
register_builtin_handlers()
|
||
|
|
|
||
|
|
assert "terminal_locate" in HANDLER_REGISTRY
|
||
|
|
assert "virus_scan" in HANDLER_REGISTRY
|
||
|
|
assert "virus_quarantine" in HANDLER_REGISTRY
|
||
|
|
assert "software_install_guide" in HANDLER_REGISTRY
|
||
|
|
assert "password_reset_link" in HANDLER_REGISTRY
|
||
|
|
|
||
|
|
def test_get_handler_returns_correct_type(self):
|
||
|
|
"""测试获取处理器返回正确类型"""
|
||
|
|
from app.services.automation.action_registry import (
|
||
|
|
get_handler,
|
||
|
|
BaseActionHandler,
|
||
|
|
)
|
||
|
|
|
||
|
|
handler = get_handler("software_install_guide")
|
||
|
|
|
||
|
|
assert handler is not None
|
||
|
|
assert isinstance(handler, BaseActionHandler)
|
||
|
|
assert handler.action_type == "software_install_guide"
|
||
|
|
|
||
|
|
def test_get_handler_unknown_returns_none(self):
|
||
|
|
"""测试获取未知处理器返回 None"""
|
||
|
|
from app.services.automation.action_registry import get_handler
|
||
|
|
|
||
|
|
handler = get_handler("unknown_action_type")
|
||
|
|
|
||
|
|
assert handler is None
|
||
|
|
|
||
|
|
|
||
|
|
class TestExceptionHandler:
|
||
|
|
"""测试异常处理"""
|
||
|
|
|
||
|
|
def test_automation_exception_creation(self):
|
||
|
|
"""测试自动化异常创建"""
|
||
|
|
from app.services.automation.exception_handler import AutomationException
|
||
|
|
from app.constants import AutomationErrorCode
|
||
|
|
|
||
|
|
exc = AutomationException(
|
||
|
|
AutomationErrorCode.SESSION_NOT_FOUND,
|
||
|
|
"会话不存在"
|
||
|
|
)
|
||
|
|
|
||
|
|
assert exc.code == AutomationErrorCode.SESSION_NOT_FOUND
|
||
|
|
assert exc.message == "会话不存在"
|
||
|
|
|
||
|
|
def test_automation_exception_str(self):
|
||
|
|
"""测试异常字符串表示"""
|
||
|
|
from app.services.automation.exception_handler import AutomationException
|
||
|
|
from app.constants import AutomationErrorCode
|
||
|
|
|
||
|
|
exc = AutomationException(
|
||
|
|
AutomationErrorCode.INTENT_FAILED,
|
||
|
|
"意图识别失败"
|
||
|
|
)
|
||
|
|
|
||
|
|
exc_str = str(exc)
|
||
|
|
assert "4002" in exc_str
|
||
|
|
assert "意图识别失败" in exc_str
|
||
|
|
|
||
|
|
|
||
|
|
class TestDefaultScenarioConfigs:
|
||
|
|
"""测试默认场景配置"""
|
||
|
|
|
||
|
|
def test_default_configs_exist(self):
|
||
|
|
"""测试默认配置存在"""
|
||
|
|
from app.services.automation import DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
assert "terminal_locate" in DEFAULT_SCENARIO_CONFIGS
|
||
|
|
assert "virus_dispose" in DEFAULT_SCENARIO_CONFIGS
|
||
|
|
assert "software_install" in DEFAULT_SCENARIO_CONFIGS
|
||
|
|
assert "password_reset" in DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
def test_terminal_locate_config(self):
|
||
|
|
"""测试终端定位配置"""
|
||
|
|
from app.services.automation import DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
config = DEFAULT_SCENARIO_CONFIGS["terminal_locate"]
|
||
|
|
|
||
|
|
assert config["name"] == "终端定位"
|
||
|
|
assert config["enabled"] is True
|
||
|
|
assert "actions" in config
|
||
|
|
assert len(config["actions"]) > 0
|
||
|
|
|
||
|
|
def test_virus_dispose_config(self):
|
||
|
|
"""测试病毒处置配置"""
|
||
|
|
from app.services.automation import DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
config = DEFAULT_SCENARIO_CONFIGS["virus_dispose"]
|
||
|
|
|
||
|
|
assert config["name"] == "病毒查杀处置"
|
||
|
|
assert config["enabled"] is True
|
||
|
|
assert "actions" in config
|
||
|
|
# 应该有扫描和隔离两个动作
|
||
|
|
action_types = [a["action_type"] for a in config["actions"]]
|
||
|
|
assert "virus_scan" in action_types
|
||
|
|
assert "virus_quarantine" in action_types
|
||
|
|
|
||
|
|
def test_software_install_config(self):
|
||
|
|
"""测试软件安装配置"""
|
||
|
|
from app.services.automation import DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
config = DEFAULT_SCENARIO_CONFIGS["software_install"]
|
||
|
|
|
||
|
|
assert config["name"] == "软件自助安装"
|
||
|
|
assert config["enabled"] is True
|
||
|
|
assert len(config["actions"]) == 1
|
||
|
|
assert config["actions"][0]["action_type"] == "software_install_guide"
|
||
|
|
|
||
|
|
def test_password_reset_config(self):
|
||
|
|
"""测试密码重置配置"""
|
||
|
|
from app.services.automation import DEFAULT_SCENARIO_CONFIGS
|
||
|
|
|
||
|
|
config = DEFAULT_SCENARIO_CONFIGS["password_reset"]
|
||
|
|
|
||
|
|
assert config["name"] == "密码重置"
|
||
|
|
assert config["enabled"] is True
|
||
|
|
assert len(config["actions"]) == 1
|
||
|
|
assert config["actions"][0]["action_type"] == "password_reset_link"
|
||
|
|
# 密码重置是高风险操作，需要 H5 确认
|
||
|
|
assert config["actions"][0]["risk_level"] == "high"
|
||
|
|
assert config["actions"][0]["confirm_channel"] == "h5"
|