93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from review_agent.models import Conversation, FileAttachment, Message, RegulatoryInfoPackageBatch, WorkflowNodeRun
|
|
from review_agent.regulatory_info_package.constants import (
|
|
REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS,
|
|
WORKFLOW_TYPE,
|
|
)
|
|
from review_agent.regulatory_info_package.workflow import (
|
|
create_regulatory_info_package_batch,
|
|
start_regulatory_info_package_workflow,
|
|
)
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
def test_create_regulatory_info_package_batch_initializes_nodes(django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
|
|
batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
|
|
|
|
assert batch.batch_no.startswith("RIP-")
|
|
assert batch.work_dir
|
|
nodes = WorkflowNodeRun.objects.filter(
|
|
workflow_type=WORKFLOW_TYPE,
|
|
workflow_batch_id=batch.pk,
|
|
).order_by("id")
|
|
assert [node.node_code for node in nodes] == [
|
|
code for code, _name, _group in REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS
|
|
]
|
|
|
|
|
|
def test_create_regulatory_info_package_batch_is_node_idempotent(django_user_model):
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
|
|
|
|
create_regulatory_info_package_batch(conversation=conversation, user=user, existing_batch=batch)
|
|
|
|
assert WorkflowNodeRun.objects.filter(
|
|
workflow_type=WORKFLOW_TYPE,
|
|
workflow_batch_id=batch.pk,
|
|
).count() == len(REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS)
|
|
|
|
|
|
def test_empty_workflow_skeleton_completes(django_user_model, settings):
|
|
settings.REGULATORY_INFO_PACKAGE_ASYNC = False
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
batch = create_regulatory_info_package_batch(conversation=conversation, user=user)
|
|
|
|
start_regulatory_info_package_workflow(batch, async_run=False)
|
|
batch.refresh_from_db()
|
|
|
|
assert batch.status == RegulatoryInfoPackageBatch.Status.SUCCESS
|
|
assert WorkflowNodeRun.objects.filter(
|
|
workflow_type=WORKFLOW_TYPE,
|
|
workflow_batch_id=batch.pk,
|
|
status=WorkflowNodeRun.Status.SUCCESS,
|
|
).count() == len(REGULATORY_INFO_PACKAGE_NODE_DEFINITIONS)
|
|
|
|
|
|
def test_completed_workflow_appends_download_summary_message(django_user_model, settings):
|
|
settings.REGULATORY_INFO_PACKAGE_ASYNC = False
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
trigger = Message.objects.create(conversation=conversation, role=Message.Role.USER, content="根据说明书生成第1章监管信息")
|
|
source = Path("docs/0.原始材料/目标产品说明书.docx").resolve()
|
|
attachment = FileAttachment.objects.create(
|
|
conversation=conversation,
|
|
user=user,
|
|
original_name="目标产品说明书.docx",
|
|
storage_path=str(source),
|
|
file_size=source.stat().st_size,
|
|
)
|
|
batch = create_regulatory_info_package_batch(
|
|
conversation=conversation,
|
|
user=user,
|
|
trigger_message=trigger,
|
|
source_attachment=attachment,
|
|
source_file_name=attachment.original_name,
|
|
source_storage_path=attachment.storage_path,
|
|
)
|
|
|
|
start_regulatory_info_package_workflow(batch, async_run=False)
|
|
|
|
message = conversation.messages.filter(role=Message.Role.ASSISTANT, content__contains=batch.batch_no).latest("id")
|
|
assert "第1章 监管信息(预生成版).zip" in message.content
|
|
assert "/api/review-agent/file-summary/exports/" in message.content
|