更新
This commit is contained in:
Your Name
@@ -2,13 +2,16 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from threading import Event
|
||||
from time import monotonic
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from PySide6.QtCore import QPoint, Qt
|
||||
from PySide6.QtCore import QPoint, Qt, QThread
|
||||
from PySide6.QtTest import QTest
|
||||
from PySide6.QtWidgets import (
|
||||
QApplication,
|
||||
@@ -21,7 +24,7 @@ from PySide6.QtWidgets import (
|
||||
)
|
||||
|
||||
from doctor_workstation.core import PermissionSet
|
||||
from doctor_workstation.services import DemoDoctorRepository
|
||||
from doctor_workstation.services import ApiClient, DemoDoctorRepository, RemoteDoctorRepository
|
||||
from doctor_workstation.ui.dialogs import ai_consult as ai_consult_module
|
||||
from doctor_workstation.ui.dialogs.ai_consult import (
|
||||
AiConsultDialog,
|
||||
@@ -1577,6 +1580,356 @@ def test_answered_stream_replaces_the_pending_placeholder(
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_stream_clock_covers_preparation_first_text_and_completion(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
now = [100.0]
|
||||
monkeypatch.setattr(ai_consult_module, "monotonic", lambda: now[0])
|
||||
dialog._ask("请生成问诊小结")
|
||||
bubble = dialog._stream_bubble
|
||||
assert bubble is not None
|
||||
generation, stream_generation = dialog._generation, dialog._stream_generation
|
||||
assert "已等待 0 秒" in bubble.stamp.text()
|
||||
assert dialog._stream_elapsed_timer.isActive()
|
||||
|
||||
now[0] = 103.0
|
||||
dialog._stream_elapsed_timer.timeout.emit()
|
||||
assert "已等待 3 秒" in bubble.stamp.text()
|
||||
dialog._stream_event(generation, stream_generation, {"event": "start"})
|
||||
assert "已等待 3 秒" in bubble.stamp.text()
|
||||
|
||||
now[0] = 104.2
|
||||
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "问诊小结"})
|
||||
# No timer flush or done event is needed to display the first text.
|
||||
assert bubble._raw_payload == "问诊小结"
|
||||
assert "生成中" in bubble.stamp.text()
|
||||
assert "已用时 4 秒" in bubble.stamp.text()
|
||||
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "：请补充舌脉。"})
|
||||
assert dialog._flush_timer.isActive()
|
||||
|
||||
now[0] = 108.7
|
||||
dialog._stream_event(
|
||||
generation, stream_generation, {"event": "done", "model_label": "千问"}
|
||||
)
|
||||
assert bubble._raw_payload == "问诊小结：请补充舌脉。"
|
||||
assert "千问" in bubble.stamp.text()
|
||||
assert "首字 4.2 秒" in bubble.stamp.text()
|
||||
assert "耗时 8.7 秒" in bubble.stamp.text()
|
||||
assert not dialog._stream_elapsed_timer.isActive()
|
||||
assert not dialog._flush_timer.isActive()
|
||||
stamp = bubble.stamp.text()
|
||||
now[0] = 130.0
|
||||
dialog._stream_finished(generation, stream_generation, dialog._stream_worker)
|
||||
dialog._update_stream_elapsed()
|
||||
dialog._stream_event(generation, stream_generation, {"event": "delta", "text": "迟到内容"})
|
||||
assert bubble.stamp.text() == stamp
|
||||
assert bubble._raw_payload == "问诊小结：请补充舌脉。"
|
||||
assert dialog.send_button.isEnabled()
|
||||
dialog.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ending", ["error", "timeout", "silent", "switch", "close"])
|
||||
def test_stream_clock_stops_on_every_exit(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
ending: str,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
now = [100.0]
|
||||
monkeypatch.setattr(ai_consult_module, "monotonic", lambda: now[0])
|
||||
dialog._ask("请总结病情")
|
||||
bubble = dialog._stream_bubble
|
||||
assert bubble is not None
|
||||
now[0] = 112.4
|
||||
generation, stream_generation = dialog._generation, dialog._stream_generation
|
||||
worker = dialog._stream_worker
|
||||
if ending == "error":
|
||||
dialog._stream_failed(generation, stream_generation, RuntimeError("模型繁忙"))
|
||||
dialog._stream_finished(generation, stream_generation, worker)
|
||||
elif ending == "timeout":
|
||||
dialog._stream_timed_out()
|
||||
elif ending == "silent":
|
||||
dialog._stream_finished(generation, stream_generation, worker)
|
||||
elif ending == "switch":
|
||||
dialog.open_for(diagnosis_id=502, patient_id=302)
|
||||
else:
|
||||
dialog.close()
|
||||
assert not dialog._stream_elapsed_timer.isActive()
|
||||
assert not dialog._stream_timeout_timer.isActive()
|
||||
assert not dialog._asking
|
||||
assert dialog.send_button.isEnabled()
|
||||
if ending in {"error", "timeout", "silent"}:
|
||||
assert "耗时 12.4 秒" in bubble.stamp.text()
|
||||
stamp = bubble.stamp.text()
|
||||
now[0] = 140.0
|
||||
dialog._update_stream_elapsed()
|
||||
assert bubble.stamp.text() == stamp
|
||||
if ending == "switch":
|
||||
dialog._ask("新患者的问题")
|
||||
assert dialog._stream_bubble is not None
|
||||
assert "已等待 0 秒" in dialog._stream_bubble.stamp.text()
|
||||
assert dialog._stream_first_delta_at is None
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_http_stream_paints_before_upstream_finishes_on_the_ui_thread(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""Exercise HTTP parsing, repository, real worker and Qt delivery together."""
|
||||
release = Event()
|
||||
waiting_for_end = Event()
|
||||
|
||||
class DelayedStream(httpx.SyncByteStream):
|
||||
def __iter__(self):
|
||||
first = (
|
||||
'event: start\r\ndata: {"model_label":"千问"}\r\n\r\n'
|
||||
'event: delta\r\ndata: {"text":"先收到的内容"}\r\n\r\n'
|
||||
).encode()
|
||||
# A network chunk may end partway through a Chinese character.
|
||||
split = first.index("先".encode()) + 1
|
||||
yield first[:split]
|
||||
yield first[split:]
|
||||
waiting_for_end.set()
|
||||
if not release.wait(5):
|
||||
raise AssertionError("UI did not receive the first streamed text")
|
||||
yield (
|
||||
'event: delta\r\ndata: {"text":"，然后完成。"}\r\n\r\n'
|
||||
'event: done\r\ndata: {"model_label":"千问"}\r\n\r\n'
|
||||
).encode()
|
||||
|
||||
client = ApiClient(
|
||||
"https://example.test",
|
||||
transport=httpx.MockTransport(
|
||||
lambda request: httpx.Response(
|
||||
200, headers={"content-type": "text/event-stream"}, stream=DelayedStream()
|
||||
)
|
||||
),
|
||||
)
|
||||
dialog = AiConsultDialog(RemoteDoctorRepository(client))
|
||||
dialog.diagnosis_id = 501
|
||||
dialog._context_notice_shown = True
|
||||
delivered_on = []
|
||||
original_event = dialog._stream_event
|
||||
|
||||
def record_event(*args: Any) -> None:
|
||||
delivered_on.append(QThread.currentThread())
|
||||
original_event(*args)
|
||||
|
||||
monkeypatch.setattr(dialog, "_stream_event", record_event)
|
||||
dialog.show()
|
||||
try:
|
||||
dialog._ask("请生成问诊小结")
|
||||
deadline = monotonic() + 3
|
||||
while (not dialog._stream_text or not waiting_for_end.is_set()) and monotonic() < deadline:
|
||||
QTest.qWait(10)
|
||||
bubble = dialog._stream_bubble
|
||||
assert bubble is not None and bubble.body is not None
|
||||
assert waiting_for_end.is_set()
|
||||
assert bubble.body.toPlainText() == "先收到的内容"
|
||||
assert not dialog._stream_completed
|
||||
assert dialog._asking
|
||||
assert "生成中" in bubble.stamp.text()
|
||||
release.set()
|
||||
deadline = monotonic() + 3
|
||||
while dialog._asking and monotonic() < deadline:
|
||||
QTest.qWait(10)
|
||||
assert dialog._stream_completed
|
||||
assert not dialog._asking
|
||||
assert bubble.body.toPlainText() == "先收到的内容，然后完成。"
|
||||
assert delivered_on and all(thread == application.thread() for thread in delivered_on)
|
||||
assert not dialog._stream_elapsed_timer.isActive()
|
||||
finally:
|
||||
release.set()
|
||||
dialog.close()
|
||||
client.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("terminal", ["done", "error"])
|
||||
def test_every_quick_tool_can_start_after_terminal_event_before_worker_exit(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
terminal: str,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
dialog.show()
|
||||
application.processEvents()
|
||||
cards = dialog.findChildren(ai_consult_module._ClickCard, "AiConsultToolCard")
|
||||
assert len(cards) == len(ai_consult_module.TOOL_ITEMS) == 6
|
||||
previous = []
|
||||
for card, (_kind, label, prompt) in zip(cards, ai_consult_module.TOOL_ITEMS, strict=True):
|
||||
# Exercise the text surface the doctor clicks, including event propagation.
|
||||
target = next(child for child in card.findChildren(QLabel) if child.text() == label)
|
||||
QTest.mouseClick(target, Qt.MouseButton.LeftButton)
|
||||
worker = dialog._stream_worker
|
||||
assert worker is not None and worker.prompt == prompt
|
||||
assert dialog._asking
|
||||
generation, stream_generation = dialog._generation, dialog._stream_generation
|
||||
for old_generation, old_stream_generation, old_worker in previous:
|
||||
dialog._stream_finished(old_generation, old_stream_generation, old_worker)
|
||||
assert dialog._stream_worker is worker
|
||||
assert dialog._asking
|
||||
previous.append((generation, stream_generation, worker))
|
||||
if terminal == "done":
|
||||
dialog._stream_event(generation, stream_generation, {"event": "done", "answer": "已完成"})
|
||||
else:
|
||||
dialog._stream_failed(generation, stream_generation, RuntimeError("模型繁忙"))
|
||||
# The transport may take longer to close; it must not keep the UI locked.
|
||||
assert not dialog._asking
|
||||
assert dialog.send_button.isEnabled()
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_clicking_another_quick_tool_switches_the_active_request(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
notices = []
|
||||
monkeypatch.setattr(ai_consult_module, "show_toast", lambda *args: notices.append(args[1]))
|
||||
dialog.show()
|
||||
application.processEvents()
|
||||
cards = dialog.findChildren(ai_consult_module._ClickCard, "AiConsultToolCard")
|
||||
QTest.mouseClick(cards[0], Qt.MouseButton.LeftButton)
|
||||
old_worker, old_bubble = dialog._stream_worker, dialog._stream_bubble
|
||||
old_generation, old_stream_generation = dialog._generation, dialog._stream_generation
|
||||
assert old_worker is not None and old_bubble is not None
|
||||
dialog._stream_event(old_generation, old_stream_generation, {"event": "delta", "text": "已收到的分析"})
|
||||
|
||||
QTest.mouseClick(cards[1], Qt.MouseButton.LeftButton)
|
||||
worker, bubble = dialog._stream_worker, dialog._stream_bubble
|
||||
assert worker is not None and worker is not old_worker
|
||||
assert worker.prompt == ai_consult_module.TOOL_ITEMS[1][2]
|
||||
assert old_worker.is_cancelled()
|
||||
assert "已收到的分析" in old_bubble._raw_payload
|
||||
assert "已停止" in old_bubble._raw_payload
|
||||
assert bubble is not None and bubble is not old_bubble
|
||||
assert bubble._raw_payload == ai_consult_module.AI_STREAM_PENDING_TEXT
|
||||
assert "已等待 0 秒" in bubble.stamp.text()
|
||||
assert dialog._asking
|
||||
|
||||
dialog._stream_event(old_generation, old_stream_generation, {"event": "done", "answer": "旧结果"})
|
||||
dialog._stream_failed(old_generation, old_stream_generation, RuntimeError("旧错误"))
|
||||
dialog._stream_finished(old_generation, old_stream_generation, old_worker)
|
||||
assert dialog._stream_worker is worker
|
||||
assert dialog._asking
|
||||
assert bubble._raw_payload == ai_consult_module.AI_STREAM_PENDING_TEXT
|
||||
|
||||
# Repeated clicks on the same running tool do not submit duplicate requests.
|
||||
QTest.mouseClick(cards[1], Qt.MouseButton.LeftButton)
|
||||
assert dialog._stream_worker is worker
|
||||
assert not worker.is_cancelled()
|
||||
assert notices and "正在生成" in notices[-1]
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_stream_worker_does_not_wait_for_another_event_after_done() -> None:
|
||||
continued_after_done = []
|
||||
|
||||
class DoneThenIdleRepository:
|
||||
def stream_diagnosis_ai(self, *args: Any, **kwargs: Any):
|
||||
yield {"event": "done", "answer": "已完成"}
|
||||
continued_after_done.append(True)
|
||||
raise AssertionError("the terminal event must end iteration")
|
||||
|
||||
worker = ai_consult_module._AiStreamWorker(
|
||||
DoneThenIdleRepository(), diagnosis_id=501, prompt="请分析", task="custom"
|
||||
)
|
||||
events, errors, finished = [], [], []
|
||||
worker.signals.event.connect(events.append)
|
||||
worker.signals.error.connect(errors.append)
|
||||
worker.signals.finished.connect(lambda: finished.append(True))
|
||||
worker.run()
|
||||
assert [event["event"] for event in events] == ["done"]
|
||||
assert continued_after_done == []
|
||||
assert errors == []
|
||||
assert finished == [True]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("entry", ["tool", "chip", "menu", "suggestion"])
|
||||
def test_quick_actions_switch_requests_and_reveal_the_chat(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
entry: str,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
dialog.show()
|
||||
application.processEvents()
|
||||
dialog._ask("之前的请求")
|
||||
old_worker = dialog._stream_worker
|
||||
assert old_worker is not None
|
||||
dialog.tabs.setCurrentIndex(1)
|
||||
dialog._set_chat_filter("doctor")
|
||||
if entry == "tool":
|
||||
target = dialog.findChildren(ai_consult_module._ClickCard, "AiConsultToolCard")[0]
|
||||
prompt = ai_consult_module.TOOL_ITEMS[0][2]
|
||||
QTest.mouseClick(target, Qt.MouseButton.LeftButton)
|
||||
elif entry == "chip":
|
||||
target = next(button for button in dialog.findChildren(QPushButton) if button.text() == "总结当前病情")
|
||||
prompt = ai_consult_module.QUICK_PROMPTS[0][1]
|
||||
target.click()
|
||||
elif entry == "menu":
|
||||
more = next(button for button in dialog.findChildren(QPushButton) if button.text() == "更多")
|
||||
prompt = ai_consult_module.QUICK_PROMPTS[3][1]
|
||||
more.menu().actions()[0].trigger()
|
||||
else:
|
||||
target = dialog.findChildren(ai_consult_module._ClickCard, "AiConsultSuggestRow")[0]
|
||||
prompt = f"请针对该问题给出问诊话术与判断要点：{ai_consult_module.SUGGESTIONS[0]}"
|
||||
QTest.mouseClick(target, Qt.MouseButton.LeftButton)
|
||||
assert old_worker.is_cancelled()
|
||||
assert dialog._stream_worker is not None and dialog._stream_worker.prompt == prompt
|
||||
assert dialog.tabs.currentIndex() == dialog.center_stack.currentIndex() == 0
|
||||
assert dialog._chat_filter == "all"
|
||||
assert dialog._stream_bubble is not None and dialog._stream_bubble.isVisible()
|
||||
dialog.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("exit_action", ["back", "side", "escape", "accept", "reject", "close"])
|
||||
def test_all_dialog_exit_paths_cancel_stream_and_allow_a_fresh_session(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
exit_action: str,
|
||||
) -> None:
|
||||
dialog = _silent_dialog(monkeypatch)
|
||||
dialog.show()
|
||||
application.processEvents()
|
||||
dialog._ask("请分析病情")
|
||||
worker = dialog._stream_worker
|
||||
assert worker is not None
|
||||
generation, stream_generation = dialog._generation, dialog._stream_generation
|
||||
if exit_action == "back":
|
||||
next(button for button in dialog.findChildren(QPushButton) if button.text() == "返回").click()
|
||||
elif exit_action == "side":
|
||||
dialog.findChild(QPushButton, "AiConsultSideClose").click()
|
||||
elif exit_action == "escape":
|
||||
QTest.keyClick(dialog, Qt.Key.Key_Escape)
|
||||
else:
|
||||
getattr(dialog, exit_action)()
|
||||
assert not dialog.isVisible()
|
||||
assert worker.is_cancelled()
|
||||
assert not dialog._asking
|
||||
assert not dialog._stream_timeout_timer.isActive()
|
||||
assert not dialog._stream_elapsed_timer.isActive()
|
||||
closed_generation = dialog._generation
|
||||
assert closed_generation > generation
|
||||
dialog.close()
|
||||
assert dialog._generation == closed_generation
|
||||
|
||||
dialog.open_for(diagnosis_id=502, patient_id=302)
|
||||
dialog.show()
|
||||
dialog._ask("新患者的问题")
|
||||
new_worker, bubble = dialog._stream_worker, dialog._stream_bubble
|
||||
assert new_worker is not None and not new_worker.is_cancelled()
|
||||
dialog._stream_event(generation, stream_generation, {"event": "done", "answer": "旧患者结果"})
|
||||
dialog._stream_finished(generation, stream_generation, worker)
|
||||
assert dialog._stream_worker is new_worker
|
||||
assert dialog._asking
|
||||
assert bubble is not None and bubble._raw_payload == ai_consult_module.AI_STREAM_PENDING_TEXT
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_ai_consult_window_can_be_maximized(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
|
||||
@@ -7,7 +7,7 @@ from typing import Any
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
import pytest
|
||||
from PySide6.QtCore import QDate, Signal
|
||||
from PySide6.QtCore import QDate, QEvent, Signal
|
||||
from PySide6.QtWidgets import (
|
||||
QApplication,
|
||||
QDialog,
|
||||
@@ -19,7 +19,7 @@ from PySide6.QtWidgets import (
|
||||
QWidget,
|
||||
)
|
||||
|
||||
from doctor_workstation.core import PermissionSet
|
||||
from doctor_workstation.core import PermissionSet, UserProfile
|
||||
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
|
||||
from doctor_workstation.ui.pages import consultations as consultations_module
|
||||
from doctor_workstation.ui.pages.consultations import (
|
||||
@@ -669,7 +669,7 @@ def test_switching_rows_invalidates_prescription_worker_and_clears_busy(
|
||||
application.processEvents()
|
||||
|
||||
|
||||
def test_native_call_does_not_reuse_video_qr_permission(
|
||||
def test_list_video_entry_uses_watch_permission_not_video_qr_permission(
|
||||
application: QApplication,
|
||||
) -> None:
|
||||
class VideoRepository:
|
||||
@@ -686,13 +686,17 @@ def test_native_call_does_not_reuse_video_qr_permission(
|
||||
pass
|
||||
|
||||
live_row = _row(
|
||||
assistant_id=66,
|
||||
video_call_hint={
|
||||
"state": "live",
|
||||
"label": "视频通话进行中",
|
||||
"start_time": 1787102100,
|
||||
}
|
||||
)
|
||||
page = ConsultationsPage(VideoRepository(), permissions=PermissionSet([]))
|
||||
page = ConsultationsPage(
|
||||
VideoRepository(), permissions=PermissionSet(["tcm.diagnosis/watchCall"]),
|
||||
current_user={"id": 66, "role_id": 2},
|
||||
)
|
||||
emitted: list[dict[str, Any]] = []
|
||||
page.video_requested.connect(emitted.append)
|
||||
page.table.set_rows([live_row])
|
||||
@@ -738,10 +742,13 @@ def test_video_join_action_is_hidden_until_doctor_session_is_live(
|
||||
def end_call(self, diagnosis_id: int) -> None:
|
||||
pass
|
||||
|
||||
page = ConsultationsPage(VideoRepository(), permissions=PermissionSet([]))
|
||||
page = ConsultationsPage(
|
||||
VideoRepository(), permissions=PermissionSet(["tcm.diagnosis/watchCall"]),
|
||||
current_user={"id": 66, "role_id": 2},
|
||||
)
|
||||
emitted: list[dict[str, Any]] = []
|
||||
page.video_requested.connect(emitted.append)
|
||||
page.table.set_rows([_row(video_call_hint=hint)])
|
||||
page.table.set_rows([_row(assistant_id=66, video_call_hint=hint)])
|
||||
page.table.selectRow(0)
|
||||
application.processEvents()
|
||||
|
||||
@@ -752,6 +759,92 @@ def test_video_join_action_is_hidden_until_doctor_session_is_live(
|
||||
assert emitted == []
|
||||
page.close()
|
||||
application.processEvents()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("user", [
|
||||
{"id": 66, "role_id": 1},
|
||||
{"id": 66, "role_id": "1"},
|
||||
{"id": 66, "role_ids": [1, 2]},
|
||||
{"id": 66, "role_id": "1,2"},
|
||||
{"id": 66, "roles": [{"id": 1}]},
|
||||
{"id": 66, "role_id": 1, "root": True},
|
||||
UserProfile.from_dict({"id": 66, "role_id": [1, 2]}),
|
||||
])
|
||||
@pytest.mark.parametrize("state,status", [
|
||||
("live", "通话中"), ("pending_room", "接通中"), ("none", "—"),
|
||||
])
|
||||
def test_doctor_video_column_is_readonly_even_with_full_permission(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
user: Any,
|
||||
state: str,
|
||||
status: str,
|
||||
) -> None:
|
||||
from doctor_workstation.services import DemoDoctorRepository
|
||||
|
||||
monkeypatch.setattr(consultations_module, "run_async", lambda *_a, **_kw: None)
|
||||
# Even a mixed-role doctor assigned to this row must not get the receiver entry.
|
||||
row = _row(assistant_id=66, video_call_hint={"state": state})
|
||||
page = ConsultationsPage(DemoDoctorRepository(), PermissionSet(["*"]), current_user=user)
|
||||
emitted = []
|
||||
page.video_requested.connect(emitted.append)
|
||||
try:
|
||||
page.table_host.set_rows([row])
|
||||
page.table.selectRow(0)
|
||||
page._selection_changed()
|
||||
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 10))
|
||||
assert cell.findChildren(QToolButton) == []
|
||||
assert status in [label.text() for label in cell.findChildren(QLabel)]
|
||||
assert page.call_toolbar_button.isHidden()
|
||||
assert not page.video_button.isEnabled()
|
||||
page._row_video(row)
|
||||
page.video_button.setEnabled(True)
|
||||
page.video_button.click()
|
||||
assert emitted == []
|
||||
finally:
|
||||
page.close()
|
||||
page.deleteLater()
|
||||
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("user,permissions,assistant_id,allowed", [
|
||||
({"id": 66, "role_id": 2}, ["tcm.diagnosis/watchCall"], 66, True),
|
||||
({"id": 66, "role_id": 2}, ["tcm.diagnosis/videoQr"], 66, False),
|
||||
({"id": 66, "role_id": 2}, [], 66, False),
|
||||
({"id": 66, "role_id": 2}, ["*"], 77, False),
|
||||
({"id": 66, "role_id": 0, "root": True}, ["*"], 77, False),
|
||||
({"id": 66, "role_id": 0, "root": True}, ["*"], 66, True),
|
||||
({"id": 0, "role_id": 2}, ["*"], 0, False),
|
||||
(None, ["*"], 66, False),
|
||||
])
|
||||
def test_list_video_entry_requires_the_assigned_user_and_watch_permission(
|
||||
application: QApplication,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
user: Any,
|
||||
permissions: list[str],
|
||||
assistant_id: int,
|
||||
allowed: bool,
|
||||
) -> None:
|
||||
from doctor_workstation.services import DemoDoctorRepository
|
||||
|
||||
monkeypatch.setattr(consultations_module, "run_async", lambda *_a, **_kw: None)
|
||||
row = _row(assistant_id=assistant_id, video_call_hint={"state": "live"})
|
||||
page = ConsultationsPage(DemoDoctorRepository(), PermissionSet(permissions), current_user=user)
|
||||
emitted = []
|
||||
page.video_requested.connect(emitted.append)
|
||||
try:
|
||||
page.table_host.set_rows([row])
|
||||
page.table.selectRow(0)
|
||||
page._selection_changed()
|
||||
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 10))
|
||||
assert bool(cell.findChildren(QToolButton)) is allowed
|
||||
assert page.video_button.isEnabled() is allowed
|
||||
page._request_video()
|
||||
assert bool(emitted) is allowed
|
||||
finally:
|
||||
page.close()
|
||||
page.deleteLater()
|
||||
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
|
||||
@@ -1121,6 +1121,76 @@ def test_appointment_history_phone_is_fail_closed_without_phone_plain(
|
||||
dialog.close()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("locked", [False, True])
|
||||
def test_doctor_can_save_clinical_fields_without_resubmitting_masked_identity(
|
||||
application: QApplication,
|
||||
locked: bool,
|
||||
) -> None:
|
||||
class MaskedRepository(VisualRepository):
|
||||
def check_diagnosis_phone(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
raise AssertionError("a masked phone must not be checked as plaintext")
|
||||
|
||||
def check_diagnosis_id_card(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
raise AssertionError("a masked identity must not be checked as plaintext")
|
||||
|
||||
repository = MaskedRepository(locked=locked)
|
||||
for source in (repository.detail["diagnosis"], repository.detail["patient"]):
|
||||
source["phone"] = "156****7862"
|
||||
source["id_card"] = "1101**********1234"
|
||||
dialog = _open_dialog(
|
||||
application, (1024, 640), mode="edit", repository=repository,
|
||||
permissions=PermissionSet(["tcm.diagnosis/detail", "tcm.diagnosis/edit"]),
|
||||
)
|
||||
saved = []
|
||||
dialog.saved.connect(lambda: saved.append(True))
|
||||
try:
|
||||
assert dialog.edit_fields["phone"].isReadOnly()
|
||||
assert dialog.edit_fields["phone"].toPlainText() == "156****7862"
|
||||
assert dialog.edit_fields["id_card"].isReadOnly()
|
||||
dialog.edit_fields["chief_complaint"].setPlainText("口渴较前改善")
|
||||
dialog._save()
|
||||
assert saved == [True]
|
||||
assert len(repository.updates) == 1
|
||||
payload = repository.updates[0]
|
||||
assert payload["chief_complaint"] == "口渴较前改善"
|
||||
assert "phone" not in payload and "id_card" not in payload
|
||||
assert "156****7862" not in str(payload)
|
||||
assert "1101**********1234" not in str(payload)
|
||||
finally:
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_plain_phone_permission_keeps_validation_and_duplicate_checks(
|
||||
application: QApplication,
|
||||
) -> None:
|
||||
class PlainRepository(VisualRepository):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.phone_checks = []
|
||||
|
||||
def check_diagnosis_phone(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
self.phone_checks.append(dict(payload))
|
||||
return {"exists": False}
|
||||
|
||||
repository = PlainRepository()
|
||||
dialog = _open_dialog(
|
||||
application, (1024, 640), mode="edit", repository=repository,
|
||||
permissions=PermissionSet(["tcm.diagnosis/edit", "tcm.diagnosis/phonePlain"]),
|
||||
)
|
||||
try:
|
||||
assert not dialog.edit_fields["phone"].isReadOnly()
|
||||
dialog.edit_fields["phone"].setPlainText("156****7862")
|
||||
dialog._save()
|
||||
assert not repository.phone_checks
|
||||
assert not repository.updates
|
||||
dialog.edit_fields["phone"].setPlainText("15600007862")
|
||||
dialog._save()
|
||||
assert repository.phone_checks == [{"id": 501, "phone": "15600007862"}]
|
||||
assert repository.updates[-1]["phone"] == "15600007862"
|
||||
finally:
|
||||
dialog.close()
|
||||
|
||||
|
||||
def test_order_detail_action_requires_exact_permission(application: QApplication) -> None:
|
||||
repository = VisualRepository()
|
||||
dialog = _open_dialog(
|
||||
|
||||
@@ -0,0 +1,96 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||||
|
||||
import pytest
|
||||
from PySide6.QtCore import QEvent
|
||||
from PySide6.QtGui import QFont
|
||||
from PySide6.QtWidgets import QApplication, QLabel, QToolButton, QVBoxLayout, QWidget
|
||||
|
||||
from doctor_workstation.ui.diagnosis_index_widgets import DiagnosisTableHost
|
||||
from doctor_workstation.ui.reception_style import body_family
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def application() -> QApplication:
|
||||
return QApplication.instance() or QApplication([])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tech_blue", [False, True])
|
||||
@pytest.mark.parametrize("size", [(1024, 640), (1710, 920)])
|
||||
@pytest.mark.parametrize("with_channel", [False, True])
|
||||
def test_public_video_status_fits_frozen_cell_with_inherited_item_padding(
|
||||
application: QApplication, tech_blue: bool, size: tuple[int, int],
|
||||
with_channel: bool,
|
||||
) -> None:
|
||||
root = QWidget()
|
||||
font = QFont(body_family())
|
||||
font.setPixelSize(13)
|
||||
root.setFont(font)
|
||||
# Match GLOBAL_QSS without changing the application's theme between tests.
|
||||
root.setStyleSheet("QTableView::item { padding: 6px 8px; border-bottom: 1px solid #E1E9F4; }")
|
||||
layout = QVBoxLayout(root)
|
||||
host = DiagnosisTableHost(
|
||||
parent=root, tech_blue=tech_blue,
|
||||
action_policy={"video_call": True, "view": True, "edit": True},
|
||||
video_entry_allowed=lambda _record: False,
|
||||
)
|
||||
layout.addWidget(host)
|
||||
statuses = [
|
||||
("ended", "视频已结束 · 10-17 14:19"),
|
||||
("ended", "视频已结束 · 10-17 14:27"),
|
||||
("ended", "视频已结束"),
|
||||
("live", "通话中"),
|
||||
("pending_room", "接通中"),
|
||||
("missed", "上次通话未接听"),
|
||||
("cancelled", "上次通话已取消"),
|
||||
("none", "—"),
|
||||
]
|
||||
rows = [
|
||||
{
|
||||
"id": index + 1, "patient_id": index + 100, "patient_name": "测试患者",
|
||||
"gender": 1, "age": 50, "has_appointment": 1,
|
||||
"appointment_id": index + 200, "appointment_status": 1,
|
||||
"appointment_type": "video",
|
||||
"latest_appointment_channel_text": "自媒体4H" if with_channel else "",
|
||||
"appointments": [{"id": index + 200, "status": 1, "doctor_name": "测试医生",
|
||||
"time_text": "2026-10-17 14:15", "appointment_type": "video"}],
|
||||
"video_call_hint": {"state": state, "label": text},
|
||||
}
|
||||
for index, (state, text) in enumerate(statuses * 4)
|
||||
]
|
||||
try:
|
||||
host.set_rows(rows)
|
||||
root.resize(*size)
|
||||
root.show()
|
||||
for _ in range(4):
|
||||
application.processEvents()
|
||||
for index, (_state, text) in enumerate(statuses):
|
||||
cell = host.fixed.indexWidget(host.model.index(index, 10))
|
||||
assert not cell.findChildren(QToolButton)
|
||||
label = cell.findChild(QLabel)
|
||||
assert label is not None
|
||||
assert label.height() >= label.heightForWidth(label.width())
|
||||
assert cell.rect().contains(label.geometry())
|
||||
assert abs(label.geometry().center().y() - cell.rect().center().y()) <= 1
|
||||
assert label.toolTip() == text
|
||||
if " · " in text:
|
||||
assert label.text() == text.replace(" · ", "\n")
|
||||
assert label.fontMetrics().horizontalAdvance("10-17 14:19") <= label.width()
|
||||
assert cell.width() >= host.fixed.columnWidth(10) - 2
|
||||
host.main.horizontalScrollBar().setValue(host.main.horizontalScrollBar().maximum())
|
||||
for fraction in (2, 3):
|
||||
host.main.verticalScrollBar().setValue(host.main.verticalScrollBar().maximum() // fraction)
|
||||
application.processEvents()
|
||||
assert host.fixed.verticalScrollBar().value() == host.main.verticalScrollBar().value()
|
||||
row = host.main.indexAt(host.main.viewport().rect().center()).row()
|
||||
assert row >= 0
|
||||
assert host.main.visualRect(host.model.index(row, 0)).top() == host.fixed.visualRect(
|
||||
host.model.index(row, 10)
|
||||
).top()
|
||||
finally:
|
||||
root.close()
|
||||
root.deleteLater()
|
||||
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
||||
@@ -7,7 +7,7 @@ import pytest
|
||||
from PySide6.QtCore import Qt
|
||||
from PySide6.QtWidgets import QApplication, QLabel, QPushButton, QToolButton
|
||||
|
||||
from doctor_workstation.core import PermissionSet
|
||||
from doctor_workstation.core import PermissionSet, UserProfile
|
||||
from doctor_workstation.core.appointment_modes import (
|
||||
appointment_type_description,
|
||||
appointment_type_value,
|
||||
@@ -76,13 +76,18 @@ def test_appointment_text_chat_carries_exact_ids_and_blocks_video_qr(application
|
||||
page.deleteLater()
|
||||
|
||||
|
||||
def test_consultation_text_entry_does_not_wait_for_live_video(application, monkeypatch):
|
||||
@pytest.mark.parametrize("current_user", [
|
||||
None,
|
||||
{"id": 66, "role_id": 1},
|
||||
UserProfile.from_dict({"id": 66, "role_id": [1, 2]}),
|
||||
])
|
||||
def test_consultation_text_entry_does_not_wait_for_live_video(application, monkeypatch, current_user):
|
||||
monkeypatch.setattr(consultations, "run_async", lambda *_a, **_kw: None)
|
||||
row = {"id": 501, "patient_id": 901, "patient_name": "图文患者", "has_appointment": 1,
|
||||
"appointment_status": 1, "latest_appointment_id": 71,
|
||||
"appointments": [{"id": 71, "status": 1, "appointment_type": "text"}],
|
||||
"video_call_hint": {"state": "none"}}
|
||||
page = consultations.ConsultationsPage(DemoDoctorRepository(), PermissionSet(["*"]))
|
||||
page = consultations.ConsultationsPage(DemoDoctorRepository(), PermissionSet(["*"]), current_user=current_user)
|
||||
try:
|
||||
page.table_host.set_rows([row])
|
||||
page.table.selectRow(0)
|
||||
|
||||
Reference in New Issue
Block a user