221 lines
9.0 KiB
Python
221 lines
9.0 KiB
Python
"""Live Markdown uses the approved character queue, with synthetic content only."""
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import pytest
|
||
from PySide6.QtCore import QCoreApplication, QEvent
|
||
from PySide6.QtGui import QTextBlockUserData, QTextCursor
|
||
from PySide6.QtWidgets import QApplication
|
||
|
||
from doctor_workstation.ui.dialogs import ai_consult as module
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application():
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
@pytest.fixture
|
||
def bubble(application):
|
||
widget = module._ChatBubble(role="ai", text=module.AI_STREAM_WAITING_TEXT)
|
||
yield widget
|
||
widget.stop_stream_presentation()
|
||
widget.deleteLater()
|
||
QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||
|
||
|
||
def tick(bubble):
|
||
bubble._presentation_timer.stop()
|
||
bubble._presentation_timer.timeout.emit()
|
||
|
||
|
||
def drain(bubble):
|
||
for _ in range(12000):
|
||
if not bubble._presentation_timer.isActive():
|
||
return
|
||
tick(bubble)
|
||
pytest.fail("synthetic character queue exceeded finite bound")
|
||
|
||
|
||
@pytest.mark.parametrize("chunk_size", [1, 7, 10000])
|
||
def test_heading_bold_and_list_render_before_terminal(bubble, chunk_size):
|
||
answer = ("### 血糖趋势评估\n\n**1. 可比较的血糖记录**\n\n"
|
||
"* **空腹血糖**：10左右 mmol/L。来源：[vital_signs]、[present_illness]。\n"
|
||
"* **餐后血糖**：未记录。来源：[vital_signs]。")
|
||
for end in range(chunk_size, len(answer) + chunk_size, chunk_size):
|
||
accepted = answer[:end]
|
||
bubble.queue_approved_stream_text(accepted)
|
||
assert bubble._raw_payload == accepted
|
||
drain(bubble)
|
||
assert not bubble._presentation_terminal
|
||
assert bubble.body.document().begin().blockFormat().headingLevel() == 3
|
||
assert bubble.body.document().find("可比较的血糖记录").charFormat().fontWeight() >= 700
|
||
fasting = bubble.body.document().find("空腹血糖")
|
||
assert fasting.charFormat().fontWeight() >= 700
|
||
assert fasting.block().textList() is not None
|
||
view = bubble.body.toPlainText()
|
||
assert all(marker not in view for marker in ("###", "**", "* "))
|
||
assert "10左右 mmol/L" in view and "[vital_signs]" in view
|
||
assert "餐后血糖：未记录" in view
|
||
assert bubble._raw_payload == answer
|
||
bubble.complete_stream_presentation(bubble.finish_stream_text)
|
||
oracle = module._RichMessage("ai")
|
||
oracle.set_payload(answer)
|
||
assert bubble.body.toHtml() == oracle.toHtml()
|
||
oracle.deleteLater()
|
||
|
||
|
||
def test_later_markdown_formats_shown_prefix_without_revealing_backlog(bubble):
|
||
initial = "普通合成正文"
|
||
bubble.queue_approved_stream_text(initial)
|
||
drain(bubble)
|
||
answer = initial + "\n\n### 新小节\n\n**新重点** 尚未轮到的尾部"
|
||
bubble.queue_approved_stream_text(answer)
|
||
assert bubble.body.toPlainText() == initial
|
||
tick(bubble)
|
||
assert bubble._presentation_offset == len(initial) + 1
|
||
assert "新小节" not in bubble.body.toPlainText()
|
||
assert "尚未轮到" not in bubble.body.toPlainText()
|
||
drain(bubble)
|
||
assert bubble.body.document().find("新小节").blockFormat().headingLevel() == 3
|
||
assert bubble.body.document().find("新重点").charFormat().fontWeight() >= 700
|
||
assert "尚未轮到的尾部" in bubble.body.toPlainText()
|
||
assert bubble._raw_payload == answer
|
||
|
||
|
||
def test_long_tail_keeps_format_and_parses_later_inline_syntax(bubble):
|
||
source = "**已有重点** " + "甲" * 2200
|
||
bubble.queue_approved_stream_text(source)
|
||
drain(bubble)
|
||
assert bubble.body._stream_tail_plain
|
||
assert "**" not in bubble.body.toPlainText()
|
||
assert bubble.body.document().find("已有重点").charFormat().fontWeight() >= 700
|
||
answer = source + " **后来重点** 唯一结尾"
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
assert "**" not in bubble.body.toPlainText()
|
||
assert bubble.body.document().find("后来重点").charFormat().fontWeight() >= 700
|
||
assert bubble.body.document().find("唯一结尾").charFormat().fontWeight() < 700
|
||
assert bubble.body.toPlainText().count("甲") == 2200
|
||
assert bubble.body.toPlainText().endswith("唯一结尾")
|
||
assert bubble._raw_payload == answer
|
||
|
||
|
||
@pytest.mark.parametrize("opening, closing", [("**", "**"), ("`", "`"), ("[", "](https://example.test)"), ("_", "_")])
|
||
def test_long_tail_reclassifies_delimiters_before_ordinary_letters(bubble, opening, closing):
|
||
source = opening + "x" * 2050 + closing
|
||
bubble.queue_approved_stream_text(source)
|
||
drain(bubble)
|
||
answer = source + "TAIL"
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
oracle = module._RichMessage("ai")
|
||
oracle.set_payload(answer)
|
||
assert bubble.body.toPlainText() == oracle.toPlainText()
|
||
actual = bubble.body.document().find("TAIL").charFormat()
|
||
expected = oracle.document().find("TAIL").charFormat()
|
||
assert actual.fontWeight() == expected.fontWeight()
|
||
assert actual.fontItalic() == expected.fontItalic()
|
||
assert actual.fontFamilies() == expected.fontFamilies()
|
||
assert actual.isAnchor() == expected.isAnchor()
|
||
assert bubble._raw_payload == answer
|
||
oracle.deleteLater()
|
||
|
||
|
||
def test_plain_character_in_long_list_only_formats_its_changed_block(application, monkeypatch):
|
||
browser = module._RichMessage("ai")
|
||
browser.append_stream_text("- synthetic item\n" * 2000 + "- last item")
|
||
visited = []
|
||
original = module.apply_reading_rhythm
|
||
|
||
def apply(view, *, role, start_position=0):
|
||
block = view.document().findBlock(start_position)
|
||
while block.isValid():
|
||
visited.append(block.blockNumber())
|
||
block = block.next()
|
||
original(view, role=role, start_position=start_position)
|
||
|
||
monkeypatch.setattr(module, "apply_reading_rhythm", apply)
|
||
browser.append_stream_text("x")
|
||
assert len(visited) == 1
|
||
assert visited[0] >= 2000
|
||
assert browser.toPlainText().endswith("last itemx")
|
||
browser.deleteLater()
|
||
|
||
|
||
@pytest.mark.parametrize("answer", [
|
||
"[name][id]\n\n[id]: https://example.test",
|
||
"<https://example.test>",
|
||
"---",
|
||
r"a\*b",
|
||
])
|
||
def test_other_markdown_syntax_is_recognized_before_terminal(bubble, answer):
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
oracle = module._RichMessage("ai")
|
||
oracle.set_payload(answer)
|
||
assert bubble.body.toPlainText() == oracle.toPlainText()
|
||
assert not bubble._presentation_terminal
|
||
if "example.test" in answer:
|
||
label = "name" if answer.startswith("[") else "https://example.test"
|
||
assert bubble.body.document().find(label).charFormat().anchorHref() == "https://example.test"
|
||
assert bubble._raw_payload == answer
|
||
oracle.deleteLater()
|
||
|
||
|
||
def test_character_updates_keep_stable_blocks_and_selection(bubble, monkeypatch):
|
||
prefix = "### 已有标题\n\n已有正文。\n\n"
|
||
bubble.queue_approved_stream_text(prefix)
|
||
drain(bubble)
|
||
block = bubble.body.document().begin()
|
||
marker = QTextBlockUserData()
|
||
block.setUserData(marker)
|
||
cursor = bubble.body.textCursor()
|
||
cursor.setPosition(0)
|
||
cursor.setPosition(4, QTextCursor.MoveMode.KeepAnchor)
|
||
bubble.body.setTextCursor(cursor)
|
||
selected = bubble.body.textCursor().selectedText()
|
||
monkeypatch.setattr(bubble.body, "set_payload", lambda _: pytest.fail("whole reply reparse"))
|
||
answer = prefix + "**新增重点** 完整尾部"
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
assert bubble.body.document().begin().userData() is marker
|
||
assert bubble.body.textCursor().selectedText() == selected
|
||
assert bubble.body.document().find("新增重点").charFormat().fontWeight() >= 700
|
||
assert bubble._raw_payload == answer
|
||
|
||
|
||
@pytest.mark.parametrize("answer", [
|
||
"### 合成🩺标题\r\n\r\n**e\u0301👩\u200d⚕️🇨🇳**\r\n\r\n尾部",
|
||
"```text\n### 原始标记 **原文**\n- 原文条目\n```\n\n**完整末尾**",
|
||
r"\# 原文 \*原文\*，血糖 <7.0 mmol/L，数值 -1。",
|
||
"| 日期 | 数值 |\n| --- | --- |\n| 合成日 | 8.2 |\n\n**表格尾部**",
|
||
])
|
||
def test_unicode_code_and_literal_content_survive_live_formatting(bubble, answer):
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
oracle = module._RichMessage("ai")
|
||
oracle.set_payload(answer)
|
||
assert bubble.body.toPlainText().split() == oracle.toPlainText().split()
|
||
assert bubble._raw_payload == answer
|
||
bubble.complete_stream_presentation(bubble.finish_stream_text)
|
||
assert bubble.body.toHtml() == oracle.toHtml()
|
||
oracle.deleteLater()
|
||
|
||
|
||
@pytest.mark.parametrize("answer", [
|
||
'<h2>合成标题</h2><p>完整末尾</p>',
|
||
'{"summary":"合成完整末尾","value":8.2}',
|
||
])
|
||
def test_complete_html_and_json_keep_terminal_auto_detection(bubble, answer):
|
||
bubble.queue_approved_stream_text(answer)
|
||
drain(bubble)
|
||
assert bubble._raw_payload == answer
|
||
bubble.complete_stream_presentation(bubble.finish_stream_text)
|
||
assert "完整末尾" in bubble.body.toPlainText()
|
||
assert "<h2>" not in bubble.body.toPlainText()
|
||
assert '{"summary"' not in bubble.body.toPlainText()
|