Simon
146 lines
5.7 KiB
Python
146 lines
5.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""知识库自动迭代 真实验证（P2-13）
|
||
|
||
真实验证点（来自功能规格说明书 + 状态看板验收标准）：
|
||
- 能基于标注(feedback=useless)生成建议行 status=pending
|
||
- 管理员 approve 后写入 knowledge_base（状态变为 applied）
|
||
- reject 正常（状态变为 rejected，且不写入知识库）
|
||
- get_suggestion_stats 统计正确
|
||
- 关键证据: _generate_update_suggestion / _generate_new_faq_suggestion 内是 TODO 桩，
|
||
返回的 title/content 是 "[待AI生成] ..." 占位符 —— 证实 AI 内容生成未实现，
|
||
数据管道(分析→建建议行→审核应用)是真实的，但 AI 生成是桩。
|
||
"""
|
||
import pytest
|
||
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
|
||
|
||
|
||
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",
|
||
)
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_analyze_generates_pending_suggestion_with_stub_content(db_session):
|
||
"""分析标注生成 pending 建议；且内容是 [待AI生成] 占位符（证明 AI 生成是桩）。"""
|
||
_seed_useless_annotations(db_session, msg_id="msg-x", n=3)
|
||
await db_session.flush()
|
||
|
||
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 # 3(msg-x) + 1(msg-other)
|
||
|
||
# 查询生成的建议
|
||
stmt = select(KnowledgeSuggestion).where(KnowledgeSuggestion.status == "pending")
|
||
suggestions = (await db_session.execute(stmt)).scalars().all()
|
||
assert len(suggestions) >= 1
|
||
|
||
# 关键证据: AI 内容生成是桩 —— title/content 含占位符
|
||
titles = [s.title for s in suggestions]
|
||
contents = [s.content for s in suggestions]
|
||
assert any("[待AI生成]" in t for t in titles)
|
||
assert any("请通过AI分析" in c for c in contents)
|
||
|
||
# 仅高频的 msg-x 生成建议，msg-other(仅1次)不应生成
|
||
generated_source = [sd for s in suggestions for sd in (s.source_data or [])]
|
||
assert "msg-x" in generated_source
|
||
assert "msg-other" not in generated_source
|
||
|
||
|
||
@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()
|
||
|
||
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生成]" in kb_rows[0].title
|
||
|
||
|
||
@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()
|
||
|
||
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()
|
||
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
|
||
|
||
# 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
|