493 lines
18 KiB
Python
493 lines
18 KiB
Python
from datetime import timedelta
|
||
|
||
import pytest
|
||
from django.contrib.auth import get_user_model
|
||
from django.core.management import call_command
|
||
from django.utils import timezone
|
||
|
||
from apps.agents.models import AgentRun
|
||
from apps.agents.tests.helpers import create_deployment
|
||
from apps.knowledge.models import KnowledgeBase, KnowledgeItem
|
||
from apps.knowledge.services import rebuild_item_chunks
|
||
from apps.tickets.draft_services import (
|
||
_auto_create_ticket,
|
||
create_agent_run_for_draft,
|
||
execute_draft_analysis,
|
||
serialize_agent_progress,
|
||
)
|
||
from apps.tickets.models import (
|
||
Ticket,
|
||
TicketActorSource,
|
||
TicketAttachment,
|
||
TicketDraft,
|
||
TicketFieldEvidence,
|
||
TicketMessageTimelineItem,
|
||
)
|
||
|
||
|
||
def _publish_agent():
|
||
deployment = create_deployment()
|
||
call_command("publish_trusted_ticket_intake", deployment_id=deployment.id)
|
||
|
||
|
||
def _user():
|
||
return get_user_model().objects.create_user(username="trusted-draft-user")
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_draft_context_contains_only_published_knowledge_items():
|
||
_publish_agent()
|
||
base = KnowledgeBase.objects.create(name="采集知识库")
|
||
published = KnowledgeItem.objects.create(
|
||
knowledge_base=base,
|
||
title="采集任务停滞",
|
||
problem="采集任务长时间没有更新",
|
||
root_cause="调度状态异常",
|
||
solution="核查调度状态后重新触发",
|
||
status="published",
|
||
)
|
||
rebuild_item_chunks(published)
|
||
draft_item = KnowledgeItem.objects.create(
|
||
knowledge_base=base,
|
||
title="未审核客户专属方案",
|
||
problem="采集任务没有更新",
|
||
solution="不得进入 Agent 的方案",
|
||
status="draft",
|
||
)
|
||
rebuild_item_chunks(draft_item)
|
||
draft = TicketDraft.objects.create(
|
||
created_by=_user(), source_text="生产环境采集任务长时间没有更新"
|
||
)
|
||
|
||
run = create_agent_run_for_draft(draft)
|
||
|
||
context = run.input_payload["knowledge_context"]
|
||
assert [item["knowledge_item_id"] for item in context] == [published.id]
|
||
assert context[0]["solution"] == "核查调度状态后重新触发"
|
||
assert "不得进入 Agent 的方案" not in str(context)
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_agent_input_contains_only_scanned_parsed_materials():
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="")
|
||
ready = draft.materials.create(
|
||
file_name="service.log",
|
||
material_type="log",
|
||
extension="log",
|
||
mime_type="text/plain",
|
||
file_size=20,
|
||
sha256="1" * 64,
|
||
storage_path="ticket-drafts/service.log",
|
||
scan_status="clean",
|
||
parse_status="parsed",
|
||
parsed_text="2026-07-25 ERROR timeout",
|
||
)
|
||
draft.materials.create(
|
||
file_name="blocked.txt",
|
||
material_type="text",
|
||
extension="txt",
|
||
mime_type="text/plain",
|
||
file_size=10,
|
||
sha256="2" * 64,
|
||
storage_path="ticket-drafts/blocked.txt",
|
||
scan_status="infected",
|
||
parse_status="blocked",
|
||
)
|
||
|
||
run = create_agent_run_for_draft(draft)
|
||
|
||
assert run.input_payload["material_texts"] == [
|
||
{
|
||
"source_ref": f"material:{ready.public_id}",
|
||
"file_name": "service.log",
|
||
"mime_type": "text/plain",
|
||
"text": "2026-07-25 ERROR timeout",
|
||
}
|
||
]
|
||
assert [item["file_name"] for item in run.input_payload["material_refs"]] == [
|
||
"service.log"
|
||
]
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_analysis_rejects_unproven_entities_and_preserves_manual_fields(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(
|
||
created_by=_user(),
|
||
source_text="张三反馈生产环境采集任务没有更新",
|
||
current_fields={"submitter_org": "人工确认机构"},
|
||
last_ai_fields={"submitter_org": "旧 AI 机构"},
|
||
)
|
||
run = create_agent_run_for_draft(draft)
|
||
ai_fields = {
|
||
"title": "生产采集任务未更新",
|
||
"description": "张三反馈生产环境采集任务未更新。",
|
||
"category_code": "technical_issue",
|
||
"priority": "medium",
|
||
"problem_performance": "采集任务未更新",
|
||
"problem_object": "知识库客户企业",
|
||
"submitter_name": "知识库联系人",
|
||
"submitter_org": "知识库机构",
|
||
"impact_scope": "",
|
||
"environment": "生产环境",
|
||
"related_system": "数据采集平台",
|
||
}
|
||
|
||
def fake_execute(run_id, **_kwargs):
|
||
AgentRun.objects.filter(id=run_id).update(
|
||
status="succeeded",
|
||
final_output={
|
||
"problem_valid": True,
|
||
"invalid_reason": "",
|
||
"fields": ai_fields,
|
||
"field_evidence": {
|
||
"description": {
|
||
"confidence": 0.95,
|
||
"source_refs": ["source_text"],
|
||
"excerpt": "采集任务没有更新",
|
||
"location": "文本",
|
||
"conflicts": [],
|
||
}
|
||
},
|
||
"message_timeline": [],
|
||
"missing_fields": [],
|
||
"questions": [],
|
||
},
|
||
)
|
||
|
||
monkeypatch.setattr("apps.tickets.draft_services.execute_agent_run", fake_execute)
|
||
|
||
execute_draft_analysis(draft.id, run.id)
|
||
|
||
draft.refresh_from_db()
|
||
assert draft.status == "ticket_created"
|
||
assert draft.confirmed_ticket.status == "pending_info"
|
||
assert draft.current_fields["problem_object"] == ""
|
||
assert draft.current_fields["submitter_name"] == ""
|
||
assert draft.current_fields["submitter_org"] == "人工确认机构"
|
||
assert set(draft.missing_fields) >= {"problem_object", "submitter_name"}
|
||
assert TicketFieldEvidence.objects.get(
|
||
ticket=draft.confirmed_ticket,
|
||
field_name="submitter_org",
|
||
).is_protected is True
|
||
description_evidence = TicketFieldEvidence.objects.get(
|
||
ticket=draft.confirmed_ticket,
|
||
field_name="description",
|
||
)
|
||
assert description_evidence.source == TicketActorSource.AGENT
|
||
assert float(description_evidence.confidence) == 0.95
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_analysis_keeps_invalid_problem_as_intake_without_ticket(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="你好，在吗")
|
||
run = create_agent_run_for_draft(draft)
|
||
|
||
def fake_execute(run_id, **_kwargs):
|
||
AgentRun.objects.filter(id=run_id).update(
|
||
status="succeeded",
|
||
final_output={
|
||
"problem_valid": False,
|
||
"invalid_reason": "请补充具体问题现象",
|
||
"fields": {"description": ""},
|
||
"field_evidence": {},
|
||
"message_timeline": [],
|
||
"missing_fields": ["description"],
|
||
"questions": ["请描述需要技术支持处理的具体问题。"],
|
||
},
|
||
)
|
||
|
||
monkeypatch.setattr("apps.tickets.draft_services.execute_agent_run", fake_execute)
|
||
|
||
execute_draft_analysis(draft.id, run.id)
|
||
|
||
draft.refresh_from_db()
|
||
assert draft.status == "needs_input"
|
||
assert draft.problem_valid is False
|
||
assert draft.invalid_reason == "请补充具体问题现象"
|
||
assert draft.confirmed_ticket_id is None
|
||
assert Ticket.objects.count() == 0
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_analysis_persists_structured_context_business_times_and_timezone(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(
|
||
created_by=_user(),
|
||
source_text=(
|
||
"张三代表华南公司反馈财务系统报销模块的报销单在生产环境报错 E500，"
|
||
"2026-07-27 10:00（北京时间）开始，已清理缓存但仍未恢复，希望尽快恢复报销。"
|
||
),
|
||
)
|
||
run = create_agent_run_for_draft(draft)
|
||
fields = {
|
||
"title": "财务系统报销模块 E500",
|
||
"description": "华南公司反馈财务系统报销模块持续报错 E500。",
|
||
"category_code": "technical_issue",
|
||
"priority": "medium",
|
||
"problem_performance": "提交报销时持续报错 E500",
|
||
"problem_object": "报销单",
|
||
"submitter_name": "张三",
|
||
"submitter_org": "华南公司",
|
||
"impact_scope": "华南公司报销用户",
|
||
"environment": "生产环境",
|
||
"related_system": "财务系统",
|
||
"product_name": "财务系统",
|
||
"module_name": "报销模块",
|
||
"tags": ["E500", "报销", "E500"],
|
||
"attempted_actions": "已清理缓存",
|
||
"error_information": "E500",
|
||
"customer_request": "尽快恢复报销",
|
||
"current_result": "清理缓存后仍未恢复",
|
||
"problem_started_at": "2026-07-27T10:00:00+08:00",
|
||
"customer_last_message_at": "2026-07-27T10:05:00+08:00",
|
||
"support_first_response_at": "2026-07-27T10:02:00+08:00",
|
||
"support_last_reply_at": "2026-07-27T10:04:00+08:00",
|
||
}
|
||
|
||
def fake_execute(run_id, **_kwargs):
|
||
AgentRun.objects.filter(id=run_id).update(
|
||
status="succeeded",
|
||
final_output={
|
||
"problem_valid": True,
|
||
"invalid_reason": "",
|
||
"fields": fields,
|
||
"field_evidence": {
|
||
"problem_started_at": {
|
||
"confidence": 0.96,
|
||
"source_refs": ["source_text"],
|
||
"excerpt": "2026-07-27 10:00（北京时间）开始",
|
||
"location": "文本第 1 行",
|
||
"raw_value": "2026-07-27 10:00",
|
||
"timezone": "Asia/Shanghai",
|
||
"conflicts": [],
|
||
}
|
||
},
|
||
"message_timeline": [
|
||
{
|
||
"speaker": "张三",
|
||
"role": "customer",
|
||
"content": "报销模块报错 E500",
|
||
"occurred_at": "2026-07-27T10:00:00+08:00",
|
||
"occurred_at_raw": "上午十点",
|
||
"timezone": "Asia/Shanghai",
|
||
"source_ref": "source_text",
|
||
"confidence": 0.92,
|
||
"conflict": "",
|
||
}
|
||
],
|
||
"missing_fields": [],
|
||
"questions": [],
|
||
},
|
||
)
|
||
|
||
monkeypatch.setattr("apps.tickets.draft_services.execute_agent_run", fake_execute)
|
||
|
||
execute_draft_analysis(draft.id, run.id)
|
||
|
||
draft.refresh_from_db()
|
||
ticket = Ticket.objects.get(id=draft.confirmed_ticket_id)
|
||
assert ticket.status == "pending"
|
||
assert ticket.product_name == "财务系统"
|
||
assert ticket.module_name == "报销模块"
|
||
assert ticket.tags == ["E500", "报销"]
|
||
assert ticket.attempted_actions == "已清理缓存"
|
||
assert ticket.error_information == "E500"
|
||
assert ticket.customer_request == "尽快恢复报销"
|
||
assert ticket.current_result == "清理缓存后仍未恢复"
|
||
assert ticket.problem_started_at.isoformat() == "2026-07-27T02:00:00+00:00"
|
||
time_evidence = TicketFieldEvidence.objects.get(
|
||
ticket=ticket,
|
||
field_name="problem_started_at",
|
||
)
|
||
assert time_evidence.evidence["raw_value"] == "2026-07-27 10:00"
|
||
assert time_evidence.evidence["timezone"] == "Asia/Shanghai"
|
||
timeline = TicketMessageTimelineItem.objects.get(ticket=ticket)
|
||
assert timeline.occurred_at_raw == "上午十点"
|
||
assert timeline.timezone == "Asia/Shanghai"
|
||
assert timeline.occurred_at.isoformat() == "2026-07-27T02:00:00+00:00"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_analysis_rejects_business_time_without_explicit_timezone(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(
|
||
created_by=_user(),
|
||
source_text="张三代表华南公司反馈生产环境报销单无法提交。",
|
||
)
|
||
run = create_agent_run_for_draft(draft)
|
||
|
||
def fake_execute(run_id, **_kwargs):
|
||
AgentRun.objects.filter(id=run_id).update(
|
||
status="succeeded",
|
||
final_output={
|
||
"problem_valid": True,
|
||
"invalid_reason": "",
|
||
"fields": {
|
||
"title": "报销单无法提交",
|
||
"description": "华南公司反馈生产环境报销单无法提交。",
|
||
"category_code": "technical_issue",
|
||
"priority": "medium",
|
||
"problem_performance": "报销单无法提交",
|
||
"problem_object": "报销单",
|
||
"submitter_name": "张三",
|
||
"submitter_org": "华南公司",
|
||
"environment": "生产环境",
|
||
"problem_started_at": "2026-07-27T10:00:00",
|
||
},
|
||
"field_evidence": {},
|
||
"message_timeline": [],
|
||
"missing_fields": [],
|
||
"questions": [],
|
||
},
|
||
)
|
||
|
||
monkeypatch.setattr("apps.tickets.draft_services.execute_agent_run", fake_execute)
|
||
|
||
execute_draft_analysis(draft.id, run.id)
|
||
|
||
draft.refresh_from_db()
|
||
ticket = Ticket.objects.get(id=draft.confirmed_ticket_id)
|
||
assert ticket.problem_started_at is None
|
||
assert ticket.status == "pending_info"
|
||
assert "problem_started_at" in draft.missing_fields
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_auto_create_is_idempotent_for_same_intake(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="采集任务没有更新")
|
||
run = create_agent_run_for_draft(draft)
|
||
fields = {
|
||
"description": "采集任务从上午十点起没有更新",
|
||
"priority": "medium",
|
||
}
|
||
|
||
first = _auto_create_ticket(
|
||
draft.id,
|
||
run,
|
||
fields=fields,
|
||
field_evidence={},
|
||
confidence_by_field={},
|
||
human_field_names=set(),
|
||
)
|
||
TicketDraft.objects.filter(id=draft.id).update(status="needs_input")
|
||
second = _auto_create_ticket(
|
||
draft.id,
|
||
run,
|
||
fields=fields,
|
||
field_evidence={},
|
||
confidence_by_field={},
|
||
human_field_names=set(),
|
||
)
|
||
|
||
assert first.confirmed_ticket_id == second.confirmed_ticket_id
|
||
assert second.status == "ticket_created"
|
||
assert Ticket.objects.count() == 1
|
||
ticket = Ticket.objects.get()
|
||
assert ticket.status == "pending_info"
|
||
assert ticket.title.startswith("待补全问题：")
|
||
monkeypatch.setattr(
|
||
"apps.tickets.draft_services.execute_agent_run",
|
||
lambda *_args, **_kwargs: pytest.fail("duplicate run must not call the model"),
|
||
)
|
||
execute_draft_analysis(draft.id, run.id)
|
||
run.refresh_from_db()
|
||
assert run.status == "succeeded"
|
||
assert Ticket.objects.count() == 1
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_auto_create_rolls_back_ticket_when_attachment_copy_fails(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="采集任务没有更新")
|
||
draft.materials.create(
|
||
file_name="proof.png",
|
||
mime_type="image/png",
|
||
file_size=10,
|
||
sha256="0" * 64,
|
||
storage_path="ticket-drafts/proof.png",
|
||
)
|
||
run = create_agent_run_for_draft(draft)
|
||
monkeypatch.setattr(
|
||
"apps.tickets.draft_services.copy_object",
|
||
lambda *_args: (_ for _ in ()).throw(RuntimeError("storage unavailable")),
|
||
)
|
||
|
||
with pytest.raises(RuntimeError, match="storage unavailable"):
|
||
_auto_create_ticket(
|
||
draft.id,
|
||
run,
|
||
fields={"description": "采集任务没有更新"},
|
||
field_evidence={},
|
||
confidence_by_field={},
|
||
human_field_names=set(),
|
||
)
|
||
|
||
assert Ticket.objects.count() == 0
|
||
draft.refresh_from_db()
|
||
assert draft.confirmed_ticket_id is None
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_auto_create_preserves_material_scan_and_parse_metadata(monkeypatch):
|
||
_publish_agent()
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="日志显示请求超时")
|
||
material = draft.materials.create(
|
||
file_name="service.log",
|
||
material_type="log",
|
||
extension="log",
|
||
mime_type="text/plain",
|
||
file_size=24,
|
||
sha256="a" * 64,
|
||
storage_path="ticket-drafts/service.log",
|
||
scan_status="clean",
|
||
scan_result={"engine": "clamav", "status": "clean"},
|
||
parse_status="parsed",
|
||
parsed_text="ERROR upstream timeout",
|
||
)
|
||
run = create_agent_run_for_draft(draft)
|
||
monkeypatch.setattr("apps.tickets.draft_services.copy_object", lambda *_args: None)
|
||
|
||
result = _auto_create_ticket(
|
||
draft.id,
|
||
run,
|
||
fields={"description": "日志显示上游请求超时"},
|
||
field_evidence={},
|
||
confidence_by_field={},
|
||
human_field_names=set(),
|
||
)
|
||
|
||
attachment = TicketAttachment.objects.get(ticket=result.confirmed_ticket)
|
||
assert attachment.file_type == "log"
|
||
assert attachment.sha256 == "a" * 64
|
||
assert attachment.scan_status == "clean"
|
||
assert attachment.parse_status == "parsed"
|
||
assert attachment.parsed_text == "ERROR upstream timeout"
|
||
material.refresh_from_db()
|
||
assert material.status == "attached"
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_agent_progress_marks_long_running_draft_as_slow(settings):
|
||
_publish_agent()
|
||
settings.AI_SLOW_RUN_THRESHOLD_SECONDS = 20
|
||
draft = TicketDraft.objects.create(created_by=_user(), source_text="采集任务没有更新")
|
||
run = create_agent_run_for_draft(draft)
|
||
run.status = "running"
|
||
run.started_at = timezone.now() - timedelta(seconds=21)
|
||
run.progress_stage = "extracting"
|
||
run.progress_message = "正在解析沟通记录"
|
||
run.save(
|
||
update_fields=["status", "started_at", "progress_stage", "progress_message"]
|
||
)
|
||
|
||
progress = serialize_agent_progress(draft)
|
||
|
||
assert progress["stage"] == "extracting"
|
||
assert progress["message"] == "正在解析沟通记录"
|
||
assert progress["elapsed_ms"] >= 20_000
|
||
assert progress["slow"] is True
|