feat(application-form-fill): 展示自动填表工作流卡片

tests/test_application_form_fill_views.py

@@ -0,0 +1,113 @@
+import json
+
+import pytest
+from django.urls import reverse
+
+from review_agent.application_form_fill.constants import FORM_FILL_NODE_DEFINITIONS
+from review_agent.models import (
+    ApplicationFormFillBatch,
+    Conversation,
+    ExportedSummaryFile,
+    FileSummaryBatch,
+    WorkflowNodeRun,
+)
+
+
+pytestmark = pytest.mark.django_db
+
+
+def test_application_form_fill_start_requires_conversation_owner(client, monkeypatch, django_user_model):
+    owner = django_user_model.objects.create_user(username="owner", password="pass")
+    other = django_user_model.objects.create_user(username="other", password="pass")
+    conversation = Conversation.objects.create(user=owner, title="会话")
+    FileSummaryBatch.objects.create(
+        conversation=conversation,
+        user=owner,
+        batch_no="FS-VIEW",
+        status=FileSummaryBatch.Status.SUCCESS,
+    )
+    monkeypatch.setattr("review_agent.application_form_fill.views.start_application_form_fill_workflow", lambda batch, async_run=True: None)
+    client.force_login(other)
+
+    response = client.post(
+        reverse("application_form_fill_start"),
+        data=json.dumps({"conversation_id": conversation.pk}),
+        content_type="application/json",
+    )
+
+    assert response.status_code == 404
+
+
+def test_application_form_fill_start_creates_batch(client, monkeypatch, django_user_model):
+    user = django_user_model.objects.create_user(username="owner", password="pass")
+    conversation = Conversation.objects.create(user=user, title="会话")
+    FileSummaryBatch.objects.create(
+        conversation=conversation,
+        user=user,
+        batch_no="FS-VIEW-OK",
+        status=FileSummaryBatch.Status.SUCCESS,
+    )
+    monkeypatch.setattr("review_agent.application_form_fill.views.start_application_form_fill_workflow", lambda batch, async_run=True: None)
+    client.force_login(user)
+
+    response = client.post(
+        reverse("application_form_fill_start"),
+        data=json.dumps({"conversation_id": conversation.pk, "template_codes": ["registration_certificate"]}),
+        content_type="application/json",
+    )
+
+    assert response.status_code == 200
+    payload = response.json()
+    assert payload["workflow_type"] == "application_form_fill"
+    assert ApplicationFormFillBatch.objects.filter(conversation=conversation).exists()
+
+
+def test_application_form_fill_status_requires_owner_and_returns_nodes_exports(client, tmp_path, django_user_model):
+    owner = django_user_model.objects.create_user(username="owner", password="pass")
+    other = django_user_model.objects.create_user(username="other", password="pass")
+    conversation = Conversation.objects.create(user=owner, title="会话")
+    summary = FileSummaryBatch.objects.create(conversation=conversation, user=owner, batch_no="FS-STATUS")
+    batch = ApplicationFormFillBatch.objects.create(
+        conversation=conversation,
+        user=owner,
+        source_summary_batch=summary,
+        batch_no="AFF-STATUS",
+        status=ApplicationFormFillBatch.Status.PARTIAL_SUCCESS,
+        selected_templates=["registration_certificate"],
+        conflict_summary=[{"field_key": "product_name"}],
+        risk_notes=[{"type": "pdf_pending"}],
+    )
+    WorkflowNodeRun.objects.create(
+        workflow_type="application_form_fill",
+        workflow_batch_id=batch.pk,
+        node_group="form_fill",
+        node_code=FORM_FILL_NODE_DEFINITIONS[0][0],
+        node_name=FORM_FILL_NODE_DEFINITIONS[0][1],
+        status=WorkflowNodeRun.Status.SUCCESS,
+        progress=100,
+    )
+    output = tmp_path / "filled.docx"
+    output.write_bytes(b"word")
+    exported = ExportedSummaryFile.objects.create(
+        batch=summary,
+        workflow_type="application_form_fill",
+        workflow_batch_id=batch.pk,
+        export_category="filled_template",
+        export_type=ExportedSummaryFile.ExportType.WORD,
+        file_name="filled.docx",
+        storage_path=str(output),
+    )
+
+    client.force_login(other)
+    denied = client.get(reverse("application_form_fill_batch_status", args=[batch.pk]))
+    assert denied.status_code == 404
+
+    client.force_login(owner)
+    allowed = client.get(reverse("application_form_fill_batch_status", args=[batch.pk]))
+    assert allowed.status_code == 200
+    payload = allowed.json()
+    assert payload["batch"]["workflow_type"] == "application_form_fill"
+    assert payload["batch"]["status"] == ApplicationFormFillBatch.Status.PARTIAL_SUCCESS
+    assert payload["batch"]["conflict_count"] == 1
+    assert payload["nodes"][0]["node_code"] == "prepare"
+    assert payload["exports"][0]["id"] == exported.pk