192 lines
8.5 KiB
Python
192 lines
8.5 KiB
Python
"""Preparation substages describe observed work without echoing source text."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, QTimer
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from doctor_workstation.core.errors import ApiBusinessError
|
|
from doctor_workstation.ui.dialogs import ai_consult as module
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def dialog(application, monkeypatch):
|
|
monkeypatch.setattr(module, "QThreadPool", SimpleNamespace(
|
|
globalInstance=lambda: SimpleNamespace(start=lambda _: None),
|
|
))
|
|
instance = module.AiConsultDialog(object())
|
|
monkeypatch.setattr(instance, "_load_workspace", lambda _: None)
|
|
instance.open_for(diagnosis_id=501, patient_id=1501)
|
|
yield instance
|
|
for timer in instance.findChildren(QTimer):
|
|
timer.stop()
|
|
instance.close()
|
|
instance.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
application.processEvents()
|
|
|
|
|
|
def emit(dialog, payload):
|
|
dialog._stream_event(dialog._generation, dialog._stream_generation, payload)
|
|
|
|
|
|
def test_queued_request_at_52_seconds_does_not_claim_preparation(dialog, monkeypatch):
|
|
now = [100.0]
|
|
monkeypatch.setattr(module, "monotonic", lambda: now[0])
|
|
dialog._ask("合成完整资料核对")
|
|
worker, bubble = dialog._stream_worker, dialog._stream_bubble
|
|
assert worker.prompt == "合成完整资料核对" and worker.diagnosis_id == 501
|
|
assert worker.run_started_at is None
|
|
now[0] = 152.0
|
|
dialog._stream_elapsed_timer.timeout.emit()
|
|
assert bubble._raw_payload == "正在连接 AI 服务，等待资料处理进度…"
|
|
assert "52 秒" in bubble.stamp.text()
|
|
assert "正在整理" not in bubble.body.toPlainText()
|
|
assert dialog._asking and dialog._stream_timeout_timer.isActive()
|
|
|
|
emit(dialog, {"event": "start", "message": "禁止显示的患者临床信息"})
|
|
assert bubble._raw_payload == module.AI_STREAM_WAITING_TEXT
|
|
assert module.AI_STREAM_WAITING_TEXT in bubble.transfer_status.text()
|
|
emit(dialog, {"event": "progress", "phase": "preparing"})
|
|
assert bubble._raw_payload == module.AI_STREAM_PREPARING_TEXT
|
|
|
|
|
|
@pytest.mark.parametrize("status,expected", [
|
|
("aggregating", "正在汇总本次授权范围内的患者资料…"),
|
|
("aggregated", "患者资料已汇总，正在检查提交内容…"),
|
|
("ready", "资料正文已整理，正在准备提交…"),
|
|
(None, module.AI_STREAM_PREPARING_TEXT),
|
|
("禁止显示的患者临床信息", module.AI_STREAM_PREPARING_TEXT),
|
|
(["禁止显示的患者临床信息"], module.AI_STREAM_PREPARING_TEXT),
|
|
])
|
|
def test_known_preparation_states_use_safe_local_copy(dialog, status, expected):
|
|
dialog._ask("合成资料核对")
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": status,
|
|
"message": "禁止显示的患者临床信息 https://example.test/private.pdf",
|
|
})
|
|
bubble = dialog._stream_bubble
|
|
assert bubble._raw_payload == expected
|
|
assert bubble.transfer_status.text().splitlines()[0] == expected
|
|
assert "禁止显示" not in bubble.body.toPlainText() + bubble.transfer_status.text()
|
|
assert "example.test" not in bubble.body.toPlainText() + bubble.transfer_status.text()
|
|
assert "资料已提交" not in bubble._raw_payload
|
|
assert dialog._asking and not dialog._stream_completed
|
|
|
|
|
|
@pytest.mark.parametrize("completed,total", [(0, 3), (1, 3), (3, 3)])
|
|
def test_compacting_counts_same_model_history_work_and_keeps_all_13_receipts(
|
|
dialog, completed, total,
|
|
):
|
|
dialog._ask("合成完整病历核对")
|
|
manifest = [
|
|
{"key": key, "label": "禁止显示的患者临床信息", "status": "prepared"}
|
|
for key, _ in module._REQUIRED_TRANSFER_SECTIONS
|
|
]
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": "aggregated",
|
|
"required_sections": manifest, "source_summary": {"diagnosis_count": 23},
|
|
})
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": "compacting",
|
|
"completed": completed, "total": total, "message": "禁止显示的患者临床信息",
|
|
})
|
|
bubble = dialog._stream_bubble
|
|
assert "当前 AI 模型整理历史资料" in bubble._raw_payload
|
|
assert f"已完成 {completed}/{total}" in bubble._raw_payload
|
|
assert "完整病历尚未提交综合分析" in bubble._raw_payload
|
|
receipt = bubble.transfer_status.text()
|
|
assert all(f"{label}：已整理，待提交" in receipt
|
|
for _, label in module._REQUIRED_TRANSFER_SECTIONS)
|
|
assert len(manifest) == 13
|
|
assert dialog._stream_meta["required_sections"] == manifest
|
|
assert "诊单 23 条" in receipt and "禁止显示" not in receipt
|
|
assert "资料正文已提交" not in receipt
|
|
assert dialog._asking and dialog._stream_first_delta_at is None
|
|
|
|
|
|
@pytest.mark.parametrize("completed,total", [
|
|
(None, 3), (0, None), (-1, 3), (4, 3), (0, 0), (-1, -2),
|
|
(True, 3), (1, False), (1.0, 3), (1, 3.0), ("1", 3), (1, "3"),
|
|
(float("nan"), 3), (1, float("inf")),
|
|
({"private": "禁止显示的患者临床信息"}, 3),
|
|
(1, ["禁止显示的患者临床信息"]),
|
|
])
|
|
def test_invalid_compacting_counts_are_not_rendered(dialog, completed, total):
|
|
dialog._ask("合成资料核对")
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": "compacting",
|
|
"completed": completed, "total": total, "message": "禁止显示的患者临床信息",
|
|
})
|
|
bubble = dialog._stream_bubble
|
|
assert bubble._raw_payload == (
|
|
"正在由当前 AI 模型整理历史资料，完整病历尚未提交综合分析…"
|
|
)
|
|
assert "已完成" not in bubble.transfer_status.text()
|
|
assert "禁止显示" not in bubble.body.toPlainText() + bubble.transfer_status.text()
|
|
|
|
|
|
def test_late_preparation_progress_never_replaces_received_answer(dialog):
|
|
dialog._ask("合成资料核对")
|
|
emit(dialog, {"event": "delta", "text": "合成正文"})
|
|
for status in ("aggregating", "aggregated", "compacting", "ready"):
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": status,
|
|
"completed": 1, "total": 2, "message": "禁止显示的患者临床信息",
|
|
})
|
|
assert dialog._stream_bubble._raw_payload == "合成正文"
|
|
emit(dialog, {"event": "delta", "text": "未刷新尾段"})
|
|
emit(dialog, {"event": "done"})
|
|
assert dialog._stream_bubble._raw_payload == "合成正文未刷新尾段"
|
|
assert dialog.send_button.isEnabled() and not dialog._asking
|
|
|
|
|
|
@pytest.mark.parametrize("ending", ["done", "error", "eof", "timeout", "cancelled"])
|
|
def test_compacting_exits_on_terminal_path_and_ignores_late_preparation(dialog, ending):
|
|
dialog._ask("合成资料核对")
|
|
emit(dialog, {
|
|
"event": "progress", "phase": "preparing", "status": "compacting",
|
|
"completed": 1, "total": 2,
|
|
})
|
|
generation, stream = dialog._generation, dialog._stream_generation
|
|
worker, bubble = dialog._stream_worker, dialog._stream_bubble
|
|
if ending == "done":
|
|
emit(dialog, {"event": "done", "answer": "合成最终正文"})
|
|
assert bubble._raw_payload == "合成最终正文"
|
|
elif ending == "error":
|
|
dialog._stream_failed(generation, stream, RuntimeError("合成失败"))
|
|
assert "本次请求失败" in bubble.transfer_status.text()
|
|
elif ending == "eof":
|
|
dialog._stream_finished(generation, stream, worker)
|
|
assert bubble._raw_payload == module.AI_STREAM_SILENT_TEXT
|
|
elif ending == "timeout":
|
|
dialog._stream_timed_out()
|
|
assert module.AI_STREAM_TIMEOUT_TEXT in bubble.transfer_status.text()
|
|
assert worker.is_cancelled()
|
|
else:
|
|
emit(dialog, {"event": "progress", "phase": "cancelled", "cancel_requested": True})
|
|
dialog._stream_failed(generation, stream, ApiBusinessError(
|
|
"已取消", code=409, data={"code": "AI_ASSISTANT_CANCELLED"},
|
|
))
|
|
assert "取消" in bubble.transfer_status.text()
|
|
assert not dialog._asking and dialog.send_button.isEnabled()
|
|
assert not dialog._stream_timeout_timer.isActive()
|
|
assert not dialog._stream_elapsed_timer.isActive()
|
|
before = bubble._raw_payload, bubble.transfer_status.text(), bubble.stamp.text()
|
|
dialog._stream_event(generation, stream, {
|
|
"event": "progress", "phase": "preparing", "status": "compacting",
|
|
"completed": 2, "total": 2,
|
|
})
|
|
assert (bubble._raw_payload, bubble.transfer_status.text(), bubble.stamp.text()) == before
|