feat(file-summary): 实现对话附件上传接口

This commit is contained in:
2026-06-06 01:13:23 +08:00
parent 855afcdee3
commit eb87d9040d
8 changed files with 298 additions and 1 deletions

View File

@@ -0,0 +1,48 @@
from django.core.files.uploadedfile import SimpleUploadedFile
import pytest
from review_agent.file_summary.storage import save_uploaded_attachment
from review_agent.models import Conversation, FileAttachment
pytestmark = pytest.mark.django_db
def test_save_uploaded_attachment_versions_same_name(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="会话")
first = save_uploaded_attachment(
conversation=conversation,
user=user,
uploaded_file=SimpleUploadedFile("资料.docx", b"first"),
)
second = save_uploaded_attachment(
conversation=conversation,
user=user,
uploaded_file=SimpleUploadedFile("资料.docx", b"second"),
)
first.refresh_from_db()
assert first.version_no == 1
assert first.is_active is False
assert second.version_no == 2
assert second.is_active is True
assert FileAttachment.objects.filter(conversation=conversation).count() == 2
assert (tmp_path / second.storage_path).read_bytes() == b"second"
def test_save_uploaded_attachment_rejects_path_traversal(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 = save_uploaded_attachment(
conversation=conversation,
user=user,
uploaded_file=SimpleUploadedFile("../资料.docx", b"content"),
)
assert ".." not in attachment.storage_path
assert (tmp_path / attachment.storage_path).exists()

View File

@@ -0,0 +1,75 @@
from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
import pytest
from review_agent.models import Conversation, FileAttachment
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