331 lines
11 KiB
Python
331 lines
11 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import pytest
|
||
from PySide6.QtWidgets import QApplication, QLabel, QTextBrowser
|
||
|
||
from doctor_workstation.core import PermissionSet
|
||
from doctor_workstation.services import DemoDoctorRepository
|
||
from doctor_workstation.ui.dialogs import ai_consult as ai_consult_module
|
||
from doctor_workstation.ui.dialogs.ai_consult import (
|
||
AiConsultDialog,
|
||
can_open_ai_consult,
|
||
present_ai_consult,
|
||
render_chat_payload,
|
||
)
|
||
from doctor_workstation.ui.pages.appointments import AppointmentsPage
|
||
from doctor_workstation.ui.pages.consultations import ConsultationsPage
|
||
from doctor_workstation.ui.pages.patients import PatientListWorkspace
|
||
from doctor_workstation.ui.pages.reception import ReceptionPage
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application() -> QApplication:
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
@pytest.fixture
|
||
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
def run_immediately(
|
||
function: Any,
|
||
*args: Any,
|
||
on_success: Any = None,
|
||
on_error: Any = None,
|
||
on_finished: Any = None,
|
||
**kwargs: Any,
|
||
) -> object:
|
||
try:
|
||
result = function(*args, **kwargs)
|
||
except Exception as error:
|
||
if on_error:
|
||
on_error(error)
|
||
else:
|
||
if on_success:
|
||
on_success(result)
|
||
finally:
|
||
if on_finished:
|
||
on_finished()
|
||
return object()
|
||
|
||
monkeypatch.setattr(ai_consult_module, "run_async", run_immediately)
|
||
|
||
|
||
def test_ai_consult_dialog_matches_workspace_chrome(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
repository = DemoDoctorRepository()
|
||
dialog = AiConsultDialog(repository, PermissionSet(["tcm.diagnosis/aiAssistant"]))
|
||
dialog.open_for(
|
||
diagnosis_id=501,
|
||
patient_id=301,
|
||
seed={"patient_name": "杨永", "age": 52, "clinical_diagnosis": "2型糖尿病"},
|
||
source_title="问诊列表",
|
||
)
|
||
dialog.show()
|
||
application.processEvents()
|
||
|
||
labels = [widget.text() for widget in dialog.findChildren(QLabel) if widget.text()]
|
||
assert "问诊详情" in labels
|
||
assert "AI 助手" in labels
|
||
assert "智能分析" in labels
|
||
assert "快捷工具" in labels
|
||
assert "对话建议" in labels
|
||
assert dialog.tabs.tabText(0) == "问诊对话"
|
||
assert dialog.send_button.objectName() == "AiConsultSend"
|
||
dialog.close()
|
||
|
||
|
||
def test_present_ai_consult_requires_diagnosis_id(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
opened: list[int] = []
|
||
monkeypatch.setattr(ai_consult_module.AiConsultDialog, "exec", lambda self: opened.append(self.diagnosis_id))
|
||
present_ai_consult(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
None,
|
||
diagnosis_id=0,
|
||
)
|
||
assert opened == []
|
||
present_ai_consult(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
None,
|
||
diagnosis_id=501,
|
||
seed={"patient_name": "杨永"},
|
||
)
|
||
assert opened == [501]
|
||
|
||
|
||
def test_four_entry_points_expose_ai_consult_action(application: QApplication) -> None:
|
||
repository = DemoDoctorRepository()
|
||
allowed = PermissionSet(["*", "tcm.diagnosis/aiAssistant"])
|
||
assert can_open_ai_consult(allowed)
|
||
|
||
patients = PatientListWorkspace(repository, allowed)
|
||
patients.show()
|
||
application.processEvents()
|
||
assert patients.ai_consult_button.text() == "AI 分析"
|
||
assert not patients.ai_consult_button.isHidden()
|
||
|
||
reception = ReceptionPage(repository, allowed)
|
||
reception.show()
|
||
application.processEvents()
|
||
menu_titles = [action.text() for action in reception.more_button.menu().actions()]
|
||
assert "AI 分析" in menu_titles
|
||
assert reception.ai_consult_button.text() == "AI 分析"
|
||
|
||
appointments = AppointmentsPage(repository, permissions=allowed)
|
||
appointments.show()
|
||
application.processEvents()
|
||
assert appointments.toolbar_ai_consult_button.text() == "AI 分析"
|
||
assert not appointments.toolbar_ai_consult_button.isHidden()
|
||
|
||
consultations = ConsultationsPage(repository, permissions=allowed)
|
||
assert consultations.table_host.action_policy.get("ai_consult") is True
|
||
patients.close()
|
||
reception.close()
|
||
appointments.close()
|
||
consultations.close()
|
||
|
||
|
||
def test_ai_consult_sidebar_loads_patient_facts_and_reports(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
dialog = AiConsultDialog(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
dialog.open_for(diagnosis_id=501, patient_id=301, seed={"patient_name": "林晓岚"})
|
||
dialog.show()
|
||
application.processEvents()
|
||
|
||
values = {
|
||
widget.text()
|
||
for widget in dialog.findChildren(QLabel)
|
||
if widget.objectName() == "AiConsultKeyValue"
|
||
}
|
||
assert "22.1" in values
|
||
assert any("病程" in text or "3" in text for text in values)
|
||
titles = {
|
||
widget.text()
|
||
for widget in dialog.findChildren(QLabel)
|
||
if widget.objectName() == "AiConsultRecordTitle"
|
||
}
|
||
assert "血糖控制评估" in titles
|
||
assert "并发症风险评估" in titles
|
||
bodies = [
|
||
widget.toPlainText()
|
||
for widget in dialog.findChildren(QTextBrowser)
|
||
if widget.objectName() == "AiConsultBubbleText"
|
||
]
|
||
assert any("病情与证候分析" in text for text in bodies)
|
||
assert any("###" not in text for text in bodies if "病情与证候分析" in text)
|
||
dialog.close()
|
||
|
||
|
||
def test_ai_consult_sidebar_survives_chat_archive_errors(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
class BrokenChatRepository(DemoDoctorRepository):
|
||
def list_im_chat_messages(self, diagnosis_id: int, *, only_archived: bool = True):
|
||
raise RuntimeError("archive unavailable")
|
||
|
||
dialog = AiConsultDialog(
|
||
BrokenChatRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
dialog.open_for(diagnosis_id=501, patient_id=301)
|
||
dialog.show()
|
||
application.processEvents()
|
||
values = {
|
||
widget.text()
|
||
for widget in dialog.findChildren(QLabel)
|
||
if widget.objectName() == "AiConsultKeyValue"
|
||
}
|
||
titles = {
|
||
widget.text()
|
||
for widget in dialog.findChildren(QLabel)
|
||
if widget.objectName() == "AiConsultRecordTitle"
|
||
}
|
||
assert "22.1" in values
|
||
assert "血糖控制评估" in titles
|
||
dialog.close()
|
||
|
||
|
||
def test_chat_payload_parses_markdown_html_and_json(application: QApplication) -> None:
|
||
browser = QTextBrowser()
|
||
render_chat_payload(browser, "### 病情摘要\n\n**核心病机**\n\n- 口干")
|
||
assert "病情摘要" in browser.toPlainText()
|
||
assert "核心病机" in browser.toPlainText()
|
||
assert "###" not in browser.toPlainText()
|
||
assert "<h3" in browser.toHtml().lower()
|
||
|
||
render_chat_payload(browser, "<p>空腹血糖 <strong>6.8</strong></p>")
|
||
assert "空腹血糖" in browser.toPlainText()
|
||
assert "6.8" in browser.toPlainText()
|
||
|
||
render_chat_payload(browser, '{"diagnosis":"肝郁脾虚证","risk":["血糖波动"]}')
|
||
assert "肝郁脾虚证" in browser.toPlainText()
|
||
browser.deleteLater()
|
||
|
||
|
||
def test_stream_chunks_update_one_ai_bubble_before_done_and_preserve_order(
|
||
application: QApplication,
|
||
) -> None:
|
||
dialog = AiConsultDialog(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
dialog.show()
|
||
dialog._stream_bubble = dialog._append_bubble("ai", "")
|
||
bubble = dialog._stream_bubble
|
||
generation = dialog._generation
|
||
stream_generation = dialog._stream_generation
|
||
|
||
dialog._stream_event(
|
||
generation,
|
||
stream_generation,
|
||
{"event": "delta", "text": "第一段"},
|
||
)
|
||
dialog._flush_timer.stop()
|
||
dialog._flush_stream_chunks()
|
||
application.processEvents()
|
||
assert bubble is not None and bubble.body is not None
|
||
assert bubble.body.toPlainText() == "第一段"
|
||
ai_bubble_count = len(
|
||
[frame for frame in dialog.findChildren(ai_consult_module.QFrame) if frame.objectName() == "AiConsultBubbleAi"]
|
||
)
|
||
|
||
dialog._stream_event(
|
||
generation,
|
||
stream_generation,
|
||
{"event": "delta", "text": "第二段"},
|
||
)
|
||
dialog._stream_event(
|
||
generation,
|
||
stream_generation,
|
||
{"event": "done", "model_label": "千问"},
|
||
)
|
||
application.processEvents()
|
||
assert bubble.body.toPlainText() == "第一段第二段"
|
||
assert len(
|
||
[frame for frame in dialog.findChildren(ai_consult_module.QFrame) if frame.objectName() == "AiConsultBubbleAi"]
|
||
) == ai_bubble_count
|
||
dialog.close()
|
||
|
||
|
||
def test_stream_error_and_cancelled_late_chunk_reuse_or_leave_current_bubble(
|
||
application: QApplication,
|
||
) -> None:
|
||
dialog = AiConsultDialog(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
dialog.show()
|
||
dialog._stream_bubble = dialog._append_bubble("ai", "")
|
||
bubble = dialog._stream_bubble
|
||
generation = dialog._generation
|
||
stream_generation = dialog._stream_generation
|
||
dialog._stream_event(
|
||
generation,
|
||
stream_generation,
|
||
{"event": "delta", "text": "已生成"},
|
||
)
|
||
dialog._stream_failed(generation, stream_generation, RuntimeError("模型繁忙"))
|
||
application.processEvents()
|
||
assert bubble is not None and bubble.body is not None
|
||
assert "已生成" in bubble.body.toPlainText()
|
||
assert "模型繁忙" in bubble.body.toPlainText()
|
||
|
||
before_cancel = bubble.body.toPlainText()
|
||
dialog.close()
|
||
application.processEvents()
|
||
dialog._stream_event(
|
||
generation,
|
||
stream_generation,
|
||
{"event": "delta", "text": "迟到内容"},
|
||
)
|
||
application.processEvents()
|
||
assert bubble.body.toPlainText() == before_cancel
|
||
|
||
|
||
def test_chat_scroll_follows_bottom_but_respects_user_scroll_and_send_restores_it(
|
||
application: QApplication,
|
||
) -> None:
|
||
dialog = AiConsultDialog(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
dialog.diagnosis_id = 501
|
||
dialog.show()
|
||
for index in range(28):
|
||
dialog._append_bubble("ai", f"历史消息 {index}：" + "辨证内容" * 16)
|
||
application.processEvents()
|
||
bar = dialog.chat_scroll.verticalScrollBar()
|
||
bar.setValue(bar.maximum())
|
||
application.processEvents()
|
||
assert dialog._follow_chat
|
||
|
||
bar.setValue(max(0, bar.maximum() // 3))
|
||
application.processEvents()
|
||
reading_position = bar.value()
|
||
assert not dialog._follow_chat
|
||
dialog._append_bubble("ai", "新的流式内容" * 20)
|
||
application.processEvents()
|
||
assert bar.value() == reading_position
|
||
|
||
dialog._ask("请继续分析")
|
||
application.processEvents()
|
||
assert dialog._follow_chat
|
||
assert bar.value() == bar.maximum()
|
||
dialog.close()
|