DEMO-AGENT/tests/test_file_summary_views.py

from django.core.files.uploadedfile import SimpleUploadedFile
from django.urls import reverse
import pytest

from review_agent.models import (
    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_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"] == "未解出任何可扫描文件"