263 lines
13 KiB
Python
263 lines
13 KiB
Python
"""Six task actions reach the real repository and preserve mocked model answers.
|
||
|
||
Finite offscreen Qt tests only. HTTP terminates in MockTransport; all clinical
|
||
text and transfer metadata below are synthetic, including incomplete sources.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from time import monotonic
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import httpx
|
||
import pytest
|
||
from PySide6.QtCore import QEvent, Qt, QThreadPool, QTimer
|
||
from PySide6.QtTest import QTest
|
||
from PySide6.QtWidgets import QApplication, QLabel, QTextBrowser
|
||
|
||
from doctor_workstation.core import PermissionSet
|
||
from doctor_workstation.services import ApiClient, RemoteDoctorRepository
|
||
from doctor_workstation.services.repository import _diagnosis_ai_request
|
||
from doctor_workstation.ui.dialogs import ai_consult as module
|
||
from doctor_workstation.ui.dialogs.prescription_ai import diagnosis_ai_task
|
||
|
||
# Explicit acceptance values; neither tasks nor answer content are derived from
|
||
# the implementation's classifier or tool-task mapping.
|
||
TOOLS = (
|
||
("分析病历", "case_analysis", "病历分析", "已有口渴记录，结合血糖异常可分析津液受损的可能；不能据此直接确立证型。"),
|
||
("血糖趋势", "glucose_trend", "血糖趋势", "已有一条空腹血糖 8.2 mmol/L；单点不能确定升降趋势，应结合日期与连续监测解释。"),
|
||
("用药评估", "medication_review", "用药评估", "已记录二甲双胍，但剂量与肾功能缺失；可提示胃肠不适监测，不能据此增加剂量。"),
|
||
("并发症风险", "complication_risk", "并发症风险", "糖尿病病史提示需关注眼、肾和足部风险；未记录相应筛查不能推定已患或排除并发症。"),
|
||
("生成问诊小结", "summary", "问诊小结", "本次主诉口渴，既往有糖尿病；已记录二甲双胍及一条空腹血糖，病程与检查日期待明确。"),
|
||
("健康建议", "lifestyle_advice", "健康建议", "可先规律分配主食并记录饮食与血糖；运动宜从适量步行开始，出现低血糖症状时应停止活动。"),
|
||
)
|
||
SOURCE_KEYS = (
|
||
"basic_info", "vital_signs", "chief_complaint", "present_illness", "past_history",
|
||
"other_history", "diagnosis_info", "doctor_image_notes", "doctor_file_notes",
|
||
"daily_tracking", "prescriptions", "video_consultation_text", "chat",
|
||
)
|
||
|
||
|
||
def source_receipt() -> dict:
|
||
return {
|
||
"transfer_manifest_final": True,
|
||
"attachment_count": 0,
|
||
"submitted_attachment_count": 0,
|
||
"items": [],
|
||
"required_sections": [
|
||
{"key": key, "availability": "present" if key in {
|
||
"chief_complaint", "past_history", "prescriptions", "daily_tracking",
|
||
} else "absent", "status": "submitted"}
|
||
for key in SOURCE_KEYS
|
||
],
|
||
}
|
||
|
||
|
||
def answer_for(tool) -> str:
|
||
_label, task, title, analysis = tool
|
||
return (
|
||
f"当前{title}依据已记录的部分资料完成。\n\n"
|
||
f"## {title}判断\n- {analysis}\n- 专属结果标记：{task}_ACTUAL_MODEL_ANSWER。\n\n"
|
||
f"## {title}后续建议\n- 根据以上已有信息执行相应随访，不将缺项改写为正常。"
|
||
"\n- 缺失资料仅影响相应结论，不取代上述分析。"
|
||
)
|
||
|
||
|
||
def sse(kind: str, data: dict) -> str:
|
||
return f"event: {kind}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
|
||
|
||
|
||
def wait_until(predicate, timeout: float = 4.0) -> None:
|
||
deadline = monotonic() + timeout
|
||
while not predicate() and monotonic() < deadline:
|
||
QTest.qWait(5)
|
||
assert predicate(), "finite mocked request did not finish"
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application():
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
@pytest.fixture
|
||
def dialog_factory(application, monkeypatch):
|
||
resources = []
|
||
|
||
def create(handler):
|
||
client = ApiClient("https://example.test", transport=httpx.MockTransport(handler))
|
||
dialog = module.AiConsultDialog(
|
||
RemoteDoctorRepository(client), PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
monkeypatch.setattr(dialog, "_load_workspace", lambda _: None)
|
||
dialog.open_for(diagnosis_id=501, patient_id=1501)
|
||
dialog._detail = {"diagnosis": {"id": 501, "patient_id": 1501}}
|
||
# The intentionally absent case and errored local previews cannot block
|
||
# inference, synthesize a result, or become a stale request context.
|
||
dialog._workspace_sections = {"diagnosis": {"value": None, "error": "合成本地资料未加载"}}
|
||
dialog._patient_ai_context = "不能发往服务的本地旧预览"
|
||
dialog.show()
|
||
application.processEvents()
|
||
resources.append((dialog, client))
|
||
return dialog
|
||
|
||
yield create
|
||
for dialog, client in resources:
|
||
assert QThreadPool.globalInstance().waitForDone(4000)
|
||
dialog.close()
|
||
for timer in dialog.findChildren(QTimer):
|
||
timer.stop()
|
||
client.close()
|
||
dialog.deleteLater()
|
||
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||
application.processEvents()
|
||
|
||
|
||
def click_tool(dialog, label: str) -> None:
|
||
cards = dialog.findChildren(module._ClickCard, "AiConsultToolCard")
|
||
target = next(child for card in cards for child in card.findChildren(QLabel)
|
||
if child.text() == label)
|
||
QTest.mouseClick(target, Qt.MouseButton.LeftButton)
|
||
|
||
|
||
def readable_answer(bubble) -> str:
|
||
if bubble._clinical_panel is None:
|
||
return bubble.body.toPlainText()
|
||
return "\n".join(view.toPlainText()
|
||
for view in bubble._clinical_panel.findChildren(QTextBrowser))
|
||
|
||
|
||
@pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS])
|
||
@pytest.mark.parametrize("route", ["stream", "legacy404"])
|
||
def test_incomplete_case_click_dispatches_distinct_task_and_shows_actual_task_answer(
|
||
dialog_factory, tool, route,
|
||
):
|
||
requests = []
|
||
answer = answer_for(tool)
|
||
label, task, title, analysis = tool
|
||
|
||
def handler(request):
|
||
assert request.url.host == "example.test"
|
||
requests.append(request)
|
||
if request.url.path.endswith("/aiAssistantStream"):
|
||
if route == "legacy404":
|
||
return httpx.Response(404, json={"msg": "合成旧部署无流式路由"})
|
||
events = sse("start", {"model_key": "qwen", "model_label": "原配置千问"})
|
||
events += sse("progress", {"phase": "attachments", "status": "submitted", **source_receipt()})
|
||
events += sse("delta", {"text": answer[:35]})
|
||
events += sse("delta", {"text": answer[35:]})
|
||
events += sse("done", {"model_key": "qwen", "model_label": "原配置千问", **source_receipt()})
|
||
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=events)
|
||
assert route == "legacy404" and request.url.path.endswith("/aiAssistant")
|
||
return httpx.Response(200, json={"code": 1, "data": {
|
||
"answer": answer, "model_key": "qwen", "model_label": "原配置千问", **source_receipt(),
|
||
}})
|
||
|
||
dialog = dialog_factory(handler)
|
||
click_tool(dialog, label)
|
||
bubble = dialog._stream_bubble
|
||
assert bubble is not None
|
||
wait_until(lambda: dialog._stream_worker is None and not dialog._asking)
|
||
assert len(requests) == (1 if route == "stream" else 2)
|
||
expected_prompt = next(prompt for _, name, prompt in module.TOOL_ITEMS if name == label)
|
||
for request in requests:
|
||
assert request.method == "POST"
|
||
body = json.loads(request.content)
|
||
assert set(body) == ({"id", "prompt", "task", "request_id"}
|
||
if request.url.path.endswith("/aiAssistantStream")
|
||
else {"id", "prompt", "task"})
|
||
assert body["task"] == task and body["id"] == 501
|
||
assert body["prompt"] == expected_prompt
|
||
assert "本地旧预览" not in request.content.decode()
|
||
assert "model" not in body and "model_key" not in body # Server model selection remains authoritative.
|
||
assert bubble._raw_payload == answer
|
||
assert isinstance(bubble._clinical_panel, module._StructuredReportPanel)
|
||
assert bubble._clinical_panel.findChild(QLabel, "AiConsultClinicalTitle").text() == f"AI · {title}"
|
||
visible = readable_answer(bubble)
|
||
assert analysis in visible and f"{task}_ACTUAL_MODEL_ANSWER" in visible
|
||
assert [section.title for section in bubble._clinical_panel.sections] == [
|
||
f"{title}判断", f"{title}后续建议",
|
||
]
|
||
assert len(dialog._stream_meta["required_sections"]) == 13
|
||
assert [item["key"] for item in dialog._stream_meta["required_sections"]] == list(SOURCE_KEYS)
|
||
assert dialog._stream_meta["model_key"] == "qwen" and "原配置千问" in bubble.stamp.text()
|
||
assert dialog._stream_completed and dialog.send_button.isEnabled()
|
||
assert not any(timer.isActive() for timer in (
|
||
dialog._delivery_timer, dialog._stream_timeout_timer, dialog._stream_elapsed_timer,
|
||
))
|
||
|
||
|
||
@pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS])
|
||
def test_each_tool_failed_model_request_is_honest_and_does_not_show_fake_success(dialog_factory, tool):
|
||
requests = []
|
||
label, task, title, _analysis = tool
|
||
|
||
def handler(request):
|
||
assert request.url.host == "example.test"
|
||
requests.append(request)
|
||
events = sse("start", {"model_key": "qwen", "model_label": "原配置千问"})
|
||
events += sse("error", {"code": "SYNTHETIC_MODEL_FAILURE", "message": f"{title}合成模型失败"})
|
||
events += sse("done", {"answer": f"{task}_FAKE_LATE_SUCCESS"})
|
||
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=events)
|
||
|
||
dialog = dialog_factory(handler)
|
||
click_tool(dialog, label)
|
||
bubble = dialog._stream_bubble
|
||
wait_until(lambda: dialog._stream_worker is None and not dialog._asking)
|
||
assert len(requests) == 1 and json.loads(requests[0].content)["task"] == task
|
||
assert f"{title}合成模型失败" in bubble._raw_payload
|
||
assert "FAKE_LATE_SUCCESS" not in bubble._raw_payload
|
||
assert "ACTUAL_MODEL_ANSWER" not in bubble._raw_payload
|
||
assert bubble._role_name.text() == f"AI · {title}"
|
||
assert bubble._clinical_panel is None and not dialog._stream_completed
|
||
assert dialog._stream_meta["code"] == "SYNTHETIC_MODEL_FAILURE"
|
||
assert dialog.send_button.isEnabled()
|
||
|
||
|
||
@pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS])
|
||
def test_task_classifier_and_real_repository_allowlist_keep_six_explicit_tasks(tool):
|
||
label, expected_task, _title, _analysis = tool
|
||
prompt = next(prompt for _, name, prompt in module.TOOL_ITEMS if name == label)
|
||
assert diagnosis_ai_task(prompt) == expected_task
|
||
assert _diagnosis_ai_request(501, prompt, expected_task) == (prompt, expected_task)
|
||
assert len(prompt) <= 500
|
||
assert "舌脉" not in prompt and "脉象" not in prompt
|
||
|
||
|
||
def test_unsupported_task_still_rejected():
|
||
with pytest.raises(ValueError, match="task is not supported"):
|
||
_diagnosis_ai_request(501, "合成提问", "not_a_supported_task")
|
||
|
||
|
||
@pytest.mark.parametrize("tool", TOOLS, ids=[tool[1] for tool in TOOLS])
|
||
def test_tool_answers_keep_all_points_and_extra_sections_in_old_clinical_heading_shape(application, tool):
|
||
_label, task, title, _analysis = tool
|
||
headings = ("证候分析", "信息缺口", "建议下一步", "风险提示", f"{title}专属补充")
|
||
points = [f"{heading}要点{index}：" + "这段完整推理与建议需要保留原意和全部内容。" * 6
|
||
+ f"{task}_{ordinal}_{index}_TAIL"
|
||
for ordinal, heading in enumerate(headings) for index in range(5)]
|
||
reply = "已按当前任务完成分析。\n\n" + "\n\n".join(
|
||
f"## {heading}\n" + "\n".join(f"- {point}" for point in points[index * 5:(index + 1) * 5])
|
||
for index, heading in enumerate(headings)
|
||
)
|
||
bubble = module._ChatBubble(role="ai", text=reply)
|
||
try:
|
||
bubble.set_analysis_task(task)
|
||
assert bubble.finalize_clinical_analysis()
|
||
assert isinstance(bubble._clinical_panel, module._StructuredReportPanel)
|
||
assert bubble._raw_payload == reply
|
||
assert [section.title for section in bubble._clinical_panel.sections] == list(headings)
|
||
assert sum(len(section.items) for section in bubble._clinical_panel.sections) == 25
|
||
visible = readable_answer(bubble)
|
||
for point in points:
|
||
assert any(point in section.items for section in bubble._clinical_panel.sections)
|
||
lead, body = point.split("：", 1)
|
||
assert lead in visible and body in visible
|
||
assert "当前更倾向" not in visible
|
||
finally:
|
||
bubble.close()
|
||
bubble.deleteLater()
|
||
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||
application.processEvents()
|