94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
import json
|
||
|
||
from review_agent.regulatory_review.services.llm_review import review_condition_fields, review_workflow_payload
|
||
|
||
|
||
def test_review_condition_fields_selects_more_complete_llm_product_name():
|
||
def completion(messages, temperature=0.0):
|
||
return json.dumps(
|
||
{
|
||
"fields": {
|
||
"产品名称": "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒 （荧光PCR法）",
|
||
"型号规格": "24人份/盒",
|
||
}
|
||
},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
result = review_condition_fields(
|
||
text="产品名称：呼吸道合胞病毒、肺炎支原体核酸检测试剂盒\n（荧光PCR法）\n型号规格：24人份/盒",
|
||
rule_fields={"产品名称": "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒", "型号规格": "24人份/盒"},
|
||
file_context="申请表.txt",
|
||
completion_func=completion,
|
||
)
|
||
|
||
assert result["selected_fields"]["产品名称"] == "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒 （荧光PCR法）"
|
||
assert result["selected_sources"]["产品名称"] == "llm"
|
||
assert result["selected_sources"]["型号规格"] == "rule"
|
||
|
||
|
||
def test_review_condition_fields_falls_back_when_llm_returns_chapter_title():
|
||
def completion(messages, temperature=0.0):
|
||
return json.dumps({"fields": {"产品名称": "第1章 监管信息"}}, ensure_ascii=False)
|
||
|
||
result = review_condition_fields(
|
||
text="产品名称：甲胎蛋白检测试剂盒",
|
||
rule_fields={"产品名称": "甲胎蛋白检测试剂盒"},
|
||
file_context="申请表.txt",
|
||
completion_func=completion,
|
||
)
|
||
|
||
assert result["selected_fields"]["产品名称"] == "甲胎蛋白检测试剂盒"
|
||
assert result["selected_sources"]["产品名称"] == "rule"
|
||
|
||
|
||
def test_review_condition_fields_rejects_garbled_llm_product_name():
|
||
def completion(messages, temperature=0.0):
|
||
return json.dumps({"fields": {"产品名称": "呼吸道合胞病毒、 <20>肺炎支原体核酸检测试剂盒 (荧光PCR法)"}}, ensure_ascii=False)
|
||
|
||
result = review_condition_fields(
|
||
text="呼吸道合胞病毒、肺炎支原体核酸检测试剂盒（荧光PCR法）",
|
||
rule_fields={"产品名称": "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒（荧光PCR法）"},
|
||
file_context="产品列表.txt",
|
||
completion_func=completion,
|
||
)
|
||
|
||
assert result["selected_fields"]["产品名称"] == "呼吸道合胞病毒、肺炎支原体核酸检测试剂盒（荧光PCR法）"
|
||
assert result["selected_sources"]["产品名称"] == "rule"
|
||
|
||
|
||
def test_review_workflow_payload_handles_timeout_without_raising():
|
||
def completion(messages, temperature=0.0):
|
||
raise TimeoutError("The read operation timed out")
|
||
|
||
result = review_workflow_payload(
|
||
stage="completeness_check",
|
||
payload={"findings": []},
|
||
completion_func=completion,
|
||
)
|
||
|
||
assert result["status"] == "failed"
|
||
assert result["stage"] == "completeness_check"
|
||
assert "timed out" in result["error_message"]
|
||
|
||
|
||
def test_review_workflow_payload_retries_timeout_before_success(settings):
|
||
settings.REGULATORY_LLM_REVIEW_RETRY_DELAY_SECONDS = 0
|
||
attempts = {"count": 0}
|
||
|
||
def completion(messages, temperature=0.0):
|
||
attempts["count"] += 1
|
||
if attempts["count"] < 3:
|
||
raise TimeoutError("The read operation timed out")
|
||
return json.dumps({"reviewed": True})
|
||
|
||
result = review_workflow_payload(
|
||
stage="completeness_check",
|
||
payload={"findings": []},
|
||
completion_func=completion,
|
||
)
|
||
|
||
assert attempts["count"] == 3
|
||
assert result["status"] == "success"
|
||
assert result["result"]["reviewed"] is True
|