import io from types import SimpleNamespace import pytest from django.core.files.uploadedfile import SimpleUploadedFile from django.utils import timezone from PIL import Image from apps.agents.models import AgentDefinition, AgentStepBinding, AgentVersion from apps.audit.models import OperationAuditLog from apps.modelhub.crypto import encrypt_secret from apps.modelhub.models import ModelDeployment, ModelProvider from apps.skills.models import SkillDefinition, SkillVersion from apps.tickets.draft_services import TicketDraftError, add_answer from apps.tickets.models import Ticket, TicketCategory, TicketDraft, TicketDraftMessage from apps.tickets.tasks import cleanup_expired_ticket_drafts def ticket_fields(**overrides): fields = { "title": "生产采集任务未更新", "description": "生产环境采集任务已两小时没有更新", "category_code": "technical_issue", "priority": "high", "problem_performance": "采集进度停滞", "problem_object": "采集任务 A", "submitter_name": "张三", "submitter_org": "华东分公司", "impact_scope": "单个任务", "environment": "生产环境", "related_system": "数据中台", } fields.update(overrides) return fields def create_ticket_agent(*, capability="text_chat", approved=False): provider = ModelProvider.objects.create( name="建单渠道", base_url="https://models.example.com/v1", api_key_encrypted=encrypt_secret("secret"), image_egress_approved=approved, ) deployment = ModelDeployment.objects.create( provider=provider, display_name="建单模型", model_name="ticket-model", capability=capability, ) skill = SkillDefinition.objects.create(code="ticket-draft", name="建单") skill_version = SkillVersion.objects.create( definition=skill, version=1, status="published", prompt_template="{{ input.source_text }}", input_schema={ "type": "object", "properties": {"source_text": {"type": "string"}}, "required": ["source_text"], }, output_schema={"type": "object"}, ) agent = AgentDefinition.objects.create(code="ticket-intake", name="建单 Agent") version = AgentVersion.objects.create( definition=agent, version=1, status="published", model_deployment=deployment, input_schema={"type": "object"}, ) AgentStepBinding.objects.create( agent_version=version, code="draft", order=1, skill_version=skill_version, input_mapping={"source_text": "$input.source_text"}, ) agent.active_version = version agent.save(update_fields=["active_version"]) return agent @pytest.fixture def category(db): return TicketCategory.objects.get(code="technical_issue") def test_feature_requires_published_ticket_agent(admin_client, db): response = admin_client.get("/api/ticket-drafts/availability") assert response.status_code == 200 assert response.json()["available"] is False create_ticket_agent() response = admin_client.get("/api/ticket-drafts/availability") assert response.json()["available"] is True 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( "/api/ticket-drafts", data={"source_text": "任务未更新"}, content_type="application/json", ).json() response = admin_client.patch( f"/api/ticket-drafts/{created['public_id']}", data={"expected_revision": created["revision"], "fields": ticket_fields()}, content_type="application/json", ) assert response.status_code == 200 assert response.json()["revision"] == 2 conflict = admin_client.patch( f"/api/ticket-drafts/{created['public_id']}", data={"expected_revision": 1, "fields": ticket_fields()}, content_type="application/json", ) assert conflict.status_code == 409 def test_confirm_is_idempotent_and_records_accepted(admin_client, category): from django.contrib.auth import get_user_model creator = get_user_model().objects.filter(is_superuser=True).first() fields = ticket_fields() draft = TicketDraft.objects.create( created_by=creator, status="ready", current_fields=fields, last_ai_fields=fields, ) first = admin_client.post( f"/api/ticket-drafts/{draft.public_id}/confirm", data={"expected_revision": draft.revision}, content_type="application/json", ) second = admin_client.post( f"/api/ticket-drafts/{draft.public_id}/confirm", data={"expected_revision": draft.revision}, content_type="application/json", ) 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 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): from django.contrib.auth import get_user_model creator = get_user_model().objects.filter(is_superuser=True).first() draft = TicketDraft.objects.create( created_by=creator, status="ready", current_fields=ticket_fields(title="用户修改后的标题"), last_ai_fields=ticket_fields(), ) response = admin_client.post( f"/api/ticket-drafts/{draft.public_id}/confirm", data={"expected_revision": 1}, content_type="application/json", ) assert response.json()["decision"] == "modified" rejected = TicketDraft.objects.create(created_by=creator, source_text="放弃") response = admin_client.post( f"/api/ticket-drafts/{rejected.public_id}/reject", data={"expected_revision": 1}, content_type="application/json", ) assert response.json()["decision"] == "rejected" def test_guided_answers_stop_after_five_rounds(admin_client, db): from django.contrib.auth import get_user_model draft = TicketDraft.objects.create( created_by=get_user_model().objects.filter(is_superuser=True).first(), guided_rounds=4, ) add_answer(draft, "第五轮回答", "第五轮问题") draft.refresh_from_db() assert draft.guided_rounds == 5 assert TicketDraftMessage.objects.filter(draft=draft).count() == 2 with pytest.raises(TicketDraftError, match="最多进行 5 轮"): add_answer(draft, "第六轮", "") def test_image_upload_validates_content_and_never_audits_source_text(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() 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="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 ) ) assert "包含敏感客户信息的原文" not in str(snapshots) def test_text_model_rejects_image_analysis(admin_client, db): create_ticket_agent(capability="text_chat") 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", ) 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 == 422 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 ): from django.contrib.auth import get_user_model creator = get_user_model().objects.filter(is_superuser=True).first() expired = TicketDraft.objects.create( created_by=creator, source_text="过期敏感原文", expires_at=timezone.now(), ) expired.materials.create( file_name="proof.png", mime_type="image/png", file_size=10, sha256="0" * 64, storage_path="ticket-drafts/proof.png", ) confirmed = TicketDraft.objects.create( created_by=creator, status="confirmed", expires_at=timezone.now(), confirmed_ticket=Ticket.objects.create( ticket_no="TK202607160001", title="已确认", description="保留" ), ) deleted_paths = [] monkeypatch.setattr("apps.tickets.tasks.delete_object", lambda path: deleted_paths.append(path)) assert cleanup_expired_ticket_drafts() == 1 assert not TicketDraft.objects.filter(id=expired.id).exists() assert TicketDraft.objects.filter(id=confirmed.id).exists() assert deleted_paths == ["ticket-drafts/proof.png"] audit = OperationAuditLog.objects.get(action="draft_expired", target_id=str(expired.public_id)) assert "过期敏感原文" not in str(audit.request_snapshot)