HaiLin/ai-ticket-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: complete multimodal ticket intake

Browse Source
This commit is contained in:
hailin
2026-07-27 18:59:28 +08:00
parent 96452934a3
commit 93065c0a00
39 changed files with 6443 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

341
backend/apps/tickets/tests/test_stage_2_3.py Normal file
View File

@@ -0,0 +1,341 @@
import pytest
from django.contrib.auth import get_user_model
from django.core.management import call_command
from apps.agents.models import AgentRun
from apps.agents.services import append_run_event, build_config_snapshot
from apps.agents.tests.helpers import create_agent, create_deployment
from apps.tickets.draft_services import (
_mark_analysis_failed,
_replace_message_timeline,
create_agent_run_for_draft,
execute_draft_analysis,
)
from apps.tickets.models import TicketCategory, TicketDraft
from apps.tickets.services import update_ticket_fields
from apps.workitems.models import WorkItem
def publish_intake_agent():
deployment = create_deployment()
call_command("publish_trusted_ticket_intake", deployment_id=deployment.id)
@pytest.mark.django_db
def test_intake_persists_timeline_creates_unique_task_and_orders_events(monkeypatch):
publish_intake_agent()
user = get_user_model().objects.create_user(username="stage-23-owner")
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": {
"description": "客户反馈生产环境从十点开始无法登录",
"priority": "medium",
},
"field_evidence": {},
"message_timeline": [
{
"speaker": "客户",
"role": "customer",
"content": "生产环境从十点开始无法登录",
"occurred_at": "2026-07-27T10:00:00+08:00",
"source_ref": "source_text",
"confidence": 0.96,
"conflict": "",
}
],
"missing_fields": ["category_code", "environment"],
"questions": ["请确认受影响机构。"],
},
)
monkeypatch.setattr("apps.tickets.draft_services.execute_agent_run", fake_execute)
execute_draft_analysis(draft.id, run.id)
execute_draft_analysis(draft.id, run.id)
draft.refresh_from_db()
ticket = draft.confirmed_ticket
timeline = draft.timeline_items.get()
assert timeline.ticket == ticket
assert timeline.role == "customer"
assert timeline.occurred_at is not None
assert float(timeline.confidence) == 0.96
assert WorkItem.objects.filter(
ticket=ticket,
task_type="ticket_missing_info",
status="active",
).count() == 1
task = WorkItem.objects.get(ticket=ticket, status="active")
assert task.payload["questions"] == ["请确认受影响机构。"]
assert task.payload["draft_public_id"] == str(draft.public_id)
event_types = list(run.events.values_list("event_type", flat=True))
required = ["queued", "parsing", "extracting", "creating_ticket", "completed"]
positions = [event_types.index(event_type) for event_type in required]
assert positions == sorted(positions)
assert event_types.count("completed") == 1
@pytest.mark.django_db
def test_missing_info_task_completes_after_required_fields_are_filled():
publish_intake_agent()
user = get_user_model().objects.create_user(username="stage-23-editor")
draft = TicketDraft.objects.create(created_by=user, source_text="无法登录")
run = create_agent_run_for_draft(draft)
from apps.tickets.draft_services import _auto_create_ticket
created = _auto_create_ticket(
draft.id,
run,
fields={"description": "无法登录", "priority": "medium"},
field_evidence={},
confidence_by_field={},
human_field_names=set(),
)
ticket = created.confirmed_ticket
assert WorkItem.objects.get(ticket=ticket).status == "active"
category, _ = TicketCategory.objects.get_or_create(
code="technical_issue",
defaults={"name": "技术问题"},
)
update_ticket_fields(
ticket,
{
"title": "生产环境无法登录",
"category_code": category.code,
"problem_performance": "登录提示服务异常",
"problem_object": "统一身份平台",
"submitter_name": "张三",
"submitter_org": "客户机构",
"environment": "生产环境",
},
operator=user,
)
task = WorkItem.objects.get(ticket=ticket)
ticket.refresh_from_db()
assert task.status == "completed"
assert task.completed_at is not None
assert task.completed_by == user
assert ticket.status == "pending"
assert ticket.events.filter(
event_type="status_change",
from_value__status="pending_info",
to_value__status="pending",
source="system",
).exists()
@pytest.mark.django_db
def test_timeline_replacement_keeps_sequences_unique():
user = get_user_model().objects.create_user(username="stage-23-timeline")
draft = TicketDraft.objects.create(created_by=user, source_text="问题")
_replace_message_timeline(
draft,
[
{
"sequence": 1,
"speaker": "客户",
"role": "customer",
"content": "第一次描述",
"occurred_at": "",
"source_ref": "source_text",
"confidence": 0.8,
"conflict": "",
}
],
)
_replace_message_timeline(
draft,
[
{
"sequence": 1,
"speaker": "支持",
"role": "support",
"content": "补充提问",
"occurred_at": "",
"source_ref": "source_text",
"confidence": None,
"conflict": "",
}
],
)
assert draft.timeline_items.count() == 1
assert draft.timeline_items.get().content == "补充提问"
@pytest.mark.django_db
def test_sse_supports_cursor_and_enforces_run_ownership(client):
owner = get_user_model().objects.create_user(username="stage-23-sse-owner")
outsider = get_user_model().objects.create_user(username="stage-23-sse-outsider")
_, version = create_agent(two_steps=False)
run = AgentRun.objects.create(
definition=version.definition,
agent_version=version,
input_payload={"problem": "无法登录"},
config_snapshot=build_config_snapshot(version),
created_by=owner,
)
append_run_event(run, "queued")
append_run_event(run, "parsing")
append_run_event(run, "completed")
client.force_login(owner)
response = client.get(f"/api/agent-runs/{run.public_id}/events?after=1")
assert response.status_code == 200
body = b"".join(response.streaming_content).decode()
assert "id: 1\n" not in body
assert "id: 2\n" in body
assert "id: 3\n" in body
assert response["Cache-Control"] == "no-cache"
response = client.get(
f"/api/agent-runs/{run.public_id}/events",
HTTP_LAST_EVENT_ID="2",
)
body = b"".join(response.streaming_content).decode()
assert "id: 2\n" not in body
assert "id: 3\n" in body
client.force_login(outsider)
response = client.get(f"/api/agent-runs/{run.public_id}/events")
assert response.status_code == 403
@pytest.mark.django_db
def test_task_api_filters_by_ticket(admin_client):
publish_intake_agent()
user = get_user_model().objects.create_user(username="stage-23-task-api")
draft = TicketDraft.objects.create(created_by=user, source_text="无法登录")
run = create_agent_run_for_draft(draft)
from apps.tickets.draft_services import _auto_create_ticket
created = _auto_create_ticket(
draft.id,
run,
fields={"description": "无法登录"},
field_evidence={},
confidence_by_field={},
human_field_names=set(),
)
response = admin_client.get(f"/api/tasks?ticket_id={created.confirmed_ticket_id}")
assert response.status_code == 200
assert len(response.json()) == 1
assert response.json()[0]["task_type"] == "ticket_missing_info"
@pytest.mark.django_db
def test_failed_intake_creates_one_takeover_task_and_retry_completes_it():
publish_intake_agent()
user = get_user_model().objects.create_user(username="stage-23-takeover")
draft = TicketDraft.objects.create(created_by=user, source_text="生产环境无法登录")
run = create_agent_run_for_draft(draft)
_mark_analysis_failed(
draft,
error_code="model_unavailable",
error_message="模型服务暂不可用",
run=run,
)
draft.refresh_from_db()
_mark_analysis_failed(
draft,
error_code="model_unavailable",
error_message="模型服务暂不可用",
run=run,
)
task = WorkItem.objects.get(
task_type="ticket_intake_failed",
object_type="ticket_draft",
object_id=str(draft.public_id),
)
assert task.status == "active"
assert task.payload["agent_run_public_id"] == str(run.public_id)
assert task.payload["error_code"] == "model_unavailable"
assert task.assignee == user
draft.refresh_from_db()
retry_run = create_agent_run_for_draft(draft)
task.refresh_from_db()
draft.refresh_from_db()
assert task.status == "completed"
assert task.completed_by == user
assert draft.status == "analyzing"
assert draft.latest_agent_run == retry_run
@pytest.mark.django_db
def test_cancel_ticket_draft_run_preserves_materials_and_cancelled_semantics(
client,
django_capture_on_commit_callbacks,
monkeypatch,
):
publish_intake_agent()
user = get_user_model().objects.create_superuser(
username="stage-23-cancel",
email="cancel@example.com",
password="test-password",
)
draft = TicketDraft.objects.create(
created_by=user,
source_text="生产环境无法登录",
)
run = create_agent_run_for_draft(draft)
run.task_id = "celery-task-1"
run.save(update_fields=["task_id"])
revoked = []
class FakeAsyncResult:
def __init__(self, task_id):
self.task_id = task_id
def revoke(self, *, terminate):
revoked.append((self.task_id, terminate))
monkeypatch.setattr(
"apps.tickets.draft_services.AsyncResult",
FakeAsyncResult,
)
client.force_login(user)
with django_capture_on_commit_callbacks(execute=True):
response = client.post(
f"/api/ticket-drafts/{draft.public_id}/cancel",
data={"expected_revision": draft.revision},
content_type="application/json",
)
assert response.status_code == 200
assert response.json()["status"] == "failed"
assert response.json()["error_code"] == "analysis_cancelled"
draft.refresh_from_db()
run.refresh_from_db()
assert run.status == "cancelled"
assert run.error_code == ""
assert revoked == [("celery-task-1", False)]
assert run.events.filter(event_type="cancelled").count() == 1
assert WorkItem.objects.filter(
task_type="ticket_intake_failed",
object_id=str(draft.public_id),
status="active",
).count() == 1
execute_draft_analysis(draft.id, run.id)
run.refresh_from_db()
assert run.status == "cancelled"
assert run.events.filter(event_type="cancelled").count() == 1

516
backend/apps/tickets/tests/test_stage_2_4.py Normal file
View File

@@ -0,0 +1,516 @@
import json
from datetime import timedelta
from io import StringIO
from pathlib import Path
import pytest
from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.core.management.base import CommandError
from django.utils import timezone
from apps.tickets.evaluation import (
IntakeEvaluationError,
build_intake_quality_metrics,
build_stage_2_acceptance_report,
import_evaluation_cases,
record_evaluation_result,
)
from apps.tickets.models import (
IntakeEvaluationCase,
IntakeEvaluationResult,
Ticket,
TicketDraft,
TicketSimilaritySuggestion,
)
from apps.tickets.similarity import find_similar_tickets, refresh_ticket_similarity_suggestions
@pytest.mark.django_db
def test_similar_ticket_candidates_are_scored_and_do_not_include_self():
historical = Ticket.objects.create(
ticket_no="TK202607270101",
title="生产环境数据同步任务停止更新",
description="同步任务报错 SYNC-504，进度长时间不更新。",
status="resolved",
)
Ticket.objects.create(
ticket_no="TK202607270102",
title="用户咨询报表下载入口",
description="需要确认报表下载菜单位置。",
)
current = Ticket.objects.create(
ticket_no="TK202607270103",
title="生产数据同步停止更新",
description="生产环境同步任务出现 SYNC-504 后没有新进度。",
)
candidates = find_similar_tickets(current)
assert candidates[0]["ticket"] == historical
assert candidates[0]["score"] > 0.5
assert all(item["ticket"].id != current.id for item in candidates)
assert "SYNC-504" in candidates[0]["matched_excerpt"]
@pytest.mark.django_db
def test_duplicate_problem_creates_independent_ticket_and_similarity_snapshot(admin_client):
first = Ticket.objects.create(
ticket_no="TK202607270111",
title="回调任务持续超时",
description="生产环境回调任务持续超时，错误码 CALLBACK-500。",
status="resolved",
)
duplicate = Ticket.objects.create(
ticket_no="TK202607270112",
title="回调任务持续超时",
description="生产环境回调任务仍然超时，错误码 CALLBACK-500。",
)
suggestions = refresh_ticket_similarity_suggestions(duplicate)
assert first.id != duplicate.id
assert Ticket.objects.count() == 2
assert suggestions[0].candidate_ticket == first
assert TicketSimilaritySuggestion.objects.filter(
source_ticket=duplicate,
candidate_ticket=first,
).exists()
response = admin_client.get(f"/api/tickets/{duplicate.id}/similar")
assert response.status_code == 200
assert response.json()[0]["ticket_id"] == first.id
assert response.json()[0]["score"] > 0
@pytest.mark.django_db
def test_real_case_metrics_use_weighted_field_accuracy_and_exclude_synthetic(admin_client):
user = get_user_model().objects.create_user(username="stage-24-reviewer")
ticket = Ticket.objects.create(
ticket_no="TK202607270121",
title="生产环境登录异常",
description="统一登录提示服务异常",
)
draft = TicketDraft.objects.create(
created_by=user,
source_text="生产环境登录异常",
problem_valid=True,
current_fields={
"title": "生产环境登录异常",
"description": "统一登录提示服务异常",
"environment": "生产环境",
},
decision="auto_created",
confirmed_ticket=ticket,
)
real_case = IntakeEvaluationCase.objects.create(
key="real-login-001",
name="脱敏登录异常",
source_kind="anonymized_real",
expected_fields={
"title": "生产环境登录异常",
"description": "统一登录提示服务异常",
"environment": "测试环境",
},
expected_auto_created=True,
)
synthetic_case = IntakeEvaluationCase.objects.create(
key="synthetic-unused-001",
name="合成样例",
source_kind="synthetic",
expected_fields={"title": "不计入默认指标"},
)
result = record_evaluation_result(real_case, draft, reviewer=user)
record_evaluation_result(synthetic_case, draft, reviewer=user)
metrics = build_intake_quality_metrics()
assert result.correct_field_count == 2
assert result.evaluated_field_count == 3
assert float(result.field_accuracy) == pytest.approx(2 / 3, rel=1e-4)
assert metrics["evaluation"]["enabled_case_count"] == 1
assert metrics["evaluation"]["reviewed_case_count"] == 1
assert metrics["evaluation"]["auto_creation_rate"] == 1.0
assert metrics["evaluation"]["field_accuracy"] == pytest.approx(2 / 3)
assert metrics["evaluation"]["evidence_coverage_rate"] == 0
assert metrics["targets"]["coverage_complete"] is True
assert metrics["targets"]["auto_creation_rate_met"] is True
assert metrics["targets"]["field_accuracy_met"] is False
response = admin_client.get("/api/tickets/intake-quality-metrics")
assert response.status_code == 200
assert response.json()["include_synthetic"] is False
@pytest.mark.django_db
def test_metrics_do_not_claim_success_without_real_case_results():
IntakeEvaluationCase.objects.create(
key="real-not-run-001",
name="尚未执行的真实案例",
source_kind="anonymized_real",
expected_fields={"description": "待执行"},
)
metrics = build_intake_quality_metrics()
assert metrics["evaluation"]["coverage_rate"] == 0
assert metrics["evaluation"]["auto_creation_rate"] is None
assert metrics["evaluation"]["field_accuracy"] is None
assert metrics["targets"]["coverage_complete"] is False
assert metrics["targets"]["auto_creation_rate_met"] is None
assert metrics["targets"]["field_accuracy_met"] is None
@pytest.mark.django_db
def test_sample_case_file_is_explicitly_synthetic():
path = Path(settings.BASE_DIR) / "docs/product/acceptance/stage2-sample-cases.json"
created, updated = import_evaluation_cases(path)
assert created == 6
assert updated == 0
assert IntakeEvaluationCase.objects.count() == 6
assert not IntakeEvaluationCase.objects.exclude(source_kind="synthetic").exists()
@pytest.mark.django_db
def test_real_case_template_is_valid_but_disabled():
path = Path(settings.BASE_DIR) / "docs/product/acceptance/stage2-real-cases.template.json"
created, updated = import_evaluation_cases(path)
assert created == 4
assert updated == 0
assert IntakeEvaluationCase.objects.filter(source_kind="anonymized_real").count() == 4
assert not IntakeEvaluationCase.objects.filter(enabled=True).exists()
report = build_stage_2_acceptance_report()
assert report["status"] == "not_ready"
assert report["reasons"][0]["code"] == "missing_real_cases"
@pytest.mark.django_db
def test_real_case_import_rejects_missing_or_non_boolean_acceptance_fields(tmp_path):
path = tmp_path / "invalid-real-cases.json"
path.write_text(
json.dumps(
{
"schema_version": 1,
"cases": [
{
"key": "real-invalid-schema-001",
"source_kind": "anonymized_real",
"valid_input": "false",
"input_manifest": {"source_text": "脱敏问题", "materials": []},
"expected_fields": {},
"expected_timeline": [],
"expected_auto_created": False,
}
],
},
ensure_ascii=False,
),
encoding="utf-8",
)
with pytest.raises(IntakeEvaluationError, match="布尔字段类型错误"):
import_evaluation_cases(path)
payload = json.loads(path.read_text(encoding="utf-8"))
payload["cases"][0]["valid_input"] = False
path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
with pytest.raises(IntakeEvaluationError, match="expected_similar_ticket"):
import_evaluation_cases(path)
@pytest.mark.django_db
def test_evaluation_audits_timeline_evidence_and_requires_full_case_coverage():
user = get_user_model().objects.create_user(username="stage-24-evidence-reviewer")
ticket = Ticket.objects.create(
ticket_no="TK202607270131",
title="订单接口超时",
description="订单接口返回 TIMEOUT-504",
)
draft = TicketDraft.objects.create(
created_by=user,
source_text="10:00 客户反馈订单接口返回 TIMEOUT-504",
problem_valid=True,
current_fields={
"description": "订单接口返回 TIMEOUT-504",
"environment": "生产环境",
},
field_evidence={
"description": {
"source_refs": ["source_text"],
"excerpt": "订单接口返回 TIMEOUT-504",
},
"environment": {
"source_refs": ["source_text"],
"excerpt": "生产环境",
},
},
message_timeline=[
{
"sequence": 1,
"role": "customer",
"content": "订单接口返回 TIMEOUT-504",
"occurred_at_raw": "10:00",
"timezone": "",
}
],
decision="auto_created",
confirmed_ticket=ticket,
)
reviewed_case = IntakeEvaluationCase.objects.create(
key="real-evidence-001",
name="脱敏订单超时",
source_kind="anonymized_real",
expected_fields={
"description": "订单接口返回 TIMEOUT-504",
"environment": "生产环境",
},
expected_timeline=[
{
"role": "customer",
"content": "订单接口返回 TIMEOUT-504",
"occurred_at_raw": "10:00",
}
],
)
IntakeEvaluationCase.objects.create(
key="real-evidence-pending-002",
name="待评审案例",
source_kind="anonymized_real",
)
result = record_evaluation_result(reviewed_case, draft, reviewer=user)
metrics = build_intake_quality_metrics()
assert result.passed is True
assert result.details["timeline_correct"] is True
assert result.details["evidence_covered_count"] == 2
assert metrics["evaluation"]["evidence_coverage_rate"] == 1.0
assert metrics["evaluation"]["timeline_accuracy"] == 1.0
assert metrics["evaluation"]["coverage_rate"] == 0.5
assert metrics["targets"]["coverage_complete"] is False
assert metrics["targets"]["auto_creation_rate_met"] is None
assert metrics["targets"]["field_accuracy_met"] is None
@pytest.mark.django_db
def test_metrics_only_use_latest_reviewed_run_for_each_case():
user = get_user_model().objects.create_user(username="stage-24-latest-run")
case = IntakeEvaluationCase.objects.create(
key="real-latest-001",
name="最新运行口径",
source_kind="anonymized_real",
expected_fields={"description": "订单同步失败"},
)
old_ticket = Ticket.objects.create(
ticket_no="TK202607270141",
title="旧运行",
description="错误结果",
)
old_draft = TicketDraft.objects.create(
created_by=user,
problem_valid=True,
current_fields={"description": "错误结果"},
decision="auto_created",
confirmed_ticket=old_ticket,
)
record_evaluation_result(case, old_draft, reviewer=user)
new_ticket = Ticket.objects.create(
ticket_no="TK202607270142",
title="新运行",
description="订单同步失败",
)
new_draft = TicketDraft.objects.create(
created_by=user,
problem_valid=True,
current_fields={"description": "订单同步失败"},
field_evidence={
"description": {
"source_refs": ["source_text"],
"excerpt": "订单同步失败",
}
},
decision="auto_created",
confirmed_ticket=new_ticket,
)
record_evaluation_result(case, new_draft, reviewer=user)
metrics = build_intake_quality_metrics()
assert metrics["evaluation"]["reviewed_case_count"] == 1
assert metrics["evaluation"]["correct_field_count"] == 1
assert metrics["evaluation"]["evaluated_field_count"] == 1
assert metrics["evaluation"]["field_accuracy"] == 1.0
assert metrics["evaluation"]["result_selection"] == "latest_per_case"
@pytest.mark.django_db
def test_rereviewed_draft_becomes_latest_case_result():
user = get_user_model().objects.create_user(username="stage-24-rereview")
case = IntakeEvaluationCase.objects.create(
key="real-rereview-001",
name="重新评审口径",
source_kind="anonymized_real",
expected_fields={"description": "订单同步失败"},
)
old_draft = TicketDraft.objects.create(
created_by=user,
problem_valid=True,
current_fields={"description": "旧错误结果"},
)
old_result = record_evaluation_result(case, old_draft, reviewer=user)
new_draft = TicketDraft.objects.create(
created_by=user,
problem_valid=True,
current_fields={"description": "新错误结果"},
)
new_result = record_evaluation_result(case, new_draft, reviewer=user)
stale_time = timezone.now() - timedelta(minutes=1)
IntakeEvaluationResult.objects.filter(pk__in=[old_result.pk, new_result.pk]).update(
updated_at=stale_time
)
old_draft.current_fields = {"description": "订单同步失败"}
old_draft.field_evidence = {
"description": {
"source_refs": ["source_text"],
"excerpt": "订单同步失败",
}
}
old_draft.save(update_fields=["current_fields", "field_evidence", "updated_at"])
rereviewed = record_evaluation_result(case, old_draft, reviewer=user)
metrics = build_intake_quality_metrics()
assert rereviewed.pk == old_result.pk
assert rereviewed.updated_at > stale_time
assert metrics["evaluation"]["reviewed_case_count"] == 1
assert metrics["evaluation"]["correct_field_count"] == 1
assert metrics["evaluation"]["field_accuracy"] == 1.0
def _record_passing_report_case(
*,
user,
key,
manifest,
valid_input=True,
):
expected_fields = {"description": f"{key} 问题"} if valid_input else {}
expected_timeline = (
[{"role": "customer", "content": f"{key} 问题"}] if valid_input else []
)
case = IntakeEvaluationCase.objects.create(
key=key,
name=key,
source_kind="anonymized_real",
valid_input=valid_input,
input_manifest=manifest,
expected_fields=expected_fields,
expected_timeline=expected_timeline,
expected_auto_created=valid_input,
)
ticket = None
if valid_input:
ticket = Ticket.objects.create(
ticket_no=f"TK-{key}",
title=key,
description=f"{key} 问题",
)
draft = TicketDraft.objects.create(
created_by=user,
source_text=str(manifest.get("source_text") or ""),
problem_valid=valid_input,
current_fields=expected_fields,
field_evidence=(
{
"description": {
"source_refs": ["source_text"],
"excerpt": f"{key} 问题",
}
}
if valid_input
else {}
),
message_timeline=(
[{"role": "customer", "content": f"{key} 问题"}]
if valid_input
else []
),
decision="auto_created" if valid_input else "",
confirmed_ticket=ticket,
)
record_evaluation_result(case, draft, reviewer=user)
@pytest.mark.django_db
def test_stage_2_report_requires_real_modality_coverage_and_can_pass():
user = get_user_model().objects.create_user(username="stage-24-report")
_record_passing_report_case(
user=user,
key="real-image-001",
manifest={"source_text": "", "materials": [{"file_name": "one.png", "type": "png"}]},
)
_record_passing_report_case(
user=user,
key="real-mixed-002",
manifest={
"source_text": "补充说明",
"materials": [
{"file_name": "one.png", "type": "png"},
{"file_name": "two.png", "type": "png"},
],
},
)
_record_passing_report_case(
user=user,
key="real-document-003",
manifest={
"source_text": "请查看日志",
"materials": [{"file_name": "service.log", "type": "log"}],
},
)
_record_passing_report_case(
user=user,
key="real-invalid-004",
manifest={"source_text": "你好，在吗？", "materials": []},
valid_input=False,
)
report = build_stage_2_acceptance_report()
assert report["status"] == "passed"
assert report["exit_ready"] is True
assert report["reasons"] == []
assert report["modality_coverage"]["multi_image_case_count"] == 1
assert report["metrics"]["targets"]["auto_creation_rate_met"] is True
assert report["metrics"]["targets"]["field_accuracy_met"] is True
@pytest.mark.django_db
def test_synthetic_report_is_diagnostic_and_strict_command_fails(tmp_path):
sample_path = Path(settings.BASE_DIR) / "docs/product/acceptance/stage2-sample-cases.json"
import_evaluation_cases(sample_path)
output_path = tmp_path / "stage2-report.json"
stdout = StringIO()
call_command(
"report_stage2_acceptance",
include_synthetic=True,
output_format="json",
output=str(output_path),
stdout=stdout,
)
payload = json.loads(stdout.getvalue())
assert payload["status"] == "diagnostic"
assert payload["exit_ready"] is False
assert payload["reasons"][0]["code"] == "synthetic_scope"
assert json.loads(output_path.read_text(encoding="utf-8"))["scope"] == (
"diagnostic_with_synthetic"
)
IntakeEvaluationCase.objects.all().delete()
with pytest.raises(CommandError, match="阶段 2 尚不可退出"):
call_command("report_stage2_acceptance", fail_on_not_ready=True)

492
backend/apps/tickets/tests/test_ticket_draft_trust.py Normal file
View File

@@ -0,0 +1,492 @@
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

75
backend/apps/tickets/tests/test_ticket_drafts.py
View File

@@ -1,4 +1,5 @@
import io
from types import SimpleNamespace
import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
@@ -95,6 +96,30 @@ def test_feature_requires_published_ticket_agent(admin_client, db):
assert response.json()["vision_available"] is False
def test_availability_uses_effective_step_vision_model(admin_client, db):
agent = create_ticket_agent()
version = agent.active_version
provider = version.model_deployment.provider
provider.image_egress_approved = True
provider.save(update_fields=["image_egress_approved"])
vision = ModelDeployment.objects.create(
provider=provider,
display_name="步骤视觉模型",
model_name="ticket-vision-model",
capability="vision_chat",
)
binding = version.steps.get()
binding.model_deployment = vision
binding.save(update_fields=["model_deployment"])
response = admin_client.get("/api/ticket-drafts/availability")
assert response.status_code == 200
assert response.json()["available"] is True
assert response.json()["vision_available"] is True
assert response.json()["vision_reason"] == ""
def test_revision_conflict_and_manual_edit(admin_client, db):
create_ticket_agent()
created = admin_client.post(
@@ -145,9 +170,16 @@ def test_confirm_is_idempotent_and_records_accepted(admin_client, category):
assert first.status_code == 200
assert second.status_code == 200
assert first.json()["confirmed_ticket"]["id"] == second.json()["confirmed_ticket"]["id"]
assert first.json()["created_ticket"]["id"] == first.json()["confirmed_ticket"]["id"]
assert first.json()["decision"] == "accepted"
assert Ticket.objects.count() == 1
assert Ticket.objects.get().ai_solution == ""
assert not hasattr(Ticket.objects.get(), "ai_solution")
ticket_id = first.json()["created_ticket"]["id"]
trace = admin_client.get(f"/api/tickets/{ticket_id}/intake-trace")
assert trace.status_code == 200
assert trace.json()["available"] is True
assert trace.json()["status"] == "ticket_created"
assert admin_client.get(f"/api/tickets/{ticket_id}/attachments").json() == []
def test_confirm_records_modified_and_reject_records_rejected(admin_client, category):
@@ -201,13 +233,14 @@ def test_image_upload_validates_content_and_never_audits_source_text(admin_clien
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *args: None)
output = io.BytesIO()
Image.new("RGB", (8, 8), color="red").save(output, format="PNG")
upload = SimpleUploadedFile("proof.png", output.getvalue(), content_type="text/plain")
upload = SimpleUploadedFile("proof.png", output.getvalue(), content_type="image/png")
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/images", {"file": upload}
)
assert response.status_code == 200
assert response.json()["images"][0]["mime_type"] == "image/png"
assert response.json()["materials"][0]["scan_status"] == "clean"
snapshots = list(
OperationAuditLog.objects.filter(target_id=created["public_id"]).values_list(
"request_snapshot", flat=True
@@ -224,12 +257,14 @@ def test_text_model_rejects_image_analysis(admin_client, db):
content_type="application/json",
).json()
draft = TicketDraft.objects.get(public_id=created["public_id"])
draft.images.create(
draft.materials.create(
file_name="proof.png",
mime_type="image/png",
file_size=10,
sha256="0" * 64,
storage_path="ticket-drafts/proof.png",
scan_status="clean",
parse_status="not_applicable",
)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/analyze",
@@ -240,6 +275,38 @@ def test_text_model_rejects_image_analysis(admin_client, db):
assert "不支持图片" in response.json()["detail"]
def test_vision_intake_can_start_with_image_only(admin_client, db, monkeypatch):
create_ticket_agent(capability="vision_chat", approved=True)
created = admin_client.post(
"/api/ticket-drafts",
data={"source_text": ""},
content_type="application/json",
).json()
draft = TicketDraft.objects.get(public_id=created["public_id"])
draft.materials.create(
file_name="proof.png",
mime_type="image/png",
file_size=10,
sha256="0" * 64,
storage_path="ticket-drafts/proof.png",
scan_status="clean",
parse_status="not_applicable",
)
monkeypatch.setattr(
"apps.tickets.draft_api.execute_ticket_draft_analysis.delay",
lambda *_args: SimpleNamespace(id="task-1"),
)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/analyze",
data={"expected_revision": created["revision"]},
content_type="application/json",
)
assert response.status_code == 200
assert response.json()["status"] == "analyzing"
def test_cleanup_deletes_unconfirmed_details_but_keeps_confirmed_draft(
admin_client, db, monkeypatch
):
@@ -251,7 +318,7 @@ def test_cleanup_deletes_unconfirmed_details_but_keeps_confirmed_draft(
source_text="过期敏感原文",
expires_at=timezone.now(),
)
expired.images.create(
expired.materials.create(
file_name="proof.png",
mime_type="image/png",
file_size=10,

277
backend/apps/tickets/tests/test_ticket_materials.py Normal file
View File

@@ -0,0 +1,277 @@
import io
from types import SimpleNamespace
import pytest
from django.core.files.uploadedfile import SimpleUploadedFile
from PIL import Image
from apps.tickets.materials import MaterialParseError, scan_bytes
from apps.tickets.models import TicketDraftMaterial
from apps.tickets.tests.test_ticket_drafts import create_ticket_agent
def _create_draft(admin_client):
create_ticket_agent(capability="vision_chat", approved=True)
response = admin_client.post(
"/api/ticket-drafts",
data={"source_text": "材料解析测试"},
content_type="application/json",
)
assert response.status_code == 200
return response.json()
def _docx_bytes():
from docx import Document
output = io.BytesIO()
document = Document()
document.add_paragraph("采集任务从上午十点起没有更新")
document.save(output)
return output.getvalue()
def _xlsx_bytes():
from openpyxl import Workbook
output = io.BytesIO()
workbook = Workbook()
workbook.active.append(["任务编号", "问题"])
workbook.active.append(["JOB-1024", "采集任务未更新"])
workbook.save(output)
workbook.close()
return output.getvalue()
def _pptx_bytes():
from pptx import Presentation
output = io.BytesIO()
presentation = Presentation()
slide = presentation.slides.add_slide(presentation.slide_layouts[5])
slide.shapes.title.text = "生产采集任务未更新"
presentation.save(output)
return output.getvalue()
def _pdf_bytes():
from pypdf import PdfWriter
output = io.BytesIO()
writer = PdfWriter()
writer.add_blank_page(width=72, height=72)
writer.write(output)
return output.getvalue()
@pytest.mark.parametrize(
("file_name", "content_type", "content", "material_type"),
[
("chat.txt", "text/plain", b"job failed with error 500", "text"),
("service.log", "text/x-log", b"2026-07-25 ERROR timeout", "log"),
("report.pdf", "application/pdf", _pdf_bytes(), "document"),
(
"report.docx",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
_docx_bytes(),
"document",
),
(
"report.xlsx",
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
_xlsx_bytes(),
"spreadsheet",
),
(
"report.pptx",
"application/vnd.openxmlformats-officedocument.presentationml.presentation",
_pptx_bytes(),
"presentation",
),
],
ids=["txt", "log", "pdf", "docx", "xlsx", "pptx"],
)
def test_material_upload_scans_and_parses_supported_documents(
admin_client,
db,
monkeypatch,
file_name,
content_type,
content,
material_type,
):
created = _create_draft(admin_client)
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile(file_name, content, content_type=content_type)},
)
assert response.status_code == 200
material = response.json()["materials"][0]
assert material["material_type"] == material_type
assert material["scan_status"] == "clean"
assert material["parse_status"] == "parsed"
assert TicketDraftMaterial.objects.get(public_id=material["id"]).storage_path
def test_material_rejects_disguised_extension_and_mime(admin_client, db, monkeypatch):
created = _create_draft(admin_client)
stored = []
monkeypatch.setattr(
"apps.tickets.draft_api.put_object",
lambda *args: stored.append(args),
)
signature = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("fake.pdf", b"MZ executable", content_type="application/pdf")},
)
output = io.BytesIO()
Image.new("RGB", (8, 8), color="red").save(output, format="PNG")
mime = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("proof.png", output.getvalue(), content_type="text/plain")},
)
assert signature.status_code == 415
assert mime.status_code == 415
assert stored == []
assert TicketDraftMaterial.objects.count() == 0
def test_infected_material_is_retained_and_not_parsed(admin_client, db, monkeypatch):
created = _create_draft(admin_client)
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
monkeypatch.setattr(
"apps.tickets.materials.scan_bytes",
lambda _raw: {
"status": "infected",
"engine": "clamav",
"signature": "Eicar-Test-Signature",
},
)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("sample.txt", b"EICAR sample", content_type="text/plain")},
)
assert response.status_code == 200
material = response.json()["materials"][0]
assert material["scan_status"] == "infected"
assert material["parse_status"] == "blocked"
stored = TicketDraftMaterial.objects.get(public_id=material["id"])
assert stored.storage_path
assert stored.parsed_text == ""
def test_clamav_instream_response_marks_eicar_as_infected(monkeypatch, settings):
class FakeConnection:
def __init__(self):
self.sent = []
self.responses = [b"stream: Eicar-Test-Signature FOUND\0"]
def __enter__(self):
return self
def __exit__(self, *_args):
return None
def sendall(self, data):
self.sent.append(data)
def recv(self, _size):
return self.responses.pop(0) if self.responses else b""
connection = FakeConnection()
settings.TICKET_MATERIAL_SCANNER_BACKEND = "clamav"
monkeypatch.setattr(
"apps.tickets.materials.socket.create_connection",
lambda *_args, **_kwargs: connection,
)
result = scan_bytes(b"standard antivirus test payload")
assert result["status"] == "infected"
assert result["signature"] == "Eicar-Test-Signature"
assert connection.sent[0] == b"zINSTREAM\0"
def test_parse_failure_isolated_to_material_and_keeps_original(admin_client, db, monkeypatch):
created = _create_draft(admin_client)
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
monkeypatch.setattr(
"apps.tickets.materials.parse_material",
lambda *_args: (_ for _ in ()).throw(MaterialParseError("文档损坏")),
)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("chat.txt", b"valid text", content_type="text/plain")},
)
assert response.status_code == 200
material = response.json()["materials"][0]
assert material["scan_status"] == "clean"
assert material["parse_status"] == "failed"
assert material["parse_error"] == "文档损坏"
assert TicketDraftMaterial.objects.filter(public_id=material["id"]).exists()
def test_material_limits_are_server_controlled(admin_client, db, monkeypatch, settings):
settings.TICKET_MATERIAL_MAX_FILE_BYTES = 4
settings.TICKET_MATERIAL_MAX_FILES = 1
created = _create_draft(admin_client)
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
oversized = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("large.txt", b"12345", content_type="text/plain")},
)
first = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("one.txt", b"1234", content_type="text/plain")},
)
second = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{"file": SimpleUploadedFile("two.txt", b"12", content_type="text/plain")},
)
assert oversized.status_code == 413
assert first.status_code == 200
assert second.status_code == 413
def test_parsed_document_can_start_intake_without_text(admin_client, db, monkeypatch):
create_ticket_agent(capability="text_chat")
created = admin_client.post(
"/api/ticket-drafts",
data={"source_text": ""},
content_type="application/json",
).json()
monkeypatch.setattr("apps.tickets.draft_api.put_object", lambda *_args: None)
uploaded = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/materials",
{
"file": SimpleUploadedFile(
"service.log",
b"2026-07-25 ERROR timeout",
content_type="text/plain",
)
},
).json()
monkeypatch.setattr(
"apps.tickets.draft_api.execute_ticket_draft_analysis.delay",
lambda *_args: SimpleNamespace(id="task-1"),
)
response = admin_client.post(
f"/api/ticket-drafts/{created['public_id']}/analyze",
data={"expected_revision": uploaded["revision"]},
content_type="application/json",
)
assert response.status_code == 200
assert response.json()["status"] == "analyzing"