141 lines
6.2 KiB
Python
141 lines
6.2 KiB
Python
"""Summary intent and the released shortcut use the summary API task.
|
|
|
|
Offscreen workstation widgets and the real repository are exercised with
|
|
synthetic answers. Every HTTP request ends in MockTransport at example.test.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import httpx
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, Qt, QTimer
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QPushButton
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.services import ApiClient, RemoteDoctorRepository
|
|
from doctor_workstation.ui.dialogs import ai_consult as module
|
|
from doctor_workstation.ui.dialogs.prescription_ai import diagnosis_ai_task
|
|
|
|
|
|
class _SyncMockTransport(httpx.BaseTransport):
|
|
"""Use the finite sync parser without Windows event-loop socket setup."""
|
|
|
|
def __init__(self, handler):
|
|
self.mock = httpx.MockTransport(handler)
|
|
|
|
def handle_request(self, request):
|
|
return self.mock.handle_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(("prompt", "expected"), [
|
|
("请根据当前病历总结患者病情、证候与风险。", "summary"),
|
|
("请概括患者现有证候和舌象。", "summary"),
|
|
("请给出当前辨证思路的摘要。", "summary"),
|
|
("请用一句话 summary 当前证候。", "summary"),
|
|
("请总结当前病情及风险。", "summary"),
|
|
("请生成完整中医辨证报告，并在末尾总结。", "tcm_pattern"),
|
|
("请生成完整的中医报告并附病情摘要。", "tcm_pattern"),
|
|
("请生成 full TCM report，附 summary。", "tcm_pattern"),
|
|
("task=tcm_pattern；总结当前病情。", "tcm_pattern"),
|
|
("请辨证并给出证候、治法。", "tcm_pattern"),
|
|
("tcm", "tcm_pattern"),
|
|
("请总结血糖趋势及可能证候。", "glucose_trend"),
|
|
("请总结用药风险及证候。", "medication_review"),
|
|
("请总结并发症风险及证候。", "complication_risk"),
|
|
("请摘要处方配伍及证候。", "prescription_review"),
|
|
("请总结检查结果及证候。", "exam_review"),
|
|
("请分析当前病历并总结辨证思路。", "case_analysis"),
|
|
])
|
|
def test_mixed_summary_and_tcm_intent_preserves_specific_tasks(prompt, expected):
|
|
assert diagnosis_ai_task(prompt) == expected
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.mark.parametrize("route", ["stream", "legacy404"])
|
|
@pytest.mark.parametrize("entry", ["shortcut", "free_text", "explicit_tcm"])
|
|
def test_summary_workstation_dispatch_reaches_real_repository(
|
|
application, monkeypatch, route, entry,
|
|
):
|
|
requests = []
|
|
expected_task = "tcm_pattern" if entry == "explicit_tcm" else "summary"
|
|
expected_prompt = "请根据当前病历总结患者病情、证候与风险。"
|
|
answer = "## 本次病情小结\n合成主诉为口渴，已记录证候待复核。\n\n## 后续事项\n合成病程日期待明确。"
|
|
workers = []
|
|
# Dispatch uses the real worker, run deterministically after the click so
|
|
# routing coverage does not depend on offscreen Qt thread scheduling.
|
|
monkeypatch.setattr(module, "QThreadPool", SimpleNamespace(
|
|
globalInstance=lambda: SimpleNamespace(start=workers.append),
|
|
))
|
|
|
|
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 = (
|
|
f"event: delta\ndata: {json.dumps({'text': answer}, ensure_ascii=False)}\n\n"
|
|
"event: done\ndata: {}\n\n"
|
|
)
|
|
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}})
|
|
|
|
client = ApiClient("https://example.test", transport=_SyncMockTransport(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.show()
|
|
application.processEvents()
|
|
try:
|
|
if entry == "shortcut":
|
|
# A stable shortcut task must survive a classifier change. This
|
|
# would still fail if only the free-text classifier were repaired.
|
|
monkeypatch.setattr(module, "diagnosis_ai_task", lambda _: "tcm_pattern")
|
|
target = next(button for button in dialog.findChildren(QPushButton)
|
|
if button.text() == "总结当前病情")
|
|
QTest.mouseClick(target, Qt.MouseButton.LeftButton)
|
|
elif entry == "explicit_tcm":
|
|
dialog._ask(expected_prompt, task="tcm_pattern")
|
|
else:
|
|
dialog.input.setText(expected_prompt)
|
|
QTest.mouseClick(dialog.send_button, Qt.MouseButton.LeftButton)
|
|
bubble = dialog._stream_bubble
|
|
assert bubble is not None
|
|
assert len(workers) == 1
|
|
workers[0].run()
|
|
application.processEvents()
|
|
assert dialog._stream_worker is None and not dialog._asking
|
|
assert len(requests) == (1 if route == "stream" else 2)
|
|
for request in requests:
|
|
body = json.loads(request.content)
|
|
assert body["id"] == 501
|
|
assert body["prompt"] == expected_prompt
|
|
assert body["task"] == expected_task
|
|
assert "model" not in body and "model_key" not in body
|
|
assert dialog._stream_task == expected_task
|
|
assert dialog._stream_completed and bubble._raw_payload == answer
|
|
if expected_task == "summary":
|
|
assert bubble._role_name.text() == "AI · 问诊小结"
|
|
assert dialog.send_button.isEnabled()
|
|
finally:
|
|
dialog.close()
|
|
for timer in dialog.findChildren(QTimer):
|
|
timer.stop()
|
|
client.close()
|
|
dialog.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
application.processEvents()
|