283 lines
11 KiB
Python
283 lines
11 KiB
Python
import pytest
|
|
from django.urls import reverse
|
|
|
|
from review_agent.models import Conversation, FileAttachment, FileSummaryBatch, Message, WorkflowNodeRun, WorkflowNotificationRecord
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_workspace_renders_summary_panel(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
Message.objects.create(
|
|
conversation=conversation,
|
|
role=Message.Role.ASSISTANT,
|
|
content="| 序号 | 文件名 |\n| --- | --- |\n| 1 | a.pdf |\n\n[下载](/api/review-agent/file-summary/exports/1/download/)",
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"{reverse('home')}?conversation={conversation.pk}")
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
assert 'id="summaryPanel"' in content
|
|
assert 'id="uploadDropzone"' in content
|
|
assert 'id="workflowCardList"' in content
|
|
assert 'data-conversation-id="' in content
|
|
assert 'data-message-id="' in content
|
|
assert 'data-message-url-template="' in content
|
|
assert 'class="message-content markdown-content"' in content
|
|
assert 'class="message-raw"' in content
|
|
assert "自动汇总文件目录与页数" in content
|
|
|
|
|
|
def test_workspace_links_to_attachment_manager(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"{reverse('home')}?conversation={conversation.pk}")
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
assert "附件管理" in content
|
|
assert "视频实时监测" not in content
|
|
assert f'href="{reverse("attachment_manager")}?conversation={conversation.pk}"' in content
|
|
assert 'class="attachment-manager-link"' in content
|
|
|
|
|
|
def test_attachment_manager_requires_conversation_selection(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
Conversation.objects.create(user=user, title="待选择会话")
|
|
client.force_login(user)
|
|
|
|
response = client.get(reverse("attachment_manager"))
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
assert "附件管理" in content
|
|
assert "请选择一个对话查看附件" in content
|
|
assert "待选择会话" in content
|
|
assert 'id="attachmentConversationSelect"' in content
|
|
|
|
|
|
def test_attachment_manager_selects_conversation_and_lists_attachments(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="资料会话")
|
|
FileAttachment.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
original_name="a.docx",
|
|
storage_path="x/a.docx",
|
|
file_size=128,
|
|
is_active=True,
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"{reverse('attachment_manager')}?conversation={conversation.pk}")
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
assert "资料会话" in content
|
|
assert "a.docx" in content
|
|
assert "下载" in content
|
|
assert "编辑" in content
|
|
assert "删除" in content
|
|
assert "attachment-manager-split" in content
|
|
assert reverse("home") + f"?conversation={conversation.pk}" in content
|
|
|
|
|
|
def test_attachment_manager_uses_compact_admin_layout(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
Conversation.objects.create(user=user, title="紧凑会话")
|
|
client.force_login(user)
|
|
|
|
response = client.get(reverse("attachment_manager"))
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
css = open("static/css/login.css", encoding="utf-8").read()
|
|
assert "attachment-manager-toolbar" in content
|
|
assert "attachment-manager-content" in content
|
|
assert "attachment-manager-select-control" in content
|
|
assert ".attachment-manager-page" in css
|
|
assert "align-content: start" in css
|
|
assert ".attachment-manager-toolbar" in css
|
|
assert ".attachment-manager-select-control" in css
|
|
assert ".attachment-manager-split" in css
|
|
|
|
|
|
def test_workspace_renders_workflow_history_as_batch_carousel(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
older = FileSummaryBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
batch_no="FS-OLDER",
|
|
status=FileSummaryBatch.Status.SUCCESS,
|
|
)
|
|
latest = FileSummaryBatch.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
batch_no="FS-LATEST",
|
|
status=FileSummaryBatch.Status.FAILED,
|
|
error_message="解压失败",
|
|
)
|
|
WorkflowNodeRun.objects.create(
|
|
batch=older,
|
|
node_code="upload",
|
|
node_name="附件固化",
|
|
status=WorkflowNodeRun.Status.SUCCESS,
|
|
progress=100,
|
|
message="附件固化完成",
|
|
)
|
|
WorkflowNodeRun.objects.create(
|
|
batch=latest,
|
|
node_code="extract",
|
|
node_name="压缩包解压",
|
|
status=WorkflowNodeRun.Status.FAILED,
|
|
progress=10,
|
|
message="压缩包损坏",
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"{reverse('home')}?conversation={conversation.pk}")
|
|
|
|
assert response.status_code == 200
|
|
content = response.content.decode("utf-8")
|
|
assert "workflow-batch-carousel" in content
|
|
assert 'class="workflow-card active"' in content
|
|
assert 'data-workflow-index="0"' in content
|
|
assert 'data-workflow-action="prev"' in content
|
|
assert 'data-workflow-action="next"' in content
|
|
assert content.index("FS-LATEST") < content.index("FS-OLDER")
|
|
assert "压缩包损坏" in content
|
|
|
|
|
|
def test_frontend_prevents_long_message_overflow():
|
|
css = open("static/css/login.css", encoding="utf-8").read()
|
|
|
|
assert ".message-bubble" in css
|
|
assert "overflow-wrap: anywhere" in css
|
|
assert "word-break: break-word" in css
|
|
|
|
|
|
def test_frontend_polls_running_workflow_cards():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert "startWorkflowPolling" in script
|
|
assert "setInterval" in script
|
|
assert "refreshRunningWorkflowCards" in script
|
|
|
|
|
|
def test_frontend_updates_sidebar_conversation_by_stable_id():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert "data-conversation-id" in script
|
|
assert "setAttribute(\"data-conversation-id\"" in script
|
|
assert ".history-item[data-conversation-id=" in script
|
|
|
|
|
|
def test_frontend_refreshes_generated_workflow_messages():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert "refreshConversationMessages" in script
|
|
assert "latestMessageId" in script
|
|
assert "data-message-url-template" in script
|
|
|
|
|
|
def test_frontend_only_scrolls_after_appending_new_messages():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert "return false;" in script
|
|
assert "return true;" in script
|
|
assert "var appendedCount = 0;" in script
|
|
assert "if (appendConversationMessage(message))" in script
|
|
assert "if (appendedCount > 0)" in script
|
|
|
|
|
|
def test_frontend_can_replace_partial_stream_content():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert 'eventName === "replace"' in script
|
|
assert "assistantText = payload.content" in script
|
|
|
|
|
|
def test_frontend_enter_sends_and_ctrl_enter_inserts_newline():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
|
|
assert "bindPromptKeyboardShortcuts" in script
|
|
assert "event.key === \"Enter\"" in script
|
|
assert "event.ctrlKey" in script
|
|
assert "composer.requestSubmit()" in script
|
|
|
|
|
|
def test_frontend_renders_workflow_error_messages():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
css = open("static/css/login.css", encoding="utf-8").read()
|
|
|
|
assert "payload.batch.error_message" in script
|
|
assert "workflow-error" in script
|
|
assert "node.message" in script
|
|
assert ".workflow-error" in css
|
|
|
|
|
|
def test_file_summary_status_includes_feishu_notification(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-FEISHU")
|
|
WorkflowNotificationRecord.objects.create(
|
|
workflow_type="file_summary",
|
|
workflow_batch_id=batch.pk,
|
|
workflow_batch_no=batch.batch_no,
|
|
workflow_status=batch.status,
|
|
dedupe_key=f"file_summary:{batch.pk}:{batch.status}",
|
|
trigger_user=user,
|
|
channel=WorkflowNotificationRecord.Channel.FEISHU_API,
|
|
target="负责人",
|
|
send_status=WorkflowNotificationRecord.SendStatus.SUCCESS,
|
|
message_title="自动汇总完成",
|
|
)
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"/api/review-agent/file-summary/{batch.pk}/status/")
|
|
|
|
payload = response.json()
|
|
assert payload["latest_notification"]["status_label"] == "飞书通知已发送"
|
|
assert payload["notifications"][0]["target"] == "负责人"
|
|
|
|
|
|
def test_frontend_renders_workflow_batches_as_carousel():
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
css = open("static/css/login.css", encoding="utf-8").read()
|
|
|
|
assert "selectWorkflowBatchIndex" in script
|
|
assert "refreshWorkflowBatchCarousel" in script
|
|
assert "data-workflow-action" in script
|
|
assert "workflow-batch-carousel" in script
|
|
assert ".workflow-batch-controls" in css
|
|
assert ".workflow-card.active" in css
|
|
|
|
|
|
def test_workspace_tool_buttons_fill_default_prompts(client, django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
client.force_login(user)
|
|
|
|
response = client.get(f"{reverse('home')}?conversation={conversation.pk}")
|
|
|
|
content = response.content.decode("utf-8")
|
|
script = open("static/js/app.js", encoding="utf-8").read()
|
|
assert "目录自动汇总" in content
|
|
assert "法规核查与风险预警" in content
|
|
assert "申报文件填表" in content
|
|
assert "说明书审查" not in content
|
|
assert ">风险预警</button>" not in content
|
|
assert 'data-prompt-template="请对当前对话已上传的文件或压缩包自动汇总文件目录' in content
|
|
assert 'data-prompt-template="请对当前对话最近成功汇总的注册资料发起 NMPA 法规核查与风险预警' in content
|
|
assert 'data-prompt-template="请基于当前对话最近成功汇总的产品资料，自动提取产品关键信息并填入申报文件模板"' in content
|
|
assert "优先生成注册证 Word 和字段来源追溯清单" not in content
|
|
assert "bindPromptTemplateButtons" in script
|
|
assert "promptInput.value = template" in script
|