Bruce/DEMO-AGENT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(regulatory): 增加适用条件确认暂停恢复

Browse Source
This commit is contained in:
bruce
2026-06-07 09:19:31 +08:00
parent bd805203f1
commit bbd2d3532a
12 changed files with 535 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
tests/test_regulatory_condition.py Normal file
View File

@@ -0,0 +1,139 @@
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_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

54
tests/test_regulatory_frontend.py
View File

@@ -50,9 +50,63 @@ def test_workspace_renders_regulatory_workflow_card(client, django_user_model):
assert "data-regulatory-status-url-template" in content
def test_workspace_renders_condition_confirmation_form(client, 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-OK",
status=FileSummaryBatch.Status.SUCCESS,
)
regulatory = RegulatoryReviewBatch.objects.create(
conversation=conversation,
user=user,
source_summary_batch=summary,
batch_no="RR-WAIT",
status=RegulatoryReviewBatch.Status.WAITING_USER,
condition_json={
"confirmed": False,
"candidates": {
"product_category": {
"label": "产品类别",
"input_type": "select",
"options": ["体外诊断试剂", "医疗器械", "其他"],
"suggested": "体外诊断试剂",
},
"product_name": {
"label": "产品名称",
"input_type": "text",
"suggested": "甲胎蛋白检测试剂盒",
},
},
},
)
WorkflowNodeRun.objects.create(
workflow_type="regulatory_review",
workflow_batch_id=regulatory.pk,
node_group="condition_confirm",
node_code="condition_confirm",
node_name="适用条件确认",
status=WorkflowNodeRun.Status.WAITING_USER,
progress=50,
)
client.force_login(user)
response = client.get(f"{reverse('home')}?conversation={conversation.pk}")
content = response.content.decode("utf-8")
assert "适用条件确认" in content
assert "data-condition-confirm-form" in content
assert "体外诊断试剂" in content
assert "甲胎蛋白检测试剂盒" in content
def test_frontend_selects_status_url_by_workflow_type():
script = open("static/js/app.js", encoding="utf-8").read()
assert "workflow_type" in script
assert "data-regulatory-status-url-template" in script
assert "statusUrlForWorkflow" in script
assert "bindConditionConfirmForms" in script
assert "data-condition-confirm-form" in script

4
tests/test_regulatory_workflow.py
View File

@@ -102,6 +102,8 @@ def test_start_regulatory_review_workflow_runs_synchronously(django_user_model):
user=user,
source_summary_batch=summary,
)
batch.condition_json = {"confirmed": True, "confirmed_conditions": {"product_category": "体外诊断试剂"}}
batch.save(update_fields=["condition_json"])
start_regulatory_review_workflow(batch, async_run=False)
@@ -187,6 +189,8 @@ def test_workflow_generates_issues_exports_and_assistant_summary(settings, tmp_p
user=user,
source_summary_batch=summary,
)
batch.condition_json = {"confirmed": True, "confirmed_conditions": {"product_category": "体外诊断试剂"}}
batch.save(update_fields=["condition_json"])
start_regulatory_review_workflow(batch, async_run=False)