Simon
278 lines
9.4 KiB
Python
278 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""知识库自动迭代 真实验证（Tier0 / T03 重写）
|
|
|
|
Tier0 变更：
|
|
- 移除 [待AI生成] 占位符断言（AI 生成已通过 WingmanService 真实实现）
|
|
- 新增 source_failed 测试（Dify 不可用/置信度不足时标记）
|
|
- 新增 audience 自动标注测试
|
|
- 新增置信门控测试
|
|
- 保留数据管道验证（分析→建建议→审核→写KB）
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from sqlalchemy import select
|
|
|
|
from app.models.conversation_annotation import ConversationAnnotation
|
|
from app.models.knowledge_base import KnowledgeBase
|
|
from app.models.knowledge_suggestion import KnowledgeSuggestion
|
|
from app.services.knowledge_iteration_service import KnowledgeIterationService
|
|
|
|
|
|
# ── Mock WingmanService 返回 ──
|
|
|
|
MOCK_AI_RESULT = {
|
|
"suggestion_type": "new_faq",
|
|
"title": "VPN无法连接的解决方案",
|
|
"content": "1. 检查网络连接 2. 重启VPN客户端 3. 联系IT支持",
|
|
"category": "网络",
|
|
"tags": ["VPN", "连接"],
|
|
"confidence": 0.86,
|
|
"issue": "VPN问题",
|
|
"action": "VPN连接修复",
|
|
"relation_type": "LEADS_TO",
|
|
"parent_issue": "网络问题",
|
|
}
|
|
|
|
MOCK_AI_RESULT_LOW_CONFIDENCE = {
|
|
"suggestion_type": "new_faq",
|
|
"title": "不确定的建议",
|
|
"content": "可能是网络问题",
|
|
"category": "网络",
|
|
"tags": [],
|
|
"confidence": 0.42,
|
|
"issue": "",
|
|
"action": "",
|
|
"relation_type": "LEADS_TO",
|
|
"parent_issue": "",
|
|
}
|
|
|
|
|
|
def _seed_useless_annotations(
|
|
db,
|
|
msg_id: str,
|
|
n: int,
|
|
conv_id: str = "conv-1",
|
|
agent_id: str = "agent-1",
|
|
):
|
|
"""播种 n 条 feedback=useless 的标注（同一 message_id 用于触发高频错误判定）。"""
|
|
for _ in range(n):
|
|
db.add(
|
|
ConversationAnnotation(
|
|
conversation_id=conv_id,
|
|
agent_id=agent_id,
|
|
message_id=msg_id,
|
|
feedback="useless",
|
|
)
|
|
)
|
|
# 再播种一条不同 message_id 的（仅 1 次，不构成高频，用于对照）
|
|
db.add(
|
|
ConversationAnnotation(
|
|
conversation_id="conv-2",
|
|
agent_id=agent_id,
|
|
message_id="msg-other",
|
|
feedback="useless",
|
|
)
|
|
)
|
|
|
|
|
|
async def _create_mock_wingman():
|
|
"""创建一个返回预置结果的 Mock WingmanService。"""
|
|
mock = MagicMock()
|
|
mock.generate_knowledge_suggestion = AsyncMock(return_value=MOCK_AI_RESULT)
|
|
mock.close = AsyncMock()
|
|
return mock
|
|
|
|
|
|
# =============================================================================
|
|
# 测试用例
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_analyze_generates_pending_suggestion(db_session):
|
|
"""分析标注生成 pending 建议；AI 生成内容不再是 [待AI生成] 占位符。"""
|
|
_seed_useless_annotations(db_session, msg_id="msg-x", n=3)
|
|
await db_session.flush()
|
|
|
|
with patch(
|
|
"app.services.wingman_service.WingmanService",
|
|
return_value=await _create_mock_wingman(),
|
|
):
|
|
service = KnowledgeIterationService()
|
|
result = await service.analyze_and_generate_suggestions(db_session, days=30)
|
|
|
|
# 高频错误(msg-x 被标注 3 次)应生成 >=1 条建议
|
|
assert result["suggestions_generated"] >= 1
|
|
assert result["annotations_analyzed"] >= 4
|
|
|
|
# 查询生成的建议
|
|
stmt = select(KnowledgeSuggestion).where(KnowledgeSuggestion.status == "pending")
|
|
suggestions = (await db_session.execute(stmt)).scalars().all()
|
|
assert len(suggestions) >= 1
|
|
|
|
# Tier0 关键变更: AI 内容不再是 [待AI生成] 占位符
|
|
titles = [s.title for s in suggestions]
|
|
contents = [s.content for s in suggestions]
|
|
assert not any("[待AI生成]" in t for t in titles)
|
|
assert not any("请通过AI分析" in c for c in contents)
|
|
|
|
# 新增字段验证
|
|
for s in suggestions:
|
|
if not s.source_failed:
|
|
assert s.confidence is not None
|
|
assert s.confidence >= 0.0
|
|
assert s.audience is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approve_writes_knowledge_base(db_session):
|
|
"""approve 后写入 knowledge_base，建议状态变为 applied。"""
|
|
_seed_useless_annotations(db_session, msg_id="msg-x", n=3)
|
|
await db_session.flush()
|
|
|
|
with patch(
|
|
"app.services.wingman_service.WingmanService",
|
|
return_value=await _create_mock_wingman(),
|
|
):
|
|
service = KnowledgeIterationService()
|
|
await service.analyze_and_generate_suggestions(db_session, days=30)
|
|
|
|
stmt = select(KnowledgeSuggestion).where(KnowledgeSuggestion.status == "pending")
|
|
suggestion = (await db_session.execute(stmt)).scalars().first()
|
|
assert suggestion is not None
|
|
|
|
approved = await service.approve_suggestion(db_session, suggestion.id, "reviewer-1")
|
|
assert approved is not None
|
|
assert approved.status == "applied"
|
|
assert approved.reviewer_id == "reviewer-1"
|
|
|
|
# knowledge_base 应新增一行
|
|
kb_rows = (await db_session.execute(select(KnowledgeBase))).scalars().all()
|
|
assert len(kb_rows) == 1
|
|
# 内容不再是占位符
|
|
assert "[待AI生成]" not in kb_rows[0].title
|
|
# 图字段应被保留
|
|
assert kb_rows[0].graph_sync_status in ("pending", "synced", "failed")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_marks_rejected(db_session):
|
|
"""reject 将建议标记为 rejected，且不写入知识库。"""
|
|
_seed_useless_annotations(db_session, msg_id="msg-x", n=3)
|
|
await db_session.flush()
|
|
|
|
with patch(
|
|
"app.services.wingman_service.WingmanService",
|
|
return_value=await _create_mock_wingman(),
|
|
):
|
|
service = KnowledgeIterationService()
|
|
await service.analyze_and_generate_suggestions(db_session, days=30)
|
|
|
|
stmt = select(KnowledgeSuggestion).where(KnowledgeSuggestion.status == "pending")
|
|
suggestion = (await db_session.execute(stmt)).scalars().first()
|
|
|
|
rejected = await service.reject_suggestion(
|
|
db_session, suggestion.id, "reviewer-2", "内容无意义"
|
|
)
|
|
assert rejected is not None
|
|
assert rejected.status == "rejected"
|
|
assert rejected.reject_reason == "内容无意义"
|
|
|
|
# 拒绝不写入知识库
|
|
kb_rows = (await db_session.execute(select(KnowledgeBase))).scalars().all()
|
|
assert len(kb_rows) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stats_counts_correctly(db_session):
|
|
"""get_suggestion_stats 统计正确（含新增状态）。"""
|
|
_seed_useless_annotations(db_session, msg_id="msg-x", n=3)
|
|
await db_session.flush()
|
|
|
|
with patch(
|
|
"app.services.wingman_service.WingmanService",
|
|
return_value=await _create_mock_wingman(),
|
|
):
|
|
service = KnowledgeIterationService()
|
|
await service.analyze_and_generate_suggestions(db_session, days=30)
|
|
|
|
stats = await service.get_suggestion_stats(db_session)
|
|
assert stats["total"] >= 1
|
|
assert stats["pending"] >= 1
|
|
# 新增状态键应存在
|
|
assert "queued" in stats
|
|
assert "graph_synced" in stats
|
|
assert "expired" in stats
|
|
|
|
# approve 一条后 applied +1
|
|
stmt = select(KnowledgeSuggestion).where(KnowledgeSuggestion.status == "pending")
|
|
s = (await db_session.execute(stmt)).scalars().first()
|
|
await service.approve_suggestion(db_session, s.id, "reviewer-1")
|
|
stats2 = await service.get_suggestion_stats(db_session)
|
|
assert stats2["applied"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_failed_on_low_confidence(db_session):
|
|
"""验证置信度 < 0.7 的提案标记 source_failed=True。"""
|
|
_seed_useless_annotations(db_session, msg_id="msg-low-conf", n=3)
|
|
await db_session.flush()
|
|
|
|
# Mock 返回低置信度结果
|
|
mock = MagicMock()
|
|
mock.generate_knowledge_suggestion = AsyncMock(
|
|
return_value=MOCK_AI_RESULT_LOW_CONFIDENCE
|
|
)
|
|
mock.close = AsyncMock()
|
|
|
|
with patch(
|
|
"app.services.wingman_service.WingmanService",
|
|
return_value=mock,
|
|
):
|
|
service = KnowledgeIterationService()
|
|
await service.analyze_and_generate_suggestions(db_session, days=30)
|
|
|
|
stmt = select(KnowledgeSuggestion).where(
|
|
KnowledgeSuggestion.source_data.contains("msg-low-conf")
|
|
)
|
|
suggestions = (await db_session.execute(stmt)).scalars().all()
|
|
assert len(suggestions) >= 1
|
|
# 置信度 < 0.7 应标记 source_failed
|
|
for s in suggestions:
|
|
if s.confidence is not None and s.confidence < 0.7:
|
|
assert s.source_failed is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queue_suggestion_transitions(db_session):
|
|
"""验证入队列状态转换。"""
|
|
# 直接创建一个 pending 的建议
|
|
suggestion = KnowledgeSuggestion(
|
|
suggestion_type="new_faq",
|
|
status="pending",
|
|
title="队列测试建议",
|
|
content="测试内容",
|
|
category="软件",
|
|
tags=["测试"],
|
|
source_type="conversation",
|
|
source_data=["conv-queue-test"],
|
|
reason="入队列测试",
|
|
confidence=0.8,
|
|
audience="employee_quick_reply",
|
|
)
|
|
db_session.add(suggestion)
|
|
await db_session.commit()
|
|
await db_session.refresh(suggestion)
|
|
|
|
service = KnowledgeIterationService()
|
|
result = await service.queue_suggestion(db_session, suggestion.id)
|
|
|
|
assert result is not None
|
|
assert result.status == "queued"
|
|
assert result.queued_at is not None
|
|
|
|
# 获取队列统计
|
|
queue_stats = await service.get_queue_stats(db_session)
|
|
assert queue_stats["queued_total"] >= 1
|
|
assert "by_audience" in queue_stats
|