325 lines
15 KiB
Python
325 lines
15 KiB
Python
"""Reachability and data preservation in the real reception shell."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import socket
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QEvent, QPoint, QRect, Qt, QTimer
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QFrame, QLabel, QPushButton
|
|
|
|
from doctor_workstation.ui import shell as shell_module
|
|
from doctor_workstation.ui.pages import reception as reception_module
|
|
from doctor_workstation.ui.pages.reception import QueueRow, ReceptionPage
|
|
from doctor_workstation.ui.shell import NavigationItem, ShellWindow
|
|
from doctor_workstation.ui.theme import apply_theme
|
|
|
|
|
|
def _settle(application: QApplication) -> None:
|
|
for _ in range(12):
|
|
application.processEvents()
|
|
QTest.qWait(5)
|
|
|
|
|
|
def _mapped_rect(widget, ancestor) -> QRect:
|
|
return QRect(widget.mapTo(ancestor, QPoint()), widget.size())
|
|
|
|
|
|
class _LocalRepository:
|
|
def __init__(self) -> None:
|
|
self.calls: list[str] = []
|
|
self.detail = {
|
|
"appointment": {
|
|
"id": 51, "patient_id": 151, "diagnosis_id": 251,
|
|
"patient_name": "本地测试患者", "status": 1,
|
|
"doctor_name": "测试医生", "assistant_name": "测试医助",
|
|
"appointment_time": "09:30", "appointment_type": "video",
|
|
},
|
|
"patient": {
|
|
"id": 151, "name": "本地测试患者", "gender": 2,
|
|
"age": 46, "height": 165, "weight": 60, "phone": "13800138000",
|
|
},
|
|
"diagnosis": {
|
|
"id": 251, "patient_id": 151, "chief_complaint": "反复口渴",
|
|
"present_illness": "近期口干，复诊时核对记录。",
|
|
"current_medications": "二甲双胍早晚服用，剂量待核对。",
|
|
"disease_course_text": "患病十年",
|
|
"clinical_diagnosis": "2 型糖尿病", "hba1c": 7.2,
|
|
},
|
|
}
|
|
|
|
def list_appointments(self, **_query: Any) -> dict[str, Any]:
|
|
self.calls.append("queue")
|
|
return {"lists": [deepcopy(self.detail["appointment"])], "count": 1}
|
|
|
|
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
|
assert appointment_id == 51
|
|
self.calls.append("detail")
|
|
return deepcopy(self.detail)
|
|
|
|
def get_diagnosis_ai_analysis(self, diagnosis_id: int, model: str) -> dict[str, Any]:
|
|
assert diagnosis_id == 251
|
|
self.calls.append(model)
|
|
return {
|
|
"diagnosis_advice": "结合当前病史复核血糖控制情况。",
|
|
"risk_assessment": [{"label": "血糖波动", "level": "medium"}],
|
|
"treatment_advice": "核对现有用药及近期检查。", "model_key": model,
|
|
}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
app = QApplication.instance() or QApplication([])
|
|
apply_theme(app)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def shell_factory(application: QApplication, monkeypatch: pytest.MonkeyPatch):
|
|
def reject_network(*_args, **_kwargs):
|
|
pytest.fail("Reception workspace tests must remain local")
|
|
|
|
def immediate(function, *args, on_success=None, on_error=None, on_finished=None,
|
|
pool=None, priority=0, **kwargs):
|
|
del pool, priority
|
|
try:
|
|
result = function(*args, **kwargs)
|
|
except Exception as error:
|
|
if on_error:
|
|
on_error(error)
|
|
raise
|
|
else:
|
|
if on_success:
|
|
on_success(result)
|
|
finally:
|
|
if on_finished:
|
|
on_finished()
|
|
return object()
|
|
|
|
monkeypatch.setattr(socket.socket, "connect", reject_network)
|
|
monkeypatch.setattr(socket.socket, "connect_ex", reject_network)
|
|
monkeypatch.setattr(socket, "create_connection", reject_network)
|
|
monkeypatch.setattr(reception_module, "run_async", immediate)
|
|
monkeypatch.setattr(shell_module.motion, "reduced_motion", lambda: True)
|
|
item = NavigationItem("reception", "接诊台", "", ReceptionPage,
|
|
("doctor.appointment/lists",))
|
|
monkeypatch.setattr(shell_module, "_resolve_navigation",
|
|
lambda *_args, **_kwargs: [(item, item.title)])
|
|
opened = []
|
|
|
|
def create(width=1920, height=1080):
|
|
repository = _LocalRepository()
|
|
shell = ShellWindow(repository, {"user": {"id": 7, "name": "测试医生"}},
|
|
permissions={
|
|
"doctor.appointment/lists", "doctor.appointment/complete",
|
|
"doctor.appointment/addDoctorNote", "tcm.diagnosis/edit",
|
|
"tcm.diagnosis/aiAnalysis", "tcm.diagnosis/aiAssistant",
|
|
})
|
|
opened.append(shell)
|
|
shell.resize(width, height)
|
|
shell.show()
|
|
_settle(application)
|
|
page = shell.pages["reception"]
|
|
page.poll_timer.stop()
|
|
shell.chat_notifications.stop()
|
|
assert page._selected_appointment_id == 51
|
|
assert page.detail_stack.currentIndex() == 1
|
|
return shell, page, repository
|
|
|
|
yield create
|
|
for shell in opened:
|
|
for timer in shell.findChildren(QTimer):
|
|
timer.stop()
|
|
shell.close()
|
|
shell.deleteLater()
|
|
application.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
application.processEvents()
|
|
|
|
|
|
@pytest.mark.parametrize("size", [(1366, 768), (1920, 1080)])
|
|
def test_real_shell_prioritizes_patient_facts_while_long_history_scrolls(
|
|
application, shell_factory, size,
|
|
):
|
|
shell, page, repository = shell_factory(*size)
|
|
assert (shell.width(), shell.height()) == size
|
|
assert not page.workbench.mode_tabs.isVisible()
|
|
hero = page.findChild(QFrame, "ReceptionHero")
|
|
hero_before = _mapped_rect(hero, page)
|
|
actions = (page.notify_button, page.history_button, page.video_button,
|
|
page.more_button, page.complete_button)
|
|
actions_before = [_mapped_rect(button, page) for button in actions]
|
|
for button in actions:
|
|
assert button.isVisibleTo(page)
|
|
assert hero.rect().contains(_mapped_rect(button, hero))
|
|
assert button.width() >= button.minimumSizeHint().width()
|
|
|
|
insights_viewport = page.insights_scroll.viewport()
|
|
summary = page.patient_summary_card
|
|
vitals = page.findChild(QFrame, "ReceptionVitalsCard")
|
|
labs = page.findChild(QFrame, "ReceptionLabsCard")
|
|
assert insights_viewport.rect().contains(_mapped_rect(summary, insights_viewport))
|
|
assert insights_viewport.rect().contains(_mapped_rect(vitals, insights_viewport))
|
|
assert _mapped_rect(summary, insights_viewport).top() == 0
|
|
assert summary.mapTo(page, QPoint()).y() < vitals.mapTo(page, QPoint()).y()
|
|
assert vitals.mapTo(page, QPoint()).y() < labs.mapTo(page, QPoint()).y()
|
|
assert labs.mapTo(page, QPoint()).y() < page.ai_analysis_card.mapTo(page, QPoint()).y()
|
|
assert {key: label.text() for key, label in page.patient_summary_labels.items()} == {
|
|
"gender_age": "女 · 46岁", "phone": "138****8000",
|
|
"course": "患病十年", "diagnosis": "2 型糖尿病",
|
|
}
|
|
assistant_title = page.ai_assistant_card.findChild(QPushButton, "ReceptionAiTitle")
|
|
insight_scroll_value = page.insights_scroll.verticalScrollBar().value()
|
|
fact_positions = [_mapped_rect(card, page) for card in (summary, vitals, labs)]
|
|
requests = list(repository.calls)
|
|
history = "完整病史：口干持续，携带记录复诊，核对既往用药。\n" * 40
|
|
page.case_labels["present"].setText(history)
|
|
_settle(application)
|
|
assert page.case_labels["present"].text() == history
|
|
assert page.case_labels["present"].height() >= page.case_labels["present"].heightForWidth(
|
|
page.case_labels["present"].width()
|
|
)
|
|
scrollbar = page.clinical_scroll.verticalScrollBar()
|
|
assert scrollbar.maximum() > 0
|
|
scrollbar.setValue(scrollbar.maximum())
|
|
_settle(application)
|
|
assert scrollbar.value() == scrollbar.maximum()
|
|
assert _mapped_rect(hero, page) == hero_before
|
|
assert [_mapped_rect(button, page) for button in actions] == actions_before
|
|
assert [_mapped_rect(card, page) for card in (summary, vitals, labs)] == fact_positions
|
|
assert page.insights_scroll.verticalScrollBar().value() == insight_scroll_value
|
|
for title in (page.ai_analysis_title, assistant_title):
|
|
page.insights_scroll.ensureWidgetVisible(title)
|
|
_settle(application)
|
|
assert insights_viewport.rect().contains(_mapped_rect(title, insights_viewport))
|
|
assert repository.calls == requests
|
|
|
|
|
|
def test_narrow_workspace_reuses_scrolls_and_preserves_drafts_tabs_and_patient(
|
|
application, shell_factory,
|
|
):
|
|
shell, page, repository = shell_factory()
|
|
clinical, insights = page.clinical_scroll, page.insights_scroll
|
|
page.detail_tabs.setCurrentIndex(4)
|
|
page.note_edit.setPlainText("未保存的本地备注")
|
|
page._pending_tongue_images = ["draft-tongue.png"]
|
|
page._pending_report_files = ["draft-report.pdf"]
|
|
requests = list(repository.calls)
|
|
for width in (1024, 1920, 1024, 1366):
|
|
shell.resize(width, 768)
|
|
_settle(application)
|
|
assert page.clinical_scroll is clinical
|
|
assert page.insights_scroll is insights
|
|
assert page.detail_tabs.currentIndex() == page.clinical_pages.currentIndex() == 4
|
|
assert page._selected_appointment_id == 51
|
|
assert page.note_edit.toPlainText() == "未保存的本地备注"
|
|
assert page._pending_tongue_images == ["draft-tongue.png"]
|
|
assert page._pending_report_files == ["draft-report.pdf"]
|
|
if width == 1024:
|
|
assert page.workbench.mode_tabs.isVisible()
|
|
for index, scroll in ((1, insights), (0, clinical)):
|
|
page.workbench.mode_tabs.setCurrentIndex(index)
|
|
_settle(application)
|
|
assert scroll.isVisibleTo(page)
|
|
assert page.workbench._narrow_stack.currentWidget() is scroll
|
|
else:
|
|
assert not page.workbench.mode_tabs.isVisible()
|
|
assert clinical.isVisibleTo(page) and insights.isVisibleTo(page)
|
|
assert repository.calls == requests
|
|
|
|
|
|
def test_missing_labs_expand_and_collapse_without_losing_recorded_values(
|
|
application, shell_factory,
|
|
):
|
|
_shell, page, _repository = shell_factory()
|
|
before = {key: label.text() for key, label in page.lab_labels.items()}
|
|
assert before["hba1c"] == "7.2 %"
|
|
assert not page.lab_cells["hba1c"].isHidden()
|
|
assert all(cell.isHidden() for key, cell in page.lab_cells.items() if key != "hba1c")
|
|
assert page.lab_missing_button.text() == "显示 7 项未记录"
|
|
page.insights_scroll.ensureWidgetVisible(page.lab_missing_button)
|
|
_settle(application)
|
|
QTest.mouseClick(page.lab_missing_button, Qt.MouseButton.LeftButton)
|
|
_settle(application)
|
|
assert all(not cell.isHidden() for cell in page.lab_cells.values())
|
|
assert page.lab_missing_button.text() == "收起未记录项"
|
|
page.lab_missing_button.click()
|
|
_settle(application)
|
|
assert {key: label.text() for key, label in page.lab_labels.items()} == before
|
|
assert all(cell.isHidden() for key, cell in page.lab_cells.items() if key != "hba1c")
|
|
|
|
|
|
def test_free_text_glucose_keeps_source_without_invented_units_or_status(
|
|
application, shell_factory,
|
|
):
|
|
_shell, page, repository = shell_factory()
|
|
description = "近期空腹7至9之间，餐后约12，尚未携带原始记录"
|
|
diagnosis = {**repository.detail["diagnosis"], "fasting_blood_sugar": description}
|
|
page._render_case(repository.detail["appointment"], repository.detail["patient"], diagnosis)
|
|
_settle(application)
|
|
assert page.lab_labels["fasting"].text() == description
|
|
assert page.lab_labels["fasting"].textFormat() == Qt.TextFormat.PlainText
|
|
assert page.lab_cells["fasting"].property("description") is True
|
|
assert not page.lab_cells["fasting"].isHidden()
|
|
assert page.lab_state_labels["fasting"].isHidden()
|
|
assert page.case_metric_labels["fasting_glucose"].text() == description
|
|
assert f"空腹血糖：{description}\n" in page.diagnosis_text.text()
|
|
assert f"{description} mmol/L" not in page.health_summary_label.text()
|
|
row = QueueRow({**repository.detail["appointment"], "diagnosis": diagnosis})
|
|
try:
|
|
metric = row.findChild(QLabel, "ReceptionQueueMetric")
|
|
assert metric.text() == "血糖有描述"
|
|
assert metric.toolTip() == description
|
|
assert metric.property("tone") == "description"
|
|
assert row.glucose_drop.isHidden()
|
|
finally:
|
|
row.deleteLater()
|
|
|
|
|
|
def test_compact_ai_prompts_keep_full_questions_and_callbacks(shell_factory, monkeypatch):
|
|
_shell, page, _repository = shell_factory()
|
|
sent = []
|
|
monkeypatch.setattr(page, "_open_ai_report_for_prompt", sent.append)
|
|
questions = [
|
|
"基于当前病史，下一步检查建议？", "该患者的用药调整建议？",
|
|
"糖尿病教育要点有哪些？", "并发症筛查建议？",
|
|
]
|
|
labels = ["检查建议", "用药核对", "健康教育", "并发症筛查"]
|
|
assert len(page.ai_prompt_buttons) == len(questions)
|
|
for button, label, question in zip(page.ai_prompt_buttons, labels, questions, strict=True):
|
|
assert button.text() == label
|
|
assert button.toolTip() == button.accessibleName() == question
|
|
assert button.isEnabled()
|
|
button.click()
|
|
assert sent == questions
|
|
|
|
|
|
def test_patient_facts_stay_first_while_ai_is_loading_or_unavailable(application, shell_factory):
|
|
_shell, page, repository = shell_factory()
|
|
before = {key: label.text() for key, label in page.patient_summary_labels.items()}
|
|
requests = list(repository.calls)
|
|
for state in ("loading", "error", "permission"):
|
|
page._set_ai_analysis_state(state)
|
|
_settle(application)
|
|
assert {key: label.text() for key, label in page.patient_summary_labels.items()} == before
|
|
assert page.patient_summary_card.mapTo(page.insights_scroll.widget(), QPoint()).y() == 0
|
|
assert repository.calls == requests
|
|
|
|
|
|
def test_patient_change_returns_to_facts_and_clears_previous_summary(application, shell_factory, monkeypatch):
|
|
_shell, page, _repository = shell_factory()
|
|
page.insights_scroll.verticalScrollBar().setValue(page.insights_scroll.verticalScrollBar().maximum())
|
|
monkeypatch.setattr(page, "_load_detail", lambda record, **_kwargs: page._reset_detail_content(record))
|
|
page._select_record({"id": 52, "patient_name": "另一位患者"})
|
|
_settle(application)
|
|
assert page.insights_scroll.verticalScrollBar().value() == 0
|
|
assert all(label.text() == "正在加载…" for label in page.patient_summary_labels.values())
|
|
page._render_detail_load_failure("详情暂不可用")
|
|
assert all(label.text() == "未记录" for label in page.patient_summary_labels.values())
|
|
assert all(label.property("missing") is True for label in page.patient_summary_labels.values())
|