Files

460 lines
20 KiB
Python
Raw Permalink Normal View History

# =============================================================================
# 企微IT智能服务台 — WingmanService 单元测试
# =============================================================================
# 测试覆盖:
# 1. _build_context_messages() — 消息角色映射
# 2. _parse_json_response() — JSON 解析三种场景
# 3. _estimate_confidence() — 置信度估算
# 4. generate_draft() — 草稿生成 + 降级
# 5. generate_summary() — 摘要生成 + 降级
# 6. suggest_tags() — 标签建议 + 降级
# =============================================================================
import json
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.services.wingman_service import WingmanService
# =============================================================================
# _build_context_messages 测试
# =============================================================================
class TestBuildContextMessages:
"""测试消息角色映射逻辑。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
def test_employee_messages_mapped_to_user(self):
"""员工消息应映射为 user 角色。"""
messages = [
{"sender_type": "employee", "content": "我的电脑开不了机"},
]
result = self.service._build_context_messages(messages, "测试 system prompt")
assert len(result) == 2 # system prompt + 1 条消息
assert result[0]["role"] == "system"
assert result[1]["role"] == "user"
assert result[1]["content"] == "我的电脑开不了机"
def test_agent_messages_mapped_to_assistant(self):
"""坐席消息应映射为 assistant 角色。"""
messages = [
{"sender_type": "agent", "content": "请尝试重启电脑"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
assert result[1]["role"] == "assistant"
def test_ai_messages_mapped_to_assistant(self):
"""AI 消息应映射为 assistant 角色。"""
messages = [
{"sender_type": "ai", "content": "建议您检查电源线连接"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
assert result[1]["role"] == "assistant"
def test_system_messages_are_skipped(self):
"""系统消息应被跳过(已有 system prompt)。"""
messages = [
{"sender_type": "system", "content": "坐席已接入"},
{"sender_type": "employee", "content": "你好"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
# system prompt + employee message, system消息被跳过
assert len(result) == 2
assert result[1]["role"] == "user"
assert result[1]["content"] == "你好"
def test_empty_content_messages_are_skipped(self):
"""内容为空的消息应被跳过。"""
messages = [
{"sender_type": "employee", "content": ""},
{"sender_type": "agent", "content": "收到"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
# system prompt + agent message, 空 content 的 employee 消息被跳过
assert len(result) == 2
assert result[1]["role"] == "assistant"
def test_unknown_sender_type_defaults_to_user(self):
"""未知发送者类型应默认映射为 user。"""
messages = [
{"sender_type": "unknown_type", "content": "未知消息"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
assert result[1]["role"] == "user"
def test_full_conversation_ordering(self):
"""多轮对话应保持正确的顺序。"""
messages = [
{"sender_type": "employee", "content": "VPN连不上"},
{"sender_type": "agent", "content": "请问是哪个VPN?"},
{"sender_type": "employee", "content": "公司内网VPN"},
{"sender_type": "ai", "content": "建议检查VPN客户端版本"},
]
result = self.service._build_context_messages(messages, "测试 prompt")
# system prompt + 4 条消息
assert len(result) == 5
assert result[1]["role"] == "user"
assert result[2]["role"] == "assistant"
assert result[3]["role"] == "user"
assert result[4]["role"] == "assistant"
# =============================================================================
# _parse_json_response 测试
# =============================================================================
class TestParseJsonResponse:
"""测试 AI 返回 JSON 解析的三种场景。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
def test_parse_pure_json(self):
"""纯 JSON 字符串应直接解析成功。"""
content = '{"problem": "VPN断连", "cause": "证书过期", "solution": "更新证书"}'
default = {"problem": "", "cause": "", "solution": ""}
result = self.service._parse_json_response(content, default)
assert result["problem"] == "VPN断连"
assert result["cause"] == "证书过期"
assert result["solution"] == "更新证书"
def test_parse_markdown_code_block(self):
"""markdown 代码块中的 JSON 应被提取并解析。"""
content = '```json\n{"suggested_tags": ["VPN", "网络"], "category": "网络", "priority": "high"}\n```'
default = {"suggested_tags": [], "category": "", "priority": "medium"}
result = self.service._parse_json_response(content, default)
assert result["suggested_tags"] == ["VPN", "网络"]
assert result["category"] == "网络"
assert result["priority"] == "high"
def test_parse_markdown_code_block_without_language(self):
"""无语言标记的 markdown 代码块也应被解析。"""
content = '```\n{"problem": "密码错误", "cause": "输入错误", "solution": "重置密码"}\n```'
default = {"problem": "", "cause": "", "solution": ""}
result = self.service._parse_json_response(content, default)
assert result["problem"] == "密码错误"
def test_parse_curly_brace_extraction(self):
"""含有额外文本的 AI 返回应通过首尾花括号提取 JSON。"""
content = '这是AI的分析结果:{"problem": "邮箱满了", "cause": "未清理", "solution": "清理邮箱"} 希望对你有帮助'
default = {"problem": "", "cause": "", "solution": ""}
result = self.service._parse_json_response(content, default)
assert result["problem"] == "邮箱满了"
assert result["solution"] == "清理邮箱"
def test_parse_empty_content_returns_default(self):
"""空内容应返回默认值。"""
default = {"problem": "无法自动生成摘要", "cause": "", "solution": ""}
result = self.service._parse_json_response("", default)
assert result == default
def test_parse_invalid_content_returns_default(self):
"""无法解析的内容应返回默认值。"""
content = "这不是JSON格式的内容"
default = {"problem": "无法自动生成摘要", "cause": "", "solution": ""}
result = self.service._parse_json_response(content, default)
assert result == default
def test_parse_none_content_returns_default(self):
"""None 内容应返回默认值。"""
default = {"suggested_tags": [], "category": "", "priority": "medium"}
result = self.service._parse_json_response(None, default)
assert result == default
# =============================================================================
# _estimate_confidence 测试
# =============================================================================
class TestEstimateConfidence:
"""测试 AI 草稿置信度估算。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
def test_short_content_low_confidence(self):
"""过短内容应返回低置信度。"""
result = self.service._estimate_confidence("hi")
# len("hi") < 5,应返回 0.2
assert result == 0.2
def test_very_short_content_low_confidence(self):
"""极短内容(<10字符)应降低置信度。"""
result = self.service._estimate_confidence("试试看")
# len("试试看") = 3 < 5,返回 0.2
assert result == 0.2
def test_moderate_content_confidence(self):
"""适中内容应返回中等置信度。"""
# 30+ 字符,无不确定措辞,无确定措辞
content = "您好,请检查您的网络连接是否正常,然后重新启动应用程序即可。"
result = self.service._estimate_confidence(content)
# 基础 0.8,长度 >= 30 不减,无不确定措辞不减,无确定措辞不加
assert 0.7 <= result <= 1.0
def test_uncertain_phrases_lower_confidence(self):
"""包含不确定措辞应降低置信度。"""
content_with_uncertain = "可能是网络问题,建议您检查一下连接"
content_without_uncertain = "这是网络问题,请检查网络连接设置"
conf_with = self.service._estimate_confidence(content_with_uncertain)
conf_without = self.service._estimate_confidence(content_without_uncertain)
# 含"可能"和"建议您"的置信度应更低
assert conf_with < conf_without
def test_confident_phrases_raise_confidence(self):
"""包含确定措辞(步骤、链接等)应提高置信度。"""
content = "请按以下步骤操作:1.打开设置 2.点击网络 3.选择连接。详细请查看 http://help.example.com"
result = self.service._estimate_confidence(content)
# 包含"步骤"、"请按以下"、"http" 至少 +0.15
assert result >= 0.9
def test_confidence_bounded_to_range(self):
"""置信度应限制在 0.0-1.0 范围内。"""
# 多个不确定措辞 + 短内容
content = "可能大概也许不确定建议您"
result = self.service._estimate_confidence(content)
assert result >= 0.0
# 多个确定措辞
content = "请按以下步骤操作:点击打开 http://link1 http://link2"
result = self.service._estimate_confidence(content)
assert result <= 1.0
def test_empty_string_returns_minimum(self):
"""空字符串应返回低置信度。"""
result = self.service._estimate_confidence("")
assert result == 0.2
def test_whitespace_only_returns_minimum(self):
"""仅空白字符应返回低置信度。"""
result = self.service._estimate_confidence(" ")
assert result == 0.2
# =============================================================================
# generate_draft 降级测试
# =============================================================================
class TestGenerateDraft:
"""测试草稿生成的降级处理。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
@pytest.mark.asyncio
async def test_draft_degradation_when_api_returns_none(self):
"""Wingman API 返回 None 时应返回降级默认值。"""
self.service._call_wingman_api = AsyncMock(return_value=None)
result = await self.service.generate_draft(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "VPN连不上"}],
)
assert result["content"] == ""
assert result["confidence"] == 0.0
assert "不可用" in result["reasoning"]
@pytest.mark.asyncio
async def test_draft_degradation_when_api_raises_exception(self):
"""Wingman API 抛异常时应返回降级默认值(不抛异常)。"""
self.service._call_wingman_api = AsyncMock(side_effect=Exception("连接超时"))
result = await self.service.generate_draft(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "VPN连不上"}],
)
assert result["content"] == ""
assert result["confidence"] == 0.0
assert "异常" in result["reasoning"]
@pytest.mark.asyncio
async def test_draft_success_returns_structured_result(self):
"""Wingman API 正常返回时应返回结构化结果。"""
self.service._call_wingman_api = AsyncMock(
return_value="请尝试重启VPN客户端并重新输入密码"
)
result = await self.service.generate_draft(
conversation_id="conv-001",
messages=[
{"sender_type": "employee", "content": "VPN连不上"},
{"sender_type": "agent", "content": "请问报什么错?"},
],
)
assert result["content"] == "请尝试重启VPN客户端并重新输入密码"
assert 0.0 < result["confidence"] <= 1.0
assert "推理" in result["reasoning"] or "对话上下文" in result["reasoning"]
# =============================================================================
# generate_summary 降级测试
# =============================================================================
class TestGenerateSummary:
"""测试摘要生成的降级处理。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
@pytest.mark.asyncio
async def test_summary_degradation_when_api_returns_none(self):
"""Wingman API 返回 None 时应返回降级默认摘要。"""
self.service._call_wingman_api = AsyncMock(return_value=None)
result = await self.service.generate_summary(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "求助"}],
)
assert result["problem"] == "无法自动生成摘要"
assert result["cause"] == ""
assert result["solution"] == ""
@pytest.mark.asyncio
async def test_summary_degradation_when_api_raises_exception(self):
"""Wingman API 抛异常时应返回降级默认摘要。"""
self.service._call_wingman_api = AsyncMock(side_effect=Exception("超时"))
result = await self.service.generate_summary(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "求助"}],
)
assert result["problem"] == "无法自动生成摘要"
@pytest.mark.asyncio
async def test_summary_success_parses_json_response(self):
"""Wingman API 正常返回 JSON 时应正确解析摘要。"""
self.service._call_wingman_api = AsyncMock(
return_value='{"problem": "VPN断连", "cause": "证书过期", "solution": "更新证书"}'
)
result = await self.service.generate_summary(
conversation_id="conv-001",
messages=[
{"sender_type": "employee", "content": "VPN连不上"},
],
)
assert result["problem"] == "VPN断连"
assert result["cause"] == "证书过期"
assert result["solution"] == "更新证书"
# =============================================================================
# suggest_tags 降级测试
# =============================================================================
class TestSuggestTags:
"""测试标签建议的降级处理。"""
def setup_method(self):
"""每个测试方法执行前的初始化。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://test-api"
mock_settings.dify_wingman_api_key = "test-key"
mock_settings.dify_wingman_timeout = 10
self.service = WingmanService()
@pytest.mark.asyncio
async def test_tags_degradation_when_api_returns_none(self):
"""Wingman API 返回 None 时应返回降级默认标签。"""
self.service._call_wingman_api = AsyncMock(return_value=None)
result = await self.service.suggest_tags(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "求助"}],
)
assert result["suggested_tags"] == []
assert result["category"] == ""
assert result["priority"] == "medium"
@pytest.mark.asyncio
async def test_tags_degradation_when_api_raises_exception(self):
"""Wingman API 抛异常时应返回降级默认标签。"""
self.service._call_wingman_api = AsyncMock(side_effect=Exception("超时"))
result = await self.service.suggest_tags(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "求助"}],
)
assert result["suggested_tags"] == []
assert result["priority"] == "medium"
@pytest.mark.asyncio
async def test_tags_success_parses_json_response(self):
"""Wingman API 正常返回 JSON 时应正确解析标签。"""
self.service._call_wingman_api = AsyncMock(
return_value='{"suggested_tags": ["VPN", "网络"], "category": "网络", "priority": "high"}'
)
result = await self.service.suggest_tags(
conversation_id="conv-001",
messages=[{"sender_type": "employee", "content": "VPN连不上"}],
)
assert result["suggested_tags"] == ["VPN", "网络"]
assert result["category"] == "网络"
assert result["priority"] == "high"
# =============================================================================
# WingmanService 初始化测试
# =============================================================================
class TestWingmanServiceInit:
"""测试 WingmanService 初始化。"""
def test_service_reads_config_from_settings(self):
"""WingmanService 应从 settings 正确读取配置。"""
with patch("app.services.wingman_service.settings") as mock_settings:
mock_settings.dify_wingman_api_url = "http://custom-api-url"
mock_settings.dify_wingman_api_key = "custom-api-key"
mock_settings.dify_wingman_timeout = 60
service = WingmanService()
assert service.api_url == "http://custom-api-url"
assert service.api_key == "custom-api-key"
assert service.timeout == 60