239 lines
10 KiB
Python
239 lines
10 KiB
Python
"""Active streams keep parsed prefixes and deliver small deltas without GUI floods."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from threading import Event
|
|
from time import monotonic
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, QThreadPool, QTimer
|
|
from PySide6.QtGui import QTextBlockUserData, QTextCursor
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.ui.dialogs import ai_consult as module
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application():
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.mark.parametrize("width", [1, 7, 127])
|
|
@pytest.mark.parametrize("body", [
|
|
"# 核对记录\n\n**完整前段**，含中文与 emoji 🩺。\n\n- 第一项\n- 第二项\n\n末尾唯一标记",
|
|
"# 合成代码\n\n```text\n第一行\n\n第二行\n```\n\n完整结尾",
|
|
"# 合成表格\n\n| 记录 | 数值 |\n| --- | --- |\n| 测试 | 6.8 |\n\n待核对",
|
|
])
|
|
def test_incremental_markdown_preserves_content_and_terminal_format(application, width, body):
|
|
bubble = module._ChatBubble(role="ai", text=module.AI_STREAM_WAITING_TEXT)
|
|
full = ""
|
|
for offset in range(0, len(body), width):
|
|
chunk = body[offset:offset + width]
|
|
full += chunk
|
|
bubble.append_stream_text(chunk, full)
|
|
assert bubble._raw_payload == body
|
|
preview = bubble.body.toPlainText()
|
|
assert "完整" in preview or "待核对" in preview
|
|
bubble.finish_stream_text()
|
|
oracle = module._RichMessage("ai")
|
|
oracle.set_payload(body)
|
|
assert bubble.body.toPlainText() == oracle.toPlainText()
|
|
assert bubble.body.toHtml() == oracle.toHtml()
|
|
bubble.deleteLater()
|
|
oracle.deleteLater()
|
|
|
|
|
|
def test_updates_keep_stable_paragraph_objects_and_user_selection(application, monkeypatch):
|
|
browser = module._RichMessage("ai")
|
|
browser.append_stream_text("# 已有标题\n\n已有正文，不重新解析。\n\n")
|
|
prefix = browser.document().begin()
|
|
marker = QTextBlockUserData()
|
|
prefix.setUserData(marker)
|
|
cursor = browser.textCursor()
|
|
cursor.setPosition(0)
|
|
cursor.setPosition(4, QTextCursor.MoveMode.KeepAnchor)
|
|
browser.setTextCursor(cursor)
|
|
selected = browser.textCursor().selectedText()
|
|
monkeypatch.setattr(module, "render_chat_payload", lambda *args, **kwargs: pytest.fail("reparsed whole stream"))
|
|
for _ in range(90):
|
|
browser.append_stream_text("持续新增的一小段。")
|
|
assert browser.document().begin().userData() is marker
|
|
assert browser.document().begin().text() == "已有标题"
|
|
assert browser.textCursor().selectedText() == selected
|
|
assert browser.toPlainText().count("持续新增的一小段。") == 90
|
|
browser.deleteLater()
|
|
|
|
|
|
def test_long_unfinished_block_is_complete_and_formatted_on_completion(application):
|
|
bubble = module._ChatBubble(role="ai", text="")
|
|
text = "**长段落** " + "核对来源。" * 1500 + "末尾唯一标记"
|
|
full = ""
|
|
for offset in range(0, len(text), 73):
|
|
chunk = text[offset:offset + 73]
|
|
full += chunk
|
|
bubble.append_stream_text(chunk, full)
|
|
assert bubble.body._stream_tail_plain
|
|
assert bubble.body.toPlainText().count("末尾唯一标记") == 1
|
|
bubble.finish_stream_text()
|
|
assert "**" not in bubble.body.toPlainText()
|
|
assert bubble._raw_payload == text
|
|
bubble.deleteLater()
|
|
|
|
|
|
def test_partial_render_failure_restores_document_layout(application, monkeypatch):
|
|
browser = module._RichMessage("ai")
|
|
def fail(*args, **kwargs):
|
|
raise ValueError("synthetic format failure")
|
|
monkeypatch.setattr(module, "apply_reading_rhythm", fail)
|
|
with pytest.raises(ValueError, match="synthetic"):
|
|
browser.append_stream_text("合成片段")
|
|
assert browser.document().isLayoutEnabled() and not browser._fitting
|
|
browser.deleteLater()
|
|
|
|
|
|
@pytest.mark.parametrize("suffix", [" ", "\t", " "])
|
|
def test_unterminated_whitespace_is_not_a_stable_paragraph(application, suffix):
|
|
first = "正文\n" + suffix
|
|
assert module._stable_markdown_prefix(first) == 0
|
|
browser = module._RichMessage("ai")
|
|
browser.append_stream_text(first)
|
|
browser.append_stream_text("续行")
|
|
oracle = module._RichMessage("ai")
|
|
oracle.set_payload(first + "续行")
|
|
assert browser.toPlainText() == oracle.toPlainText()
|
|
assert browser.document().blockCount() == oracle.document().blockCount()
|
|
browser.deleteLater()
|
|
oracle.deleteLater()
|
|
|
|
|
|
def test_heading_size_and_list_spacing_are_stable_during_generation(application):
|
|
browser = module._RichMessage("ai")
|
|
for text in ["普通前段\n\n", "## 小节标题\n\n", "- 第一条\n- 第二条"]:
|
|
browser.append_stream_text(text)
|
|
block = browser.document().begin().next()
|
|
assert block.text() == "小节标题" and block.blockFormat().headingLevel() == 2
|
|
separator = block.next()
|
|
assert not separator.text()
|
|
assert separator.blockFormat().topMargin() == separator.blockFormat().bottomMargin() == 0
|
|
assert separator.blockFormat().lineHeight() == 1
|
|
assert "第一条" in browser.toPlainText() and "第二条" in browser.toPlainText()
|
|
browser.deleteLater()
|
|
|
|
|
|
def test_long_fence_can_close_after_plain_preview_without_losing_lines(application):
|
|
browser = module._RichMessage("ai")
|
|
source = "## 标题 🩺\n\n```text\n" + "唯一源码一行\n\n" * 350
|
|
browser.append_stream_text(source)
|
|
assert browser._stream_tail_plain and browser._stream_tail.startswith("```text")
|
|
browser.append_stream_text("```\n\n## 后续小节\n\n唯一结尾")
|
|
assert browser.toPlainText().count("唯一源码一行") == 350
|
|
assert browser.toPlainText().count("唯一结尾") == 1
|
|
assert "```" not in browser.toPlainText()
|
|
browser.deleteLater()
|
|
|
|
|
|
@pytest.mark.parametrize("exit_path", ["replace", "patient", "close"])
|
|
def test_queued_mailbox_cannot_change_new_request_or_closed_patient(application, monkeypatch, exit_path):
|
|
monkeypatch.setattr(module, "QThreadPool", SimpleNamespace(globalInstance=lambda: SimpleNamespace(start=lambda _: None)))
|
|
dialog = module.AiConsultDialog(object(), PermissionSet(["tcm.diagnosis/aiAssistant"]))
|
|
monkeypatch.setattr(dialog, "_load_workspace", lambda _: None)
|
|
dialog.open_for(diagnosis_id=501, patient_id=1501)
|
|
dialog._ask("旧问题")
|
|
old_worker = dialog._stream_worker
|
|
old_worker.signals.blockSignals(True)
|
|
old_worker._queue_event({"event": "delta", "text": "旧患者排队的尾段"})
|
|
if exit_path == "replace":
|
|
dialog._ask("新问题", replace_active=True)
|
|
elif exit_path == "patient":
|
|
dialog.open_for(diagnosis_id=502, patient_id=1502)
|
|
dialog._ask("新问题")
|
|
else:
|
|
dialog.close()
|
|
current = dialog._stream_bubble
|
|
expected = current._raw_payload if current is not None else None
|
|
old_worker.signals.blockSignals(False)
|
|
old_worker.signals.ready.emit()
|
|
old_worker._queue_event({"event": "delta", "text": "取消后迟到片段"})
|
|
QTest.qWait(10)
|
|
assert old_worker.is_cancelled() and old_worker.take_events() == []
|
|
if current is not None:
|
|
assert current._raw_payload == expected == module.AI_STREAM_PENDING_TEXT
|
|
assert dialog._session_closed is (exit_path == "close")
|
|
dialog.close()
|
|
dialog.deleteLater()
|
|
|
|
|
|
def test_worker_coalesces_thousands_of_deltas_before_the_gui(application):
|
|
worker = module._AiStreamWorker(object(), diagnosis_id=501, prompt="合成测试", task="summary")
|
|
wakes = []
|
|
worker.signals.ready.connect(lambda: wakes.append(True))
|
|
worker._queue_event({"event": "start", "model_label": "千问"})
|
|
chunks = ["中", "🩺", "\n", " ", "**"] * 2000
|
|
for chunk in chunks:
|
|
worker._queue_event({"event": "delta", "text": chunk})
|
|
worker._queue_event({"event": "done", "answer": "".join(chunks)})
|
|
events = worker.take_events()
|
|
assert [event["event"] for event in events] == ["start", "delta", "done"]
|
|
assert events[1]["text"] == "".join(chunks)
|
|
assert len(wakes) == 3 # start, first real text, terminal; independent of chunk count.
|
|
assert worker.take_events() == []
|
|
worker._queue_event({"event": "delta", "text": "过期片段"})
|
|
worker.cancel()
|
|
worker._queue_event({"event": "delta", "text": "取消后片段"})
|
|
assert worker.take_events() == []
|
|
|
|
|
|
@pytest.mark.parametrize("ending", ["done", "error", "eof"])
|
|
def test_paused_upstream_tail_paints_and_terminal_drain_is_lossless(application, monkeypatch, ending):
|
|
paused, release = Event(), Event()
|
|
class Repository:
|
|
def stream_diagnosis_ai(self, *args, **kwargs):
|
|
yield {"event": "start", "model_label": "千问"}
|
|
yield {"event": "delta", "text": "首段即时显示"}
|
|
yield {"event": "delta", "text": "，暂停前的尾段"}
|
|
paused.set()
|
|
assert release.wait(4)
|
|
yield {"event": "delta", "text": "，终态前的最后一字"}
|
|
if ending == "done":
|
|
yield {"event": "done"}
|
|
elif ending == "error":
|
|
raise RuntimeError("合成连接中断")
|
|
|
|
dialog = module.AiConsultDialog(Repository(), PermissionSet(["tcm.diagnosis/aiAssistant"]))
|
|
monkeypatch.setattr(dialog, "_load_workspace", lambda _: None)
|
|
dialog.open_for(diagnosis_id=501, patient_id=1501)
|
|
dialog.show()
|
|
try:
|
|
dialog._ask("合成小结")
|
|
deadline = monotonic() + 2
|
|
while (not paused.is_set() or "暂停前的尾段" not in dialog._stream_text) and monotonic() < deadline:
|
|
QTest.qWait(10)
|
|
assert "暂停前的尾段" in dialog._stream_bubble.body.toPlainText()
|
|
assert dialog._asking and not dialog._stream_completed
|
|
# A queued GUI action runs while the network worker is paused.
|
|
actions = []
|
|
QTimer.singleShot(0, lambda: actions.append(True))
|
|
QTest.qWait(10)
|
|
assert actions == [True]
|
|
release.set()
|
|
deadline = monotonic() + 2
|
|
while dialog._asking and monotonic() < deadline:
|
|
QTest.qWait(10)
|
|
assert not dialog._asking and dialog.send_button.isEnabled()
|
|
assert "终态前的最后一字" in dialog._stream_bubble.body.toPlainText()
|
|
assert not dialog._delivery_timer.isActive()
|
|
assert dialog._stream_completed is (ending == "done")
|
|
finally:
|
|
release.set()
|
|
for timer in dialog.findChildren(QTimer):
|
|
timer.stop()
|
|
dialog.close()
|
|
QThreadPool.globalInstance().waitForDone(2000)
|
|
dialog.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|