277 lines
14 KiB
Python
277 lines
14 KiB
Python
"""Synthetic Qt regressions for batch receipts and isolated prescription results."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from copy import deepcopy
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, Qt, QTimer
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QLabel, QPushButton
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
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),
|
|
))
|
|
repository = SimpleNamespace(
|
|
create_prescription=lambda *_: pytest.fail("test must never create a prescription"),
|
|
generate_ai_prescription=lambda *_: pytest.fail("test must never invoke an upstream"),
|
|
cancel_diagnosis_ai=lambda *_: {"cancel_requested": True},
|
|
)
|
|
instance = module.AiConsultDialog(repository, PermissionSet([
|
|
"tcm.diagnosis/aiAssistant", "tcm.diagnosis/chufang",
|
|
"tcm.diagnosis/aiGeneratePrescription",
|
|
]))
|
|
monkeypatch.setattr(instance, "_load_workspace", lambda _: None)
|
|
monkeypatch.setattr(module, "show_toast", lambda *_: None)
|
|
instance.open_for(diagnosis_id=501, patient_id=1501)
|
|
instance.resize(1280, 820)
|
|
instance.show()
|
|
application.processEvents()
|
|
yield instance
|
|
for timer in instance.findChildren(QTimer):
|
|
timer.stop()
|
|
instance.close()
|
|
instance.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
application.processEvents()
|
|
|
|
|
|
def sections(status):
|
|
return [{"key": key, "status": status, "availability": "present"}
|
|
for key, _ in module._REQUIRED_TRANSFER_SECTIONS]
|
|
|
|
|
|
def receipt(status="submitted", nested=False):
|
|
data = {
|
|
"items": [
|
|
{"kind": "image", "label": "图片附件1", "status": status},
|
|
{"kind": "image", "label": "图片附件2", "status": "rejected", "reason": "file_type"},
|
|
],
|
|
"attachment_count": 2, "submitted_attachment_count": int(status == "submitted"),
|
|
"rejected_attachment_count": 1, "attachment_transfer": "uploaded_files",
|
|
"attachment_mode": "batch_evidence", "attachment_batch_count": 3,
|
|
"transfer_manifest_final": True,
|
|
}
|
|
manifest = sections("submitted" if status == "submitted" else "failed")
|
|
if nested:
|
|
data["source_summary"] = {"required_sections": manifest}
|
|
else:
|
|
data["required_sections"] = manifest
|
|
return data
|
|
|
|
|
|
def emit(dialog, payload):
|
|
dialog._stream_event(dialog._generation, dialog._stream_generation, payload)
|
|
|
|
|
|
def assert_collapsed(bubble):
|
|
assert bubble.transfer_status.isHidden() and not bubble.transfer_toggle.isChecked()
|
|
assert "展开详情" in bubble.transfer_toggle.text()
|
|
assert bubble._inner.itemAt(0).widget() is bubble.transfer_receipt
|
|
assert len(bubble.findChildren(QLabel, "AiConsultTransferStatus")) == 1
|
|
QTest.mouseClick(bubble.transfer_toggle, Qt.MouseButton.LeftButton)
|
|
assert bubble.transfer_status.isVisible()
|
|
QTest.mouseClick(bubble.transfer_toggle, Qt.MouseButton.LeftButton)
|
|
assert bubble.transfer_status.isHidden()
|
|
|
|
|
|
@pytest.mark.parametrize("nested", [False, True])
|
|
@pytest.mark.parametrize("outcome", ["success", "error"])
|
|
def test_prescription_receipt_is_local_collapsed_and_cannot_mutate_active_question(
|
|
dialog, monkeypatch, nested, outcome,
|
|
):
|
|
dialog._ask("独立的当前合成问答")
|
|
emit(dialog, {"event": "progress", "phase": "attachments", "status": "submitted",
|
|
"items": [{"kind": "document", "label": "文档附件9", "status": "submitted"}],
|
|
"attachment_count": 1, "submitted_attachment_count": 1,
|
|
"required_sections": sections("submitted")})
|
|
current_bubble, current_worker = dialog._stream_bubble, dialog._stream_worker
|
|
current_meta = deepcopy(dialog._stream_meta)
|
|
current_text = current_bubble.transfer_status.text()
|
|
status = dialog._append_bubble("ai", "合成处方等待状态")
|
|
editor_calls = []
|
|
monkeypatch.setattr(dialog, "_open_prescription_editor_from_chat",
|
|
lambda *args: editor_calls.append(args))
|
|
payload = receipt("submitted" if outcome == "success" else "failed", nested)
|
|
original = deepcopy(payload)
|
|
if outcome == "success":
|
|
result = {**payload, "diagnosis_id": 501, "task": "prescription_generate",
|
|
"prescription_draft": {"clinical_diagnosis": "合成诊断",
|
|
"herbs": [{"name": "合成药材", "dosage": 1}]}}
|
|
dialog._ai_prescription_ready(dialog._generation, 501, status, {}, result)
|
|
assert len(editor_calls) == 1
|
|
assert "AI 已生成处方草稿" in status._raw_payload
|
|
assert "处方草稿资料处理已完成。" in status.transfer_status.text()
|
|
else:
|
|
payload["code"] = "REQUIRED_FILES_REJECTED"
|
|
original = deepcopy(payload)
|
|
dialog._prescription_action_error(dialog._generation, status, ApiBusinessError(
|
|
"合成附件拒绝", data=payload,
|
|
), prefix="AI处方草稿生成失败")
|
|
assert editor_calls == []
|
|
assert status._raw_payload == "AI处方草稿生成失败：合成附件拒绝"
|
|
assert "本次请求失败，资料或回复未完成。" in status.transfer_status.text()
|
|
text = status.transfer_status.text()
|
|
assert "附件处理：同一模型已进行 3 批提交尝试。" in text
|
|
assert "图片附件2（图片）：上游拒收，未读取（格式不支持）" in text
|
|
assert "文档附件9" not in text
|
|
assert f"本请求附件通道：{int(outcome == 'success')} 项" in text
|
|
assert payload == original
|
|
assert dialog._stream_meta == current_meta
|
|
assert dialog._stream_worker is current_worker and dialog._stream_bubble is current_bubble
|
|
assert current_bubble.transfer_status.text() == current_text
|
|
assert dialog._asking and not dialog._stream_completed
|
|
assert_collapsed(status)
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", ["success", "error"])
|
|
def test_prescription_without_manifest_cannot_borrow_active_question_receipt(dialog, monkeypatch, outcome):
|
|
dialog._ask("独立的合成问答")
|
|
emit(dialog, {"event": "progress", "phase": "attachments", "status": "submitted", **receipt()})
|
|
current_meta = deepcopy(dialog._stream_meta)
|
|
status = dialog._append_bubble("ai", "合成处方等待状态")
|
|
monkeypatch.setattr(dialog, "_open_prescription_editor_from_chat", lambda *_: None)
|
|
if outcome == "success":
|
|
dialog._ai_prescription_ready(dialog._generation, 501, status, {}, {
|
|
"diagnosis_id": 501, "task": "prescription_generate",
|
|
"prescription_draft": {"clinical_diagnosis": "合成诊断",
|
|
"herbs": [{"name": "合成药材", "dosage": 1}]},
|
|
})
|
|
else:
|
|
dialog._prescription_action_error(dialog._generation, status, ApiBusinessError(
|
|
"合成无清单错误", data={"code": "REQUIRED_FILES_REJECTED"},
|
|
), prefix="合成失败")
|
|
assert not status.findChildren(QLabel, "AiConsultTransferStatus")
|
|
assert dialog._stream_meta == current_meta
|
|
|
|
|
|
def test_prescription_button_cancels_old_stream_before_beginning_owned_action(dialog, monkeypatch, application):
|
|
begin = []
|
|
monkeypatch.setattr(dialog, "_begin_prescription_action", lambda *args: begin.append(args))
|
|
dialog._ask("旧的合成问答")
|
|
old_worker, old_bubble = dialog._stream_worker, dialog._stream_bubble
|
|
old_generation, old_stream = dialog._generation, dialog._stream_generation
|
|
emit(dialog, {"event": "delta", "text": "旧问答已收到正文"})
|
|
next(button for button in dialog.findChildren(QPushButton)
|
|
if button.text() == "AI 生成处方").click()
|
|
assert old_worker.is_cancelled()
|
|
assert dialog._stream_generation > old_stream
|
|
assert dialog._stream_worker is None and not dialog._asking
|
|
assert dialog._stream_meta == {}
|
|
assert "旧问答已收到正文" in old_bubble._raw_payload
|
|
assert "上游是否停止尚未确认" in old_bubble.transfer_status.text()
|
|
application.processEvents()
|
|
assert len(begin) == 1
|
|
generation, status = begin[0]
|
|
assert generation == dialog._generation and status is not old_bubble
|
|
expected = status._raw_payload
|
|
dialog._stream_event(old_generation, old_stream, {"event": "done", "answer": "不应外显的旧结果"})
|
|
dialog._stream_failed(old_generation, old_stream, RuntimeError("不应外显的旧错误"))
|
|
dialog._stream_finished(old_generation, old_stream, old_worker)
|
|
assert status._raw_payload == expected and dialog._stream_worker is None
|
|
|
|
|
|
@pytest.mark.parametrize("nested", [False, True])
|
|
@pytest.mark.parametrize(("phase", "status"), [
|
|
("attachments", "uploading"), ("attachments", "submitted"), ("upstream", "waiting"),
|
|
])
|
|
def test_batch_progress_never_presents_prepared_full_text_as_already_submitted(dialog, nested, phase, status):
|
|
dialog._ask("合成批处理问答")
|
|
payload = receipt(nested=nested)
|
|
payload["transfer_manifest_final"] = False
|
|
payload["attachment_batch_count"] = 2
|
|
if nested:
|
|
payload["source_summary"]["required_sections"] = sections("prepared")
|
|
else:
|
|
payload["required_sections"] = sections("prepared")
|
|
emit(dialog, {"event": "progress", "phase": phase, "status": status, **payload})
|
|
bubble = dialog._stream_bubble
|
|
text = bubble.transfer_status.text()
|
|
assert text.splitlines()[0] == "正在分批提交附件并提取资料，完整病历尚未提交综合分析…"
|
|
assert bubble._raw_payload == text.splitlines()[0]
|
|
assert "基本信息：已整理，待提交" in text
|
|
assert "基本信息：已提交" not in text
|
|
assert "资料正文已提交" not in text and "资料已提交，正在等待 AI 正文" not in text
|
|
assert dialog._stream_meta["attachment_mode"] == "batch_evidence"
|
|
assert dialog._stream_meta["attachment_batch_count"] == 2
|
|
assert dialog._asking and not dialog._stream_completed
|
|
assert_collapsed(bubble)
|
|
# Once the complete body is submitted, normal synthesis waiting is accurate.
|
|
emit(dialog, {"event": "progress", "phase": "upstream", "status": "waiting",
|
|
"required_sections": sections("submitted")})
|
|
assert "基本信息：已提交" in bubble.transfer_status.text()
|
|
assert "完整病历尚未提交" not in bubble.transfer_status.text()
|
|
|
|
|
|
@pytest.mark.parametrize(("reason", "expected"), [
|
|
("file_count", "（数量限制）"), ("file_type", "（格式不支持）"),
|
|
("file_size", "（大小限制）"), ("file_access", "（附件不可读取）"),
|
|
("private test filename https://example.test/hidden.png?secret=synthetic", ""),
|
|
("<b>synthetic-private-error</b>", ""),
|
|
])
|
|
def test_rejected_receipt_uses_only_safe_reason_codes_and_stays_collapsed(dialog, reason, expected):
|
|
dialog._ask("合成拒绝回执")
|
|
payload = receipt()
|
|
payload["items"][1].update({"reason": reason, "label": "synthetic-private-filename.png"})
|
|
emit(dialog, {"event": "progress", "phase": "attachments", "status": "submitted", **payload})
|
|
bubble = dialog._stream_bubble
|
|
text = bubble.transfer_status.text()
|
|
assert f"图片附件2（图片）：上游拒收，未读取{expected}" in text
|
|
assert "synthetic-private" not in text and "https://" not in text
|
|
assert bubble.transfer_status.textFormat() == Qt.TextFormat.PlainText
|
|
assert "本请求附件通道：1 项" in text
|
|
assert_collapsed(bubble)
|
|
|
|
|
|
@pytest.mark.parametrize("with_skipped", [False, True])
|
|
@pytest.mark.parametrize("reported", [None, True, -1, "1"])
|
|
def test_rejected_and_failed_files_never_inflate_fallback_submitted_count(dialog, with_skipped, reported):
|
|
items = [
|
|
{"kind": "image", "label": "图片附件1", "status": "submitted"},
|
|
{"kind": "image", "label": "图片附件2", "status": "rejected", "reason": "file_type"},
|
|
{"kind": "image", "label": "图片附件3", "status": "failed"},
|
|
]
|
|
if with_skipped:
|
|
items.append({"kind": "image", "label": "图片附件4", "status": "skipped", "reason": "not_found"})
|
|
dialog._ask("合成计数回执")
|
|
emit(dialog, {"event": "progress", "phase": "attachments", "status": "submitted",
|
|
"items": items, "attachment_count": len(items), "submitted_attachment_count": reported})
|
|
assert "本请求附件通道：1 项" in dialog._stream_bubble.transfer_status.text()
|
|
|
|
|
|
def test_nonfinal_error_manifest_keeps_batch_rejections_and_explicit_outcome_counts(dialog):
|
|
dialog._ask("合成错误回执纠正")
|
|
emit(dialog, {"event": "progress", "phase": "attachments", "status": "submitted",
|
|
"items": [{"kind": "image", "label": "图片附件1", "status": "submitted"}],
|
|
"attachment_count": 1, "submitted_attachment_count": 1})
|
|
failed = receipt("failed")
|
|
failed.pop("transfer_manifest_final")
|
|
dialog._stream_failed(dialog._generation, dialog._stream_generation, ApiBusinessError(
|
|
"合成批处理失败", data={"code": "REQUIRED_FILES_REJECTED", **failed},
|
|
))
|
|
text = dialog._stream_bubble.transfer_status.text()
|
|
assert "图片附件1（图片）：提交失败" in text
|
|
assert "图片附件2（图片）：上游拒收，未读取（格式不支持）" in text
|
|
assert "本请求附件通道：0 项" in text
|
|
assert dialog._stream_meta["rejected_attachment_count"] == 1
|
|
assert dialog._stream_meta["attachment_mode"] == "batch_evidence"
|
|
assert dialog._stream_meta["attachment_batch_count"] == 3
|
|
assert not dialog._asking and not dialog._stream_completed
|
|
assert_collapsed(dialog._stream_bubble)
|