252 lines
9.6 KiB
Python
252 lines
9.6 KiB
Python
import json
|
||
|
||
import pytest
|
||
from django.urls import reverse
|
||
|
||
from review_agent.models import (
|
||
Conversation,
|
||
FileSummaryBatch,
|
||
FileSummaryItem,
|
||
RegulatoryReviewBatch,
|
||
WorkflowEvent,
|
||
WorkflowNodeRun,
|
||
)
|
||
from review_agent.regulatory_review.services.info_extract import detect_regulatory_condition_candidates
|
||
from review_agent.regulatory_review.workflow import (
|
||
create_regulatory_review_batch,
|
||
start_regulatory_review_workflow,
|
||
)
|
||
|
||
|
||
pytestmark = pytest.mark.django_db
|
||
|
||
|
||
def test_detect_regulatory_condition_candidates_from_summary_items(django_user_model):
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=user, title="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="甲胎蛋白检测试剂盒",
|
||
)
|
||
FileSummaryItem.objects.create(
|
||
batch=summary,
|
||
file_index=1,
|
||
directory_level="临床评价资料",
|
||
file_name="免临床评价资料.docx",
|
||
file_type="docx",
|
||
relative_path="4.临床评价资料/免临床评价资料.docx",
|
||
storage_path="missing.docx",
|
||
)
|
||
|
||
candidates = detect_regulatory_condition_candidates(summary)
|
||
|
||
assert candidates["product_category"]["suggested"] == "体外诊断试剂"
|
||
assert candidates["registration_type"]["suggested"] == "首次注册"
|
||
assert candidates["clinical_evaluation_path"]["suggested"] == "免临床"
|
||
assert candidates["product_name"]["suggested"] == "甲胎蛋白检测试剂盒"
|
||
|
||
|
||
def test_detect_regulatory_condition_prefers_attachment_fields_over_chapter_title(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="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="第1章 监管信息",
|
||
)
|
||
application = tmp_path / "application.txt"
|
||
application.write_text(
|
||
"产品名称：甲胎蛋白检测试剂盒\n型号规格：20人份/盒\n预期用途：用于人血清中甲胎蛋白检测\n注册类型：首次注册\n",
|
||
encoding="utf-8",
|
||
)
|
||
FileSummaryItem.objects.create(
|
||
batch=summary,
|
||
file_index=1,
|
||
directory_level="1. 监管信息 / 1.2 申请表",
|
||
file_name="申请表.txt",
|
||
file_type="txt",
|
||
relative_path="1.监管信息/申请表.txt",
|
||
storage_path=str(application),
|
||
)
|
||
|
||
candidates = detect_regulatory_condition_candidates(summary)
|
||
|
||
assert candidates["product_name"]["suggested"] == "甲胎蛋白检测试剂盒"
|
||
assert candidates["model_spec"]["suggested"] == "20人份/盒"
|
||
assert candidates["intended_use"]["suggested"] == "用于人血清中甲胎蛋白检测"
|
||
|
||
|
||
def test_detect_regulatory_condition_keeps_wrapped_product_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="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="第1章 监管信息",
|
||
)
|
||
application = tmp_path / "application.txt"
|
||
application.write_text(
|
||
"产品名称：呼吸道合胞病毒、肺炎支原体核酸检测试剂盒\n"
|
||
"（荧光PCR法）\n"
|
||
"型号规格：24人份/盒\n"
|
||
"预期用途：用于呼吸道合胞病毒、肺炎支原体核酸检测\n",
|
||
encoding="utf-8",
|
||
)
|
||
FileSummaryItem.objects.create(
|
||
batch=summary,
|
||
file_index=1,
|
||
directory_level="1. 监管信息 / 1.2 申请表",
|
||
file_name="申请表.txt",
|
||
file_type="txt",
|
||
relative_path="1.监管信息/申请表.txt",
|
||
storage_path=str(application),
|
||
)
|
||
|
||
candidates = detect_regulatory_condition_candidates(summary)
|
||
|
||
assert candidates["product_name"]["suggested"] == "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒 （荧光PCR法）"
|
||
assert candidates["model_spec"]["suggested"] == "24人份/盒"
|
||
|
||
|
||
def test_detect_regulatory_condition_uses_llm_review_for_better_product_name(
|
||
monkeypatch, settings, tmp_path, django_user_model
|
||
):
|
||
settings.MEDIA_ROOT = tmp_path
|
||
settings.REGULATORY_LLM_REVIEW_ALLOW_TEST_CALLS = True
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=user, title="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="第1章 监管信息",
|
||
)
|
||
application = tmp_path / "application.txt"
|
||
application.write_text(
|
||
"产品名称：呼吸道合胞病毒、肺炎支原体核酸检测试剂盒\n"
|
||
"型号规格：24人份/盒\n",
|
||
encoding="utf-8",
|
||
)
|
||
FileSummaryItem.objects.create(
|
||
batch=summary,
|
||
file_index=1,
|
||
directory_level="1. 监管信息 / 1.2 申请表",
|
||
file_name="申请表.txt",
|
||
file_type="txt",
|
||
relative_path="1.监管信息/申请表.txt",
|
||
storage_path=str(application),
|
||
)
|
||
|
||
monkeypatch.setattr(
|
||
"review_agent.regulatory_review.services.llm_review.generate_completion",
|
||
lambda messages, temperature=0.0: json.dumps(
|
||
{"fields": {"产品名称": "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒 （荧光PCR法）"}},
|
||
ensure_ascii=False,
|
||
),
|
||
)
|
||
|
||
candidates = detect_regulatory_condition_candidates(summary)
|
||
|
||
assert candidates["product_name"]["suggested"] == "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒 （荧光PCR法）"
|
||
assert candidates["product_name"]["source"] == "llm"
|
||
|
||
|
||
def test_workflow_pauses_before_rule_scope_until_conditions_confirmed(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="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="甲胎蛋白检测试剂盒",
|
||
)
|
||
batch = create_regulatory_review_batch(
|
||
conversation=conversation,
|
||
user=user,
|
||
source_summary_batch=summary,
|
||
)
|
||
|
||
start_regulatory_review_workflow(batch, async_run=False)
|
||
|
||
batch.refresh_from_db()
|
||
condition_node = WorkflowNodeRun.objects.get(
|
||
workflow_type="regulatory_review",
|
||
workflow_batch_id=batch.pk,
|
||
node_code="condition_confirm",
|
||
)
|
||
rule_scope_node = WorkflowNodeRun.objects.get(
|
||
workflow_type="regulatory_review",
|
||
workflow_batch_id=batch.pk,
|
||
node_code="rule_scope",
|
||
)
|
||
assert batch.status == RegulatoryReviewBatch.Status.WAITING_USER
|
||
assert condition_node.status == WorkflowNodeRun.Status.WAITING_USER
|
||
assert rule_scope_node.status == WorkflowNodeRun.Status.PENDING
|
||
assert batch.condition_json["candidates"]["product_category"]["suggested"] == "体外诊断试剂"
|
||
assert WorkflowEvent.objects.filter(
|
||
workflow_type="regulatory_review",
|
||
workflow_batch_id=batch.pk,
|
||
event_type="waiting_user",
|
||
).exists()
|
||
|
||
|
||
def test_confirm_conditions_endpoint_resumes_workflow(client, settings, tmp_path, django_user_model):
|
||
settings.MEDIA_ROOT = tmp_path
|
||
settings.REGULATORY_REVIEW_ASYNC = False
|
||
user = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=user, title="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=user,
|
||
batch_no="FS-COND",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="甲胎蛋白检测试剂盒",
|
||
)
|
||
batch = create_regulatory_review_batch(
|
||
conversation=conversation,
|
||
user=user,
|
||
source_summary_batch=summary,
|
||
)
|
||
start_regulatory_review_workflow(batch, async_run=False)
|
||
client.force_login(user)
|
||
|
||
response = client.post(
|
||
reverse("regulatory_review_confirm_conditions", args=[batch.pk]),
|
||
data=json.dumps(
|
||
{
|
||
"conditions": {
|
||
"product_category": "体外诊断试剂",
|
||
"registration_type": "首次注册",
|
||
"clinical_evaluation_path": "免临床",
|
||
"product_name": "甲胎蛋白检测试剂盒",
|
||
"model_spec": "卡型",
|
||
"intended_use": "用于甲胎蛋白检测",
|
||
}
|
||
}
|
||
),
|
||
content_type="application/json",
|
||
)
|
||
|
||
batch.refresh_from_db()
|
||
assert response.status_code == 200
|
||
assert response.json()["batch"]["status"] == RegulatoryReviewBatch.Status.SUCCESS
|
||
assert batch.condition_json["confirmed"] is True
|
||
assert batch.condition_json["confirmed_conditions"]["model_spec"] == "卡型"
|
||
assert WorkflowNodeRun.objects.get(
|
||
workflow_type="regulatory_review",
|
||
workflow_batch_id=batch.pk,
|
||
node_code="condition_confirm",
|
||
).status == WorkflowNodeRun.Status.SUCCESS
|