255 lines
10 KiB
Python
255 lines
10 KiB
Python
"""Long replies retain content and chart snapshots while sharing GUI time."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, Qt, QTimer
|
|
from PySide6.QtGui import QImage, QPainter, QTextDocument
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QLabel
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.ui.dialogs import ai_consult as module
|
|
from doctor_workstation.ui.dialogs.ai_consult_charts import GlucoseReportCard
|
|
from doctor_workstation.ui.dialogs.ai_consult_metric_state import (
|
|
ABNORMAL_COLOR,
|
|
NORMAL_COLOR,
|
|
UNKNOWN_COLOR,
|
|
)
|
|
from doctor_workstation.ui.dialogs.ai_consult_report_visuals import (
|
|
build_report_visuals,
|
|
create_report_visual,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.mark.parametrize("rich", [False, True])
|
|
def test_wrapping_measurements_reuse_parsed_text_and_follow_text_font_width(application, rich):
|
|
text = ("核对原始资料及记录日期。" * 70) + "末尾标记"
|
|
if rich:
|
|
text = f'<div style="line-height:172%">{text}<br><b>末行</b></div>'
|
|
label = module._FittedText(text, rich=rich)
|
|
changes = []
|
|
label._measure_document.contentsChanged.connect(lambda: changes.append(True))
|
|
first = label._height_at_width(480)
|
|
parses = len(changes)
|
|
for _ in range(20):
|
|
assert label._height_at_width(480) == first
|
|
assert label._height_at_width(240) > first
|
|
assert len(changes) == parses # Layout negotiation must not keep reparsing HTML.
|
|
label.resize(480, 100)
|
|
label._fit()
|
|
measured_width = label._measure_document.textWidth()
|
|
for hypothetical in range(1, 800, 7):
|
|
assert label.heightForWidth(hypothetical) == label.minimumHeight()
|
|
assert label._measure_document.textWidth() == measured_width
|
|
|
|
font = label.font()
|
|
font.setPixelSize(22)
|
|
label.setFont(font)
|
|
changed = label._height_at_width(240)
|
|
oracle = QTextDocument()
|
|
oracle.setDefaultFont(font)
|
|
if rich:
|
|
oracle.setHtml(text)
|
|
else:
|
|
oracle.setPlainText(text)
|
|
oracle.setTextWidth(240)
|
|
assert changed == int(oracle.size().height()) + 1
|
|
label.setText("更短的记录")
|
|
assert label._height_at_width(240) < changed
|
|
label.deleteLater()
|
|
|
|
|
|
@pytest.mark.parametrize("payload", [
|
|
"## 标题\n\n- **重点**：核对原始记录。\n- 完整末行",
|
|
"<h2>标题</h2><p>完整末行</p>",
|
|
{"summary": "完整末行"}, "",
|
|
])
|
|
def test_rich_update_restores_document_layout_and_keeps_full_content(application, payload):
|
|
browser = module._RichMessage("ai")
|
|
browser.resize(420, 200)
|
|
browser.set_payload(payload)
|
|
assert browser.document().isLayoutEnabled()
|
|
if payload:
|
|
assert "完整末行" in browser.toPlainText()
|
|
assert browser.minimumHeight() >= int(browser.document().size().height())
|
|
browser.document().setLayoutEnabled(False)
|
|
browser.set_payload(payload)
|
|
assert not browser.document().isLayoutEnabled()
|
|
browser.deleteLater()
|
|
|
|
|
|
def test_failed_rich_render_does_not_leave_document_layout_disabled(application, monkeypatch):
|
|
browser = module._RichMessage("ai")
|
|
|
|
def fail(*args, **kwargs):
|
|
raise ValueError("synthetic render failure")
|
|
|
|
monkeypatch.setattr(module, "render_chat_payload", fail)
|
|
with pytest.raises(ValueError, match="synthetic"):
|
|
browser.set_payload("未完成的内容")
|
|
assert browser.document().isLayoutEnabled()
|
|
assert not browser._fitting
|
|
browser.deleteLater()
|
|
|
|
|
|
@pytest.fixture
|
|
def dialog(application, monkeypatch):
|
|
instance = module.AiConsultDialog(object(), PermissionSet(["tcm.diagnosis/aiAssistant"]))
|
|
monkeypatch.setattr(instance, "_load_workspace", lambda _generation: None)
|
|
instance.open_for(diagnosis_id=501, patient_id=1501)
|
|
instance._workspace_sections = {"tracking": {"value": {"diagnosis_id": 501, "blood_records": [
|
|
{"diagnosis_id": 501, "patient_id": 1501, "record_date": "2026-10-17",
|
|
"fasting_blood_sugar": 6.2, "fasting_blood_sugar_status": "normal"},
|
|
{"diagnosis_id": 501, "patient_id": 1501, "record_date": "2026-10-18",
|
|
"fasting_blood_sugar": 8.2, "fasting_blood_sugar_status": "high"},
|
|
]}}}
|
|
instance.resize(1280, 820)
|
|
instance.show()
|
|
yield instance
|
|
for timer in instance.findChildren(QTimer):
|
|
timer.stop()
|
|
instance.close()
|
|
instance.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
application.processEvents()
|
|
|
|
|
|
def _complete_long_reply(dialog):
|
|
repeated = "核对血糖原始记录及时间，不补写缺失资料。" * 280
|
|
reply = ("完整合成资料：空腹血糖 6.2 mmol/L，另一条空腹血糖 8.2 mmol/L，糖化血红蛋白未记录。\n\n"
|
|
f"1. 血糖记录核对\n- 来源：{repeated}\n\n2. 随访记录\n- 复核：{repeated}\n- 独有末尾标记")
|
|
bubble = dialog._append_bubble("ai", "")
|
|
dialog._stream_bubble = bubble
|
|
generation, stream = dialog._generation, dialog._stream_generation
|
|
for start in range(0, len(reply), 137):
|
|
dialog._stream_event(generation, stream, {"event": "delta", "text": reply[start:start + 137]})
|
|
assert dialog._pending_chunks # Exercise a large unflushed tail at done.
|
|
dialog._stream_event(generation, stream, {"event": "done", "model_label": "千问"})
|
|
return bubble, reply, generation, stream
|
|
|
|
|
|
def _wait_for_report(bubble):
|
|
for _ in range(600):
|
|
if not bubble._report_render_pending:
|
|
return
|
|
QTest.qWait(5)
|
|
pytest.fail("completed report remained pending")
|
|
|
|
|
|
def test_deferred_long_report_keeps_tail_chart_snapshot_and_terminal_semantics(dialog):
|
|
bubble, reply, generation, stream = _complete_long_reply(dialog)
|
|
assert bubble._report_render_pending and bubble._clinical_panel is None
|
|
assert bubble._raw_payload == reply == dialog._stream_text
|
|
assert "独有末尾标记" in bubble.body.toPlainText()
|
|
assert dialog.send_button.isEnabled() and not dialog._asking
|
|
dialog._workspace_sections["tracking"]["value"]["blood_records"][0]["fasting_blood_sugar"] = 99
|
|
_wait_for_report(bubble)
|
|
assert bubble._clinical_panel is not None
|
|
card = bubble.findChild(GlucoseReportCard)
|
|
assert card.chart.data.readings[0].fasting == 6.2
|
|
intro = bubble._clinical_panel.findChild(QLabel, "AiConsultReportIntroBody").text()
|
|
assert all(color in intro for color in (NORMAL_COLOR, ABNORMAL_COLOR, UNKNOWN_COLOR))
|
|
dialog._stream_event(generation, stream, {"event": "delta", "text": "迟到内容"})
|
|
assert bubble._raw_payload == reply
|
|
assert len(bubble.findChildren(GlucoseReportCard)) == 1
|
|
|
|
|
|
@pytest.mark.parametrize("exit_path", ["patient", "close"])
|
|
def test_queued_report_cannot_render_after_patient_switch_or_close(dialog, exit_path):
|
|
bubble, _, _, _ = _complete_long_reply(dialog)
|
|
if exit_path == "patient":
|
|
dialog.open_for(diagnosis_id=502, patient_id=1502)
|
|
else:
|
|
dialog.close()
|
|
_wait_for_report(bubble)
|
|
assert bubble._clinical_panel is None
|
|
|
|
|
|
@pytest.mark.parametrize("field, changed", [
|
|
("patient_id", 1999), ("_chart_patient_id", 1999), ("diagnosis_id", 502),
|
|
])
|
|
def test_authoritative_owner_change_in_same_generation_cannot_attach_old_charts(dialog, field, changed):
|
|
bubble, reply, generation, _ = _complete_long_reply(dialog)
|
|
setattr(dialog, field, changed)
|
|
assert dialog._generation == generation
|
|
_wait_for_report(bubble)
|
|
assert bubble._clinical_panel is None
|
|
assert bubble.findChild(GlucoseReportCard) is None
|
|
assert bubble._raw_payload == reply
|
|
|
|
|
|
def test_new_request_keeps_previous_completed_report_and_its_frozen_records(dialog):
|
|
bubble, reply, generation, old_stream = _complete_long_reply(dialog)
|
|
dialog._cancel_stream()
|
|
current = dialog._append_bubble("ai", "")
|
|
dialog._stream_bubble = current
|
|
dialog._stream_event(generation, dialog._stream_generation, {"event": "delta", "text": "新的提问回复"})
|
|
dialog._stream_event(generation, old_stream, {"event": "done", "answer": "旧请求迟到内容"})
|
|
_wait_for_report(bubble)
|
|
assert bubble._clinical_panel is not None and bubble._raw_payload == reply
|
|
assert current._raw_payload == "新的提问回复"
|
|
assert current._clinical_panel is None
|
|
|
|
|
|
@pytest.mark.parametrize("width", [1080, 1600])
|
|
def test_deferred_long_report_tail_stays_reachable_after_wrapping(dialog, application, width):
|
|
dialog.resize(width, 820)
|
|
bubble, _, _, _ = _complete_long_reply(dialog)
|
|
_wait_for_report(bubble)
|
|
for _ in range(8):
|
|
QTest.qWait(5)
|
|
dialog.chat_scroll.verticalScrollBar().setValue(dialog.chat_scroll.verticalScrollBar().maximum())
|
|
for _ in range(8):
|
|
QTest.qWait(5)
|
|
tail = bubble._clinical_panel.findChildren(QLabel, "AiConsultReportBody")[-1]
|
|
assert "独有末尾标记" in tail.text()
|
|
bottom = tail.mapTo(dialog.chat_scroll.viewport(), tail.rect().bottomLeft())
|
|
assert dialog.chat_scroll.viewport().rect().contains(bottom)
|
|
assert dialog.chat_scroll.horizontalScrollBar().maximum() == 0
|
|
|
|
|
|
def test_trend_paint_and_hover_share_one_geometry_scan(application, monkeypatch):
|
|
rows = [
|
|
{"diagnosis_id": 501, "patient_id": 1501,
|
|
"record_date": (datetime(2026, 8, 1) + timedelta(days=day)).date().isoformat(),
|
|
"systolic_pressure": 130 + day % 12, "diastolic_pressure": 80 + day % 8}
|
|
for day in range(60)
|
|
]
|
|
data = build_report_visuals("血压", "", sections={"tracking": {"value": {
|
|
"diagnosis_id": 501, "blood_records": rows,
|
|
}}}, diagnosis_id=501, patient_id=1501)[0]
|
|
card = create_report_visual(data)
|
|
plot = card.chart
|
|
plot.resize(640, plot.height())
|
|
original = plot._geometry
|
|
calls = []
|
|
|
|
def geometry():
|
|
calls.append(True)
|
|
return original()
|
|
|
|
monkeypatch.setattr(plot, "_geometry", geometry)
|
|
image = QImage(plot.size(), QImage.Format.Format_ARGB32)
|
|
image.fill(Qt.GlobalColor.white)
|
|
painter = QPainter(image)
|
|
try:
|
|
plot._paint_trend(painter)
|
|
finally:
|
|
painter.end()
|
|
assert len(calls) == 1
|
|
position = plot._position(data.items[0])
|
|
calls.clear()
|
|
assert data.items[0].value_text in plot._point_tooltip(position)
|
|
assert len(calls) == 1
|
|
card.deleteLater()
|