598 lines
19 KiB
Python
598 lines
19 KiB
Python
from django.urls import reverse
|
|
|
|
from agent_core.results import AgentResult
|
|
from apps.audit.models import AgentAuditLog, DemoBusinessRecord, NotificationRecord
|
|
from apps.audit.services import create_audit_log, create_notification_record
|
|
from apps.chat.models import Conversation
|
|
from apps.documents.models import SubmissionBatch
|
|
from agent_core.tools.builtin_tools import query_demo_records
|
|
|
|
|
|
def test_create_audit_log_records_success_result(db):
|
|
result = AgentResult(answer="回答", structured_output={"x": 1}, status="success")
|
|
|
|
log = create_audit_log("knowledge_qa", "知识库问答助手", "问题", result)
|
|
|
|
assert AgentAuditLog.objects.count() == 1
|
|
assert log.final_answer == "回答"
|
|
assert log.structured_output == {"x": 1}
|
|
assert log.status == "success"
|
|
|
|
|
|
def test_audit_list_page_shows_log(client, db):
|
|
result = AgentResult(answer="回答", status="success")
|
|
create_audit_log("knowledge_qa", "知识库问答助手", "问题", result)
|
|
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
assert response.status_code == 200
|
|
assert "知识库问答助手" in response.content.decode("utf-8")
|
|
|
|
|
|
def test_audit_list_can_filter_by_scenario(client, db):
|
|
create_audit_log(
|
|
"knowledge_qa",
|
|
"知识库问答助手",
|
|
"制度问题",
|
|
AgentResult(answer="回答一", status="success"),
|
|
)
|
|
create_audit_log(
|
|
"quality_analysis",
|
|
"质量异常分析助手",
|
|
"质量问题",
|
|
AgentResult(answer="回答二", status="success"),
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"), {"scenario_id": "knowledge_qa"})
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "知识库问答助手" in content
|
|
assert "质量异常分析助手" not in content
|
|
|
|
|
|
def test_audit_list_page_shows_user_input_summary(client, db):
|
|
create_audit_log(
|
|
"knowledge_qa",
|
|
"知识库问答助手",
|
|
"这是一个比较长的用户输入，用于确认列表页会展示输入摘要。",
|
|
AgentResult(answer="回答", status="success"),
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
assert "这是一个比较长的用户输入" in response.content.decode("utf-8")
|
|
|
|
|
|
def test_audit_detail_page_shows_raw_output(client, db):
|
|
result = AgentResult(
|
|
answer="结构化回答",
|
|
raw_output='{"answer":"结构化回答","confidence":"high"}',
|
|
status="success",
|
|
)
|
|
log = create_audit_log("knowledge_qa", "知识库问答助手", "问题", result)
|
|
|
|
response = client.get(reverse("audit:detail", args=[log.id]))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "原始输出" in content
|
|
assert "confidence" in content
|
|
assert "high" in content
|
|
|
|
|
|
def test_create_audit_log_masks_api_keys_from_error_message(db):
|
|
result = AgentResult(
|
|
answer="",
|
|
status="failed",
|
|
error="LLM_API_KEY=sk-secret-value 调用失败",
|
|
)
|
|
|
|
log = create_audit_log("knowledge_qa", "知识库问答助手", "问题", result)
|
|
|
|
assert "sk-secret-value" not in log.error_message
|
|
assert "sk-***" in log.error_message
|
|
|
|
|
|
def test_create_audit_log_masks_embedding_api_keys_from_error_message(db):
|
|
result = AgentResult(
|
|
answer="",
|
|
status="failed",
|
|
error="EMBEDDING_API_KEY=embed-secret 调用失败",
|
|
)
|
|
|
|
log = create_audit_log("knowledge_qa", "知识库问答助手", "问题", result)
|
|
|
|
assert "embed-secret" not in log.error_message
|
|
assert "EMBEDDING_API_KEY=***" in log.error_message
|
|
|
|
|
|
def test_query_demo_records_reads_demo_business_record_table(db):
|
|
DemoBusinessRecord.objects.create(
|
|
scenario_id="quality_analysis",
|
|
record_type="defect",
|
|
title="A线缺陷",
|
|
payload={"rate": 0.12},
|
|
)
|
|
|
|
result = query_demo_records(user_input="quality_analysis defect")
|
|
|
|
assert result["records"][0]["title"] == "A线缺陷"
|
|
assert result["records"][0]["payload"] == {"rate": 0.12}
|
|
|
|
|
|
def test_audit_log_records_batch_conversation_and_product_context(db):
|
|
result = AgentResult(answer="回答", status="success")
|
|
|
|
log = create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"开始审核",
|
|
result,
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="新型冠状病毒 2019-nCoV 核酸检测试剂盒",
|
|
)
|
|
|
|
assert log.batch_id == "SUB-20260604-001"
|
|
assert log.conversation_id == "conv-001"
|
|
assert log.product_name == "新型冠状病毒 2019-nCoV 核酸检测试剂盒"
|
|
|
|
|
|
def test_create_notification_record_persists_task_completed_and_task_failed(db):
|
|
completed = create_notification_record(
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_completed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_1",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/1",
|
|
receipt={"message_id": "msg-1"},
|
|
)
|
|
failed = create_notification_record(
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_failed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_1",
|
|
message_status="failed",
|
|
web_detail_url="https://example.com/detail/1",
|
|
receipt={"message_id": "msg-2"},
|
|
)
|
|
|
|
assert NotificationRecord.objects.count() == 2
|
|
assert completed.notify_reason == "task_completed"
|
|
assert failed.notify_reason == "task_failed"
|
|
|
|
|
|
def test_create_notification_record_rejects_unsupported_notify_reason(db):
|
|
try:
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
trigger_source="risk_report",
|
|
notify_reason="custom_reason",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_1",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/1",
|
|
receipt={"message_id": "msg-1"},
|
|
)
|
|
assert False, "expected ValueError"
|
|
except ValueError as exc:
|
|
assert "notify_reason" in str(exc)
|
|
|
|
|
|
def test_audit_list_supports_batch_and_product_filters(client, db):
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(answer="回答一", status="success"),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题二",
|
|
AgentResult(answer="回答二", status="success"),
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"), {"keyword": "产品A"})
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "产品A" in content
|
|
assert "产品B" not in content
|
|
|
|
|
|
def test_audit_list_shows_risk_and_notification_status(client, db):
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(
|
|
answer="回答一",
|
|
status="success",
|
|
structured_output={"highest_risk_level": "high"},
|
|
),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_completed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_1",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/1",
|
|
receipt={"message_id": "msg-1"},
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "风险状态" in content
|
|
assert "已阻断" in content
|
|
assert "通知状态" in content
|
|
assert "已发送" in content
|
|
assert "通知原因" in content
|
|
assert "task_completed" in content
|
|
|
|
|
|
def test_audit_list_can_filter_by_notification_status(client, db):
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(answer="回答一", status="success"),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题二",
|
|
AgentResult(answer="回答二", status="failed"),
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
)
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_completed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_1",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/1",
|
|
receipt={"message_id": "msg-1"},
|
|
)
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_failed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_2",
|
|
message_status="failed",
|
|
web_detail_url="https://example.com/detail/2",
|
|
receipt={"message_id": "msg-2"},
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"), {"notify_status": "failed"})
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "产品B" in content
|
|
assert "产品A" not in content
|
|
|
|
|
|
def test_audit_list_can_filter_by_risk_status(client, db):
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(
|
|
answer="回答一",
|
|
status="success",
|
|
structured_output={"highest_risk_level": "high"},
|
|
),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题二",
|
|
AgentResult(
|
|
answer="回答二",
|
|
status="success",
|
|
structured_output={"highest_risk_level": "low"},
|
|
),
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"), {"risk_status": "high"})
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "产品A" in content
|
|
assert "产品B" not in content
|
|
|
|
|
|
def test_audit_detail_page_shows_conversation_node_results(client, db):
|
|
Conversation.objects.create(
|
|
conversation_id="conv-001",
|
|
title="产品A",
|
|
product_name="产品A",
|
|
batch_id="SUB-20260604-001",
|
|
task_status="failed",
|
|
node_results=[
|
|
{"label": "资料包导入", "status": "已完成"},
|
|
{"label": "风险预警", "status": "已阻断"},
|
|
{"label": "飞书通知", "status": "失败"},
|
|
],
|
|
)
|
|
log = create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(answer="回答一", status="failed"),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
|
|
response = client.get(reverse("audit:detail", args=[log.id]))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "会话节点结果" in content
|
|
assert "风险预警 / 已阻断" in content
|
|
assert "飞书通知 / 失败" in content
|
|
|
|
|
|
def test_audit_list_shows_batch_scale_and_conversation_status(client, db):
|
|
SubmissionBatch.objects.create(
|
|
batch_id="SUB-20260604-001",
|
|
product_name="产品A",
|
|
workflow_type="registration",
|
|
conversation_id="conv-001",
|
|
file_count=4,
|
|
page_count=26,
|
|
import_status="review_required",
|
|
)
|
|
Conversation.objects.create(
|
|
conversation_id="conv-001",
|
|
title="产品A",
|
|
product_name="产品A",
|
|
batch_id="SUB-20260604-001",
|
|
task_status="failed",
|
|
node_results=[
|
|
{"label": "风险预警", "status": "已阻断"},
|
|
{"label": "飞书通知", "status": "失败"},
|
|
],
|
|
)
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"问题一",
|
|
AgentResult(answer="回答一", status="failed"),
|
|
batch_id="SUB-20260604-001",
|
|
conversation_id="conv-001",
|
|
product_name="产品A",
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "资料规模" in content
|
|
assert "4 份 / 26 页" in content
|
|
assert "会话状态" in content
|
|
assert "失败" in content
|
|
assert "待复核" in content
|
|
|
|
|
|
def test_audit_list_shows_history_metrics_and_context_links(client, db):
|
|
SubmissionBatch.objects.create(
|
|
batch_id="SUB-20260604-009",
|
|
product_name="产品C",
|
|
workflow_type="registration",
|
|
conversation_id="conv-009",
|
|
file_count=5,
|
|
page_count=31,
|
|
import_status="completed",
|
|
)
|
|
Conversation.objects.create(
|
|
conversation_id="conv-009",
|
|
title="产品C",
|
|
product_name="产品C",
|
|
batch_id="SUB-20260604-009",
|
|
task_status="success",
|
|
node_results=[
|
|
{"label": "风险预警", "status": "已完成"},
|
|
{"label": "飞书通知", "status": "已完成"},
|
|
],
|
|
)
|
|
create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"产品C 审核",
|
|
AgentResult(
|
|
answer="完成",
|
|
status="success",
|
|
structured_output={"highest_risk_level": "medium"},
|
|
),
|
|
batch_id="SUB-20260604-009",
|
|
conversation_id="conv-009",
|
|
product_name="产品C",
|
|
)
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-009",
|
|
conversation_id="conv-009",
|
|
product_name="产品C",
|
|
trigger_source="risk_report",
|
|
notify_reason="task_completed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_9",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/9",
|
|
receipt={"message_id": "msg-9"},
|
|
)
|
|
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "处理任务数" in content
|
|
assert "成功执行" in content
|
|
assert "通知已发送" in content
|
|
assert "高风险阻断" in content
|
|
assert reverse("chat:detail", args=["conv-009"]) in content
|
|
assert f"{reverse('documents:list')}?keyword=SUB-20260604-009" in content
|
|
assert f"{reverse('audit:list')}?keyword=产品C" in content
|
|
assert "已发送" in content
|
|
assert "待复核" in content
|
|
|
|
|
|
def test_audit_detail_page_shows_export_summary_and_notification_receipt(client, db):
|
|
Conversation.objects.create(
|
|
conversation_id="conv-002",
|
|
title="产品B",
|
|
product_name="产品B",
|
|
batch_id="SUB-20260604-002",
|
|
task_status="success",
|
|
node_results=[
|
|
{"label": "Word 回填导出", "status": "待复核"},
|
|
{"label": "飞书通知", "status": "已完成"},
|
|
],
|
|
)
|
|
log = create_audit_log(
|
|
"document_review",
|
|
"注册审核智能体",
|
|
"导出任务",
|
|
AgentResult(
|
|
answer="已生成导出草稿",
|
|
status="success",
|
|
structured_output={
|
|
"export_status": "draft_only",
|
|
"download_url": "/downloads/registration-report.docx",
|
|
"blocked_items": ["风险项未清零"],
|
|
},
|
|
),
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
)
|
|
create_notification_record(
|
|
batch_id="SUB-20260604-002",
|
|
conversation_id="conv-002",
|
|
product_name="产品B",
|
|
trigger_source="word_export",
|
|
notify_reason="task_completed",
|
|
owner_role="注册资料负责人",
|
|
feishu_user_id="ou_demo_9",
|
|
message_status="sent",
|
|
web_detail_url="https://example.com/detail/9",
|
|
receipt={"message_id": "msg-9", "status": "sent"},
|
|
)
|
|
|
|
response = client.get(reverse("audit:detail", args=[log.id]))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "导出状态摘要" in content
|
|
assert "待复核" in content
|
|
assert "/downloads/registration-report.docx" in content
|
|
assert "风险项未清零" in content
|
|
assert "通知回执" in content
|
|
assert "msg-9" in content
|
|
assert "飞书通知 / 已发送" in content
|
|
assert "已发送" in content
|
|
|
|
|
|
def test_audit_detail_page_shows_export_file_name_and_mode(client, db):
|
|
Conversation.objects.create(
|
|
conversation_id="conv-003",
|
|
title="产品C",
|
|
product_name="产品C",
|
|
batch_id="SUB-20260604-003",
|
|
task_status="success",
|
|
node_results=[
|
|
{"label": "Word 回填导出", "status": "待复核"},
|
|
{"label": "飞书通知", "status": "已完成"},
|
|
],
|
|
)
|
|
log = create_audit_log(
|
|
"document_review",
|
|
"Word 回填导出",
|
|
"导出任务",
|
|
AgentResult(
|
|
answer="已生成导出草稿",
|
|
status="success",
|
|
structured_output={
|
|
"export_status": "draft_only",
|
|
"template_name": "注册证导出模板",
|
|
"template_version": "V1.0",
|
|
"download_url": "/media/exports/20260604/SUB-20260604-003-draft.docx",
|
|
"output_file": {
|
|
"file_name": "SUB-20260604-003-draft.docx",
|
|
"relative_path": "exports/20260604/SUB-20260604-003-draft.docx",
|
|
"export_mode": "draft",
|
|
},
|
|
},
|
|
),
|
|
batch_id="SUB-20260604-003",
|
|
conversation_id="conv-003",
|
|
product_name="产品C",
|
|
)
|
|
|
|
response = client.get(reverse("audit:detail", args=[log.id]))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert "输出文件名" in content
|
|
assert "SUB-20260604-003-draft.docx" in content
|
|
assert "导出模式" in content
|
|
assert "draft" in content
|
|
assert "模板版本" in content
|
|
assert "V1.0" in content
|
|
assert "待复核" in content
|
|
|
|
|
|
def test_audit_list_uses_chinese_filter_labels_for_risk_and_notification_status(client, db):
|
|
response = client.get(reverse("audit:list"))
|
|
|
|
content = response.content.decode("utf-8")
|
|
assert response.status_code == 200
|
|
assert ">已阻断<" in content
|
|
assert ">待复核<" in content
|
|
assert ">已完成<" in content
|
|
assert ">已发送<" in content
|
|
assert ">失败<" in content
|