337 lines
13 KiB
Python
337 lines
13 KiB
Python
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.urls import reverse
|
|
import json
|
|
import pytest
|
|
|
|
from review_agent.models import (
|
|
ApplicationFormFillBatch,
|
|
Conversation,
|
|
ExportedSummaryFile,
|
|
FileAttachment,
|
|
FileSummaryBatch,
|
|
Message,
|
|
WorkflowNodeRun,
|
|
)
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_upload_attachments_requires_conversation_owner(client, settings, tmp_path, django_user_model):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
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="会话")
|
|
client.force_login(other)
|
|
|
|
response = client.post(
|
|
reverse("file_summary_attachment_upload", args=[conversation.pk]),
|
|
{"files": [SimpleUploadedFile("a.docx", b"a")]},
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_attachment_api_requires_login(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
|
|
response = client.get(reverse("file_summary_attachment_list", args=[conversation.pk]))
|
|
|
|
assert response.status_code == 302
|
|
|
|
|
|
def test_upload_and_list_current_conversation_attachments(client, settings, tmp_path, django_user_model):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
client.force_login(user)
|
|
|
|
upload_response = client.post(
|
|
reverse("file_summary_attachment_upload", args=[conversation.pk]),
|
|
{
|
|
"files": [
|
|
SimpleUploadedFile("a.docx", b"a", content_type="application/docx"),
|
|
SimpleUploadedFile("b.zip", b"b", content_type="application/zip"),
|
|
]
|
|
},
|
|
)
|
|
list_response = client.get(reverse("file_summary_attachment_list", args=[conversation.pk]))
|
|
|
|
assert upload_response.status_code == 200
|
|
assert upload_response.json()["attachments"][0]["original_name"] == "a.docx"
|
|
assert len(list_response.json()["attachments"]) == 2
|
|
|
|
|
|
def test_delete_attachment_is_logical_and_scoped(client, settings, tmp_path, django_user_model):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
attachment = FileAttachment.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
original_name="a.docx",
|
|
storage_path="x/a.docx",
|
|
file_size=1,
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.delete(reverse("file_summary_attachment_detail", args=[conversation.pk, attachment.pk]))
|
|
|
|
attachment.refresh_from_db()
|
|
assert response.status_code == 200
|
|
assert attachment.upload_status == FileAttachment.UploadStatus.DELETED
|
|
assert attachment.is_active is False
|
|
|
|
|
|
def test_export_download_requires_batch_owner(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="会话")
|
|
batch = FileSummaryBatch.objects.create(conversation=conversation, user=owner, batch_no="FS-DL")
|
|
report_path = tmp_path / "summary.md"
|
|
report_path.write_text("ok", encoding="utf-8")
|
|
exported = ExportedSummaryFile.objects.create(
|
|
batch=batch,
|
|
export_type=ExportedSummaryFile.ExportType.MARKDOWN,
|
|
file_name="summary.md",
|
|
storage_path=str(report_path),
|
|
)
|
|
|
|
client.force_login(other)
|
|
denied = client.get(reverse("file_summary_export_download", args=[exported.pk]))
|
|
assert denied.status_code == 404
|
|
|
|
client.force_login(owner)
|
|
allowed = client.get(reverse("file_summary_export_download", args=[exported.pk]))
|
|
assert allowed.status_code == 200
|
|
assert "attachment" in allowed["Content-Disposition"]
|
|
assert "summary.md" in allowed["Content-Disposition"]
|
|
assert allowed["Content-Type"].startswith("text/markdown")
|
|
|
|
|
|
def test_export_download_checks_application_form_fill_batch_owner(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")
|
|
owner_conversation = Conversation.objects.create(user=owner, title="自动填表")
|
|
other_conversation = Conversation.objects.create(user=other, title="其他对话")
|
|
owner_summary = FileSummaryBatch.objects.create(
|
|
conversation=owner_conversation,
|
|
user=owner,
|
|
batch_no="FS-AFF-OWNER",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
other_summary = FileSummaryBatch.objects.create(
|
|
conversation=other_conversation,
|
|
user=other,
|
|
batch_no="FS-AFF-OTHER",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
form_batch = ApplicationFormFillBatch.objects.create(
|
|
conversation=owner_conversation,
|
|
user=owner,
|
|
source_summary_batch=owner_summary,
|
|
batch_no="AFF-DL",
|
|
)
|
|
report_path = tmp_path / "filled.docx"
|
|
report_path.write_bytes(b"word-content")
|
|
exported = ExportedSummaryFile.objects.create(
|
|
batch=other_summary,
|
|
workflow_type="application_form_fill",
|
|
workflow_batch_id=form_batch.pk,
|
|
export_category="filled_template",
|
|
export_type=ExportedSummaryFile.ExportType.WORD,
|
|
file_name="filled.docx",
|
|
storage_path=str(report_path),
|
|
)
|
|
|
|
client.force_login(other)
|
|
denied = client.get(reverse("file_summary_export_download", args=[exported.pk]))
|
|
assert denied.status_code == 404
|
|
|
|
client.force_login(owner)
|
|
allowed = client.get(reverse("file_summary_export_download", args=[exported.pk]))
|
|
assert allowed.status_code == 200
|
|
assert allowed["Content-Type"].startswith(
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
)
|
|
assert b"".join(allowed.streaming_content) == b"word-content"
|
|
|
|
|
|
def test_conversation_messages_returns_incremental_messages(client, 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="会话")
|
|
first = Message.objects.create(
|
|
conversation=conversation,
|
|
role=Message.Role.USER,
|
|
content="用户消息",
|
|
)
|
|
second = Message.objects.create(
|
|
conversation=conversation,
|
|
role=Message.Role.ASSISTANT,
|
|
content="报告消息",
|
|
)
|
|
|
|
client.force_login(other)
|
|
denied = client.get(reverse("review_agent_conversation_messages", args=[conversation.pk]))
|
|
assert denied.status_code == 404
|
|
|
|
client.force_login(owner)
|
|
response = client.get(
|
|
f"{reverse('review_agent_conversation_messages', args=[conversation.pk])}?after={first.pk}"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["latest_message_id"] == second.pk
|
|
assert payload["messages"] == [
|
|
{
|
|
"id": second.pk,
|
|
"role": Message.Role.ASSISTANT,
|
|
"content": "报告消息",
|
|
"created_at": second.created_at.isoformat(),
|
|
}
|
|
]
|
|
|
|
|
|
def test_batch_status_exposes_batch_and_node_errors(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
batch = FileSummaryBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
batch_no="FS-ERR",
|
|
status=FileSummaryBatch.Status.FAILED,
|
|
error_message="压缩包解压失败",
|
|
)
|
|
WorkflowNodeRun.objects.create(
|
|
batch=batch,
|
|
node_code="extract",
|
|
node_name="压缩包解压",
|
|
status=WorkflowNodeRun.Status.FAILED,
|
|
progress=10,
|
|
message="未解出任何可扫描文件",
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.get(reverse("file_summary_batch_status", args=[batch.pk]))
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["batch"]["error_message"] == "压缩包解压失败"
|
|
assert payload["nodes"][0]["message"] == "未解出任何可扫描文件"
|
|
|
|
|
|
def test_conversation_list_api_returns_owned_conversations_with_attachment_counts(client, 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")
|
|
owned = Conversation.objects.create(user=owner, title="有附件会话")
|
|
Conversation.objects.create(user=other, title="其他用户会话")
|
|
FileAttachment.objects.create(
|
|
conversation=owned,
|
|
user=owner,
|
|
original_name="a.docx",
|
|
storage_path="x/a.docx",
|
|
file_size=1,
|
|
)
|
|
FileAttachment.objects.create(
|
|
conversation=owned,
|
|
user=owner,
|
|
original_name="deleted.docx",
|
|
storage_path="x/deleted.docx",
|
|
file_size=1,
|
|
upload_status=FileAttachment.UploadStatus.DELETED,
|
|
is_active=False,
|
|
)
|
|
client.force_login(owner)
|
|
|
|
response = client.get(reverse("review_agent_conversation_list"))
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert [item["title"] for item in payload["conversations"]] == ["有附件会话"]
|
|
assert payload["conversations"][0]["attachment_count"] == 1
|
|
|
|
|
|
def test_conversation_delete_api_removes_owned_conversation(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
other = django_user_model.objects.create_user(username="other", password="pass")
|
|
owned = Conversation.objects.create(user=user, title="待删除")
|
|
other_conversation = Conversation.objects.create(user=other, title="别人的会话")
|
|
client.force_login(user)
|
|
|
|
response = client.delete(reverse("review_agent_conversation_detail", args=[owned.pk]))
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["ok"] is True
|
|
assert not Conversation.objects.filter(pk=owned.pk).exists()
|
|
assert Conversation.objects.filter(pk=other_conversation.pk).exists()
|
|
|
|
|
|
def test_conversation_delete_api_rejects_unowned_conversation(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
other = django_user_model.objects.create_user(username="other", password="pass")
|
|
other_conversation = Conversation.objects.create(user=other, title="别人的会话")
|
|
client.force_login(user)
|
|
|
|
response = client.delete(reverse("review_agent_conversation_detail", args=[other_conversation.pk]))
|
|
|
|
assert response.status_code == 404
|
|
assert Conversation.objects.filter(pk=other_conversation.pk).exists()
|
|
|
|
|
|
def test_patch_attachment_updates_name_and_active_state(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
attachment = FileAttachment.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
original_name="old.docx",
|
|
storage_path="x/old.docx",
|
|
file_size=1,
|
|
is_active=True,
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.patch(
|
|
reverse("file_summary_attachment_detail", args=[conversation.pk, attachment.pk]),
|
|
data=json.dumps({"original_name": "new.docx", "is_active": False}),
|
|
content_type="application/json",
|
|
)
|
|
|
|
attachment.refresh_from_db()
|
|
assert response.status_code == 200
|
|
assert attachment.original_name == "new.docx"
|
|
assert attachment.is_active is False
|
|
assert response.json()["attachment"]["original_name"] == "new.docx"
|
|
|
|
|
|
def test_attachment_download_requires_owner_and_returns_file(client, settings, tmp_path, django_user_model):
|
|
settings.MEDIA_ROOT = tmp_path
|
|
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="会话")
|
|
attachment_path = tmp_path / "uploads" / "a.docx"
|
|
attachment_path.parent.mkdir(parents=True)
|
|
attachment_path.write_bytes(b"attachment-content")
|
|
attachment = FileAttachment.objects.create(
|
|
conversation=conversation,
|
|
user=owner,
|
|
original_name="a.docx",
|
|
storage_path=str(attachment_path),
|
|
file_size=attachment_path.stat().st_size,
|
|
content_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
)
|
|
|
|
client.force_login(other)
|
|
denied = client.get(reverse("file_summary_attachment_download", args=[conversation.pk, attachment.pk]))
|
|
assert denied.status_code == 404
|
|
|
|
client.force_login(owner)
|
|
allowed = client.get(reverse("file_summary_attachment_download", args=[conversation.pk, attachment.pk]))
|
|
assert allowed.status_code == 200
|
|
assert "attachment" in allowed["Content-Disposition"]
|
|
assert "a.docx" in allowed["Content-Disposition"]
|
|
assert b"".join(allowed.streaming_content) == b"attachment-content"
|