114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
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
|