137 lines
5.1 KiB
Python
137 lines
5.1 KiB
Python
import pytest
|
||
from django.urls import reverse
|
||
|
||
from review_agent.models import Conversation, FileSummaryBatch, RegulatoryReviewBatch, WorkflowNodeRun
|
||
|
||
|
||
pytestmark = pytest.mark.django_db
|
||
|
||
|
||
def test_regulatory_batch_status_requires_owner(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="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
batch_no="FS-OK",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
)
|
||
batch = RegulatoryReviewBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
source_summary_batch=summary,
|
||
batch_no="RR-STATUS",
|
||
)
|
||
WorkflowNodeRun.objects.create(
|
||
workflow_type="regulatory_review",
|
||
workflow_batch_id=batch.pk,
|
||
node_group="regulatory_review",
|
||
node_code="prepare",
|
||
node_name="准备",
|
||
progress=50,
|
||
)
|
||
|
||
client.force_login(other)
|
||
denied = client.get(reverse("regulatory_review_batch_status", args=[batch.pk]))
|
||
assert denied.status_code == 404
|
||
|
||
client.force_login(owner)
|
||
allowed = client.get(reverse("regulatory_review_batch_status", args=[batch.pk]))
|
||
assert allowed.status_code == 200
|
||
payload = allowed.json()
|
||
assert payload["batch"]["workflow_type"] == "regulatory_review"
|
||
assert payload["batch"]["batch_no"] == "RR-STATUS"
|
||
assert payload["nodes"][0]["node_code"] == "prepare"
|
||
|
||
|
||
def test_regulatory_batch_status_exposes_condition_confirmation(client, django_user_model):
|
||
owner = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=owner, title="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
batch_no="FS-OK",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
)
|
||
batch = RegulatoryReviewBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
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": "体外诊断试剂",
|
||
}
|
||
},
|
||
},
|
||
)
|
||
client.force_login(owner)
|
||
|
||
response = client.get(reverse("regulatory_review_batch_status", args=[batch.pk]))
|
||
|
||
payload = response.json()
|
||
assert payload["batch"]["status"] == RegulatoryReviewBatch.Status.WAITING_USER
|
||
assert payload["condition_confirmation"]["batch_id"] == batch.pk
|
||
assert payload["condition_confirmation"]["candidates"]["product_category"]["suggested"] == "体外诊断试剂"
|
||
|
||
|
||
def test_regulatory_batch_status_refreshes_incomplete_condition_candidates(
|
||
client, settings, tmp_path, django_user_model
|
||
):
|
||
settings.MEDIA_ROOT = tmp_path
|
||
owner = django_user_model.objects.create_user(username="owner", password="pass")
|
||
conversation = Conversation.objects.create(user=owner, title="会话")
|
||
summary = FileSummaryBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
batch_no="FS-OK",
|
||
status=FileSummaryBatch.Status.SUCCESS,
|
||
product_name="第1章 监管信息",
|
||
)
|
||
application = tmp_path / "application.txt"
|
||
application.write_text(
|
||
"卡尤迪生物科技宜兴有限公司申请境内第三类体外诊断试剂"
|
||
"呼吸道合胞病毒、肺炎支原体核酸检测试剂盒（荧光PCR法）产品注册。",
|
||
encoding="utf-8",
|
||
)
|
||
from review_agent.models import FileSummaryItem
|
||
|
||
FileSummaryItem.objects.create(
|
||
batch=summary,
|
||
file_index=1,
|
||
directory_level="第1章 监管信息",
|
||
file_name="符合标准的清单.txt",
|
||
file_type="txt",
|
||
relative_path="第1章 监管信息/符合标准的清单.txt",
|
||
storage_path=str(application),
|
||
)
|
||
batch = RegulatoryReviewBatch.objects.create(
|
||
conversation=conversation,
|
||
user=owner,
|
||
source_summary_batch=summary,
|
||
batch_no="RR-WAIT-EMPTY",
|
||
status=RegulatoryReviewBatch.Status.WAITING_USER,
|
||
condition_json={
|
||
"confirmed": False,
|
||
"candidates": {
|
||
"product_category": {"suggested": "其他"},
|
||
"product_name": {"suggested": ""},
|
||
},
|
||
},
|
||
)
|
||
client.force_login(owner)
|
||
|
||
response = client.get(reverse("regulatory_review_batch_status", args=[batch.pk]))
|
||
|
||
payload = response.json()
|
||
candidates = payload["condition_confirmation"]["candidates"]
|
||
assert candidates["product_category"]["suggested"] == "体外诊断试剂"
|
||
assert candidates["product_name"]["suggested"] == "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒（荧光PCR法）"
|