This commit is contained in:
Your Name
2026-08-11 09:12:51 +08:00
parent c3ceb0dd0f
commit cfe4c82c90
111 changed files with 26110 additions and 826 deletions
+566
View File
@@ -0,0 +1,566 @@
from __future__ import annotations
import os
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QDate, QSize, Qt
from PySide6.QtWidgets import QApplication, QStyle, QStyleOptionButton, QWidget
from doctor_workstation.ui.appointment_drawer import APPOINTMENT_DRAWER_QSS, AppointmentDrawer
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
def _immediate_async(
function: Any,
*args: Any,
on_success: Any = None,
on_error: Any = None,
on_finished: Any = None,
**kwargs: Any,
) -> object:
try:
result = function(*args, **kwargs)
except Exception as error:
if on_error:
on_error(error)
else:
if on_success:
on_success(result)
finally:
if on_finished:
on_finished()
return object()
class _DeferredAsync:
def __init__(self) -> None:
self.calls: list[dict[str, Any]] = []
def __call__(
self,
function: Any,
*args: Any,
on_success: Any = None,
on_error: Any = None,
on_finished: Any = None,
**kwargs: Any,
) -> object:
self.calls.append(
{
"function": function,
"args": args,
"kwargs": kwargs,
"on_success": on_success,
"on_error": on_error,
"on_finished": on_finished,
}
)
return object()
def succeed(self, index: int = 0) -> Any:
call = self.calls.pop(index)
try:
result = call["function"](*call["args"], **call["kwargs"])
except Exception as error:
if call["on_error"]:
call["on_error"](error)
raise
else:
if call["on_success"]:
call["on_success"](result)
return result
finally:
if call["on_finished"]:
call["on_finished"]()
class _VisualRepository:
def __init__(
self,
*,
today_conflict: bool = True,
doctors: bool = True,
rosters: bool = True,
slot_error: bool = False,
) -> None:
self.today_conflict = today_conflict
self.doctors = doctors
self.rosters = rosters
self.slot_error = slot_error
self.roster_queries: list[dict[str, Any]] = []
self.slot_queries: list[dict[str, Any]] = []
def list_diagnosis_doctors(self) -> list[dict[str, Any]]:
if not self.doctors:
return []
return [
{"id": 77, "name": "陈医生", "department_name": "中医科"},
{"id": 88, "name": "周医生", "department_name": "内科"},
{"id": 99, "name": "林医生", "department_name": "全科"},
]
def get_dictionary(self, dictionary_type: str) -> list[dict[str, Any]]:
assert dictionary_type == "channels"
return [
{"id": 2, "name": "自媒体4H", "value": "self-4h", "status": 1, "sort": 20},
{"id": 1, "name": "线上复诊", "value": "online", "status": 1, "sort": 10},
]
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
if kwargs.get("status") == 3:
return {
"lists": [
{
"appointment_date": "2026-08-01",
"appointment_time": "10:30-11:00",
}
],
"count": 1,
}
if self.today_conflict:
return {"lists": [{"status": 1}], "count": 1}
return {"lists": [], "count": 0}
def list_appointment_rosters(self, **kwargs: Any) -> dict[str, Any]:
self.roster_queries.append(kwargs)
if not self.rosters:
return {"lists": [], "count": 0}
today = QDate.currentDate()
return {
"lists": [
{"date": today.toString("yyyy-MM-dd")},
{"date": today.addDays(1).toString("yyyy-MM-dd")},
{"date": today.addDays(2).toString("yyyy-MM-dd")},
],
"count": 3,
}
def get_available_appointment_slots(self, **kwargs: Any) -> dict[str, Any]:
self.slot_queries.append(kwargs)
if self.slot_error:
raise RuntimeError("号源服务暂时不可用")
return {
"slots": [
{"time": "09:30-10:00", "available": True, "quota": 2},
{"time": "10:00-10:30", "available": False, "quota": 0},
{"time": "14:30-15:00", "available": True, "quota": 1},
{"time": "15:00-15:30", "available": False, "quota": 0},
]
}
def _show_drawer(
application: QApplication,
size: tuple[int, int],
*,
repository: Any = None,
autoload: bool = False,
async_runner: Any = _immediate_async,
) -> tuple[QWidget, AppointmentDrawer]:
host = QWidget()
host.resize(*size)
host.show()
drawer = AppointmentDrawer(
{
"id": 501,
"diagnosis_id": 501,
"patient_id": 999,
"patient_name": "林晓岚",
"doctor_id": 77,
},
repository=repository,
parent=host,
autoload=autoload,
async_runner=async_runner,
)
drawer.show()
application.processEvents()
application.processEvents()
return host, drawer
@pytest.mark.parametrize("size,expected_width", [((1024, 640), 614), ((1440, 900), 864)])
def test_full_window_rtl_drawer_geometry_and_fixed_regions(
application: QApplication,
size: tuple[int, int],
expected_width: int,
) -> None:
application_style = application.styleSheet()
host, drawer = _show_drawer(application, size)
assert drawer.size() == host.size()
assert drawer.drawer_width == expected_width
assert drawer.panel.geometry().right() == drawer.rect().right()
assert drawer.panel.height() == drawer.height()
assert drawer.header.height() == 58
assert drawer.header.geometry().top() == 0
assert drawer.footer.geometry().bottom() == drawer.panel.rect().bottom()
assert drawer.body_scroll.geometry().top() == drawer.header.geometry().bottom() + 1
assert drawer.body_scroll.geometry().bottom() + 1 == drawer.footer.geometry().top()
assert drawer.windowFlags() & Qt.WindowType.FramelessWindowHint
assert drawer.testAttribute(Qt.WidgetAttribute.WA_TranslucentBackground)
footer_geometry = drawer.footer.geometry()
scroll_bar = drawer.body_scroll.verticalScrollBar()
scroll_bar.setValue(scroll_bar.maximum())
application.processEvents()
assert drawer.footer.geometry() == footer_geometry
assert application.styleSheet() == application_style
drawer.close()
host.close()
application.processEvents()
def test_field_order_density_and_conditional_channel_row(
application: QApplication,
) -> None:
host, drawer = _show_drawer(application, (1024, 640))
assert [label.text() for label in drawer.form_labels] == [
"上次就诊:",
"预约方式:",
"预约类型:",
"选择患者:",
"渠道来源 *:",
"自媒体补充 *:",
"预约医生:",
"预约时间:",
"备注:",
]
assert all(label.width() == 100 for label in drawer.form_labels)
assert drawer.channel_source.maximumWidth() == 360
assert drawer.channel_source_detail.maximumWidth() == 360
assert drawer.channel_source.height() == 32
assert drawer.cancel_button.height() == 32
assert drawer.ok_button.height() == 32
assert drawer.ok_button.text() == "确定"
assert drawer.remark.height() == 70
assert drawer.channel_detail_row.isHidden()
assert drawer.time_empty.illustration.size() == QSize(80, 60)
assert drawer.time_empty.illustration.accessibleName() == "空状态插图"
for radio in (
drawer.appointment_method_radio,
drawer.appointment_type_radio,
drawer.patient_radio,
):
option = QStyleOptionButton()
radio.initStyleOption(option)
indicator = radio.style().subElementRect(
QStyle.SubElement.SE_RadioButtonIndicator, option, radio
)
assert indicator.size() == QSize(14, 14)
drawer._channel_names = {"self-4h": "自媒体4H", "online": "线上复诊"}
drawer.channel_source.clear()
drawer.channel_source.addItem("请选择渠道来源", "")
drawer.channel_source.addItem("自媒体4H", "self-4h")
drawer.channel_source.addItem("线上复诊", "online")
drawer.channel_source.setCurrentIndex(drawer.channel_source.findData("self-4h"))
application.processEvents()
assert not drawer.channel_detail_row.isHidden()
assert drawer.channel_source_detail.height() == 32
drawer.channel_source_detail.setText("视频号")
drawer.channel_source.setCurrentIndex(drawer.channel_source.findData("online"))
assert drawer.channel_detail_row.isHidden()
assert drawer.channel_source_detail.text() == ""
drawer.close()
host.close()
application.processEvents()
def test_roster_slot_states_conflict_refresh_and_submit_contract(
application: QApplication,
) -> None:
repository = _VisualRepository(today_conflict=True)
host, drawer = _show_drawer(
application,
(1440, 900),
repository=repository,
autoload=True,
)
today = QDate.currentDate().toString("yyyy-MM-dd")
tomorrow = QDate.currentDate().addDays(1).toString("yyyy-MM-dd")
assert drawer.last_visit_label.text() == "2026-08-01 10:30-11:00"
assert list(drawer.doctor_buttons) == [77, 88, 99]
assert [button.text() for button in drawer.doctor_buttons.values()] == [
"陈医生",
"周医生",
"林医生",
]
assert all("·" not in button.text() for button in drawer.doctor_buttons.values())
assert all(button.size() == QSize(130, 40) for button in drawer.date_buttons.values())
assert all(button.minimumSize() == QSize(130, 40) for button in drawer.date_buttons.values())
assert all(button.maximumSize() == QSize(130, 40) for button in drawer.date_buttons.values())
assert drawer.date_combo.currentData() == tomorrow
assert drawer.conflict_alert.property("kind") == "info"
assert "可为患者预约其他日期" in drawer.conflict_alert.label.text()
available = drawer.slot_buttons["09:30-10:00"]
unavailable = drawer.slot_buttons["10:00-10:30"]
assert available.minimumWidth() >= 110
assert available.minimumHeight() >= 70
assert available.property("slotState") == "available"
assert available.isEnabled()
assert unavailable.property("slotState") == "unavailable"
assert not unavailable.isEnabled()
assert unavailable.status_label.text() == "已约"
assert unavailable.status_label.isVisible()
assert unavailable.accessibleName() == "10:00-10:30 已约"
available.click()
drawer.channel_source.setCurrentIndex(drawer.channel_source.findData("online"))
assert drawer.ok_button.isEnabled()
expected_payload = {
"diagnosis_id": 501,
"patient_id": 501,
"doctor_id": 77,
"appointment_date": tomorrow,
"appointment_time": "09:30-10:00",
"period": "all",
"appointment_type": "video",
"remark": "",
"channel_source": "online",
"channel_source_detail": "",
}
assert drawer.payload() == expected_payload
assert repository.roster_queries[-1] == {
"doctor_id": 77,
"start_date": today,
"end_date": QDate.currentDate().addDays(6).toString("yyyy-MM-dd"),
"status": 1,
"page_no": 1,
"page_size": 100,
}
assert repository.slot_queries[-1] == {
"doctor_id": 77,
"appointment_date": tomorrow,
"period": "all",
}
query_count = len(repository.slot_queries)
drawer.refresh_slots_button.click()
assert len(repository.slot_queries) == query_count + 1
assert drawer.slot_combo.currentData() == "09:30-10:00"
assert drawer.ok_button.isEnabled()
drawer.date_buttons[today].click()
assert drawer.date_combo.currentData() == today
assert drawer.conflict_alert.property("kind") == "warning"
assert "不能重复预约今天" in drawer.conflict_alert.label.text()
assert not drawer.ok_button.isEnabled()
generation = drawer._slot_generation
drawer._apply_slots(
{"slots": [{"time": "00:00-00:30", "available": True, "quota": 1}]},
77,
today,
generation,
)
application.processEvents()
expired = drawer.slot_buttons["00:00-00:30"]
assert expired.property("slotState") == "unavailable"
assert not expired.isEnabled()
assert expired.status_label.text() == "已约"
assert expired.status_label.isVisible()
assert "已过期" not in expired.accessibleName()
drawer.close()
host.close()
application.processEvents()
def test_initial_loading_state_covers_the_drawer_body(
application: QApplication,
) -> None:
deferred = _DeferredAsync()
host, drawer = _show_drawer(
application,
(1024, 640),
repository=_VisualRepository(),
autoload=True,
async_runner=deferred,
)
assert len(deferred.calls) == 1
assert drawer._active_loading == "initial"
assert drawer.loading_overlay.isVisible()
assert drawer.loading_overlay.label.text() == "正在加载医生、渠道与挂号状态…"
assert drawer.loading_overlay.geometry() == drawer.body_scroll.viewport().rect()
assert drawer.loading_overlay.spinner._timer.isActive()
assert not drawer.ok_button.isEnabled()
drawer.close()
host.close()
application.processEvents()
def test_empty_doctor_state_is_explicit_and_blocks_submit(
application: QApplication,
) -> None:
host, drawer = _show_drawer(
application,
(1024, 640),
repository=_VisualRepository(doctors=False),
autoload=True,
)
assert not drawer.doctor_buttons
assert drawer.doctor_empty.text() == "暂无可预约医生"
assert drawer.doctor_empty.isVisible()
assert drawer.time_stack.currentWidget() is drawer.time_empty
assert drawer.time_empty.label.text() == "暂无可预约医生"
assert drawer.time_empty.illustration.size() == QSize(80, 60)
assert drawer.banner.property("kind") == "warning"
assert not drawer._initial_loaded
assert not drawer.ok_button.isEnabled()
drawer.close()
host.close()
application.processEvents()
def test_empty_roster_state_is_explicit_and_blocks_submit(
application: QApplication,
) -> None:
repository = _VisualRepository(rosters=False)
host, drawer = _show_drawer(
application,
(1024, 640),
repository=repository,
autoload=True,
)
assert list(drawer.doctor_buttons) == [77, 88, 99]
assert repository.roster_queries
assert not drawer.date_buttons
assert drawer.time_stack.currentWidget() is drawer.time_empty
assert drawer.time_empty.label.text() == "该医生暂无排班"
assert drawer.banner.property("kind") == "warning"
assert "未来 7 天暂无可预约排班" in drawer.banner.label.text()
assert not drawer.ok_button.isEnabled()
drawer.close()
host.close()
application.processEvents()
def test_refreshing_state_and_generation_ignore_stale_slot_results(
application: QApplication,
) -> None:
repository = _VisualRepository(today_conflict=False)
host, drawer = _show_drawer(
application,
(1440, 900),
repository=repository,
autoload=True,
)
tomorrow = QDate.currentDate().addDays(1).toString("yyyy-MM-dd")
drawer.date_buttons[tomorrow].click()
selected = "09:30-10:00"
drawer.slot_buttons[selected].click()
drawer.channel_source.setCurrentIndex(drawer.channel_source.findData("online"))
assert drawer.ok_button.isEnabled()
deferred = _DeferredAsync()
drawer._run_async = deferred
drawer.refresh_slots_button.click()
application.processEvents()
assert len(deferred.calls) == 1
assert drawer._active_loading == "slots"
assert drawer.loading_overlay.isVisible()
assert drawer.loading_overlay.label.text() == "正在刷新可用号源…"
assert drawer.refresh_slots_button.text() == "刷新中…"
assert not drawer.refresh_slots_button.isEnabled()
assert drawer.slot_state_stack.currentWidget() is drawer.slot_empty
assert drawer.slot_empty.label.text() == "正在刷新可用号源…"
assert not drawer.ok_button.isEnabled()
doctor_id = int(drawer.doctor_combo.currentData())
appointment_date = str(drawer.date_combo.currentData())
stale_generation = drawer._slot_generation
drawer._request_slots(doctor_id, appointment_date, restore_selection=selected)
current_generation = drawer._slot_generation
assert current_generation == stale_generation + 1
assert len(deferred.calls) == 2
deferred.succeed()
application.processEvents()
assert drawer._slot_generation == current_generation
assert drawer.loading_overlay.isVisible()
assert not drawer.slot_buttons
deferred.succeed()
application.processEvents()
assert drawer.loading_overlay.isHidden()
assert drawer.refresh_slots_button.text() == "刷新"
assert drawer.refresh_slots_button.isEnabled()
assert drawer.slot_combo.currentData() == selected
assert drawer.slot_buttons[selected].isChecked()
assert drawer.ok_button.isEnabled()
drawer.close()
host.close()
application.processEvents()
def test_slot_error_state_is_visible_and_recoverable(
application: QApplication,
) -> None:
host, drawer = _show_drawer(
application,
(1024, 640),
repository=_VisualRepository(today_conflict=False, slot_error=True),
autoload=True,
)
assert drawer.loading_overlay.isHidden()
assert drawer.banner.isVisible()
assert drawer.banner.property("kind") == "danger"
assert "号源加载失败" in drawer.banner.label.text()
assert drawer.slot_state_stack.currentWidget() is drawer.slot_empty
assert drawer.slot_empty.label.text() == "号源加载失败"
assert drawer.refresh_slots_button.text() == "刷新"
assert drawer.refresh_slots_button.isEnabled()
assert not drawer.slot_buttons
assert not drawer.ok_button.isEnabled()
drawer.close()
host.close()
application.processEvents()
def test_keyboard_focus_has_a_visible_state(
application: QApplication,
) -> None:
host, drawer = _show_drawer(
application,
(1440, 900),
repository=_VisualRepository(today_conflict=False),
autoload=True,
)
focus_target = list(drawer.date_buttons.values())[1]
drawer.activateWindow()
focus_target.setFocus(Qt.FocusReason.TabFocusReason)
application.processEvents()
assert focus_target.focusPolicy() != Qt.FocusPolicy.NoFocus
assert focus_target.hasFocus()
assert application.focusWidget() is focus_target
assert 'QPushButton[appointmentDate="true"]:focus' in APPOINTMENT_DRAWER_QSS
assert "border-color: #79BBFF;" in APPOINTMENT_DRAWER_QSS
drawer.close()
host.close()
application.processEvents()
+881 -3
View File
@@ -7,8 +7,15 @@ from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QDate
from PySide6.QtWidgets import QApplication, QDialog
from PySide6.QtCore import QDate, Signal
from PySide6.QtWidgets import (
QApplication,
QDialog,
QInputDialog,
QMessageBox,
QToolButton,
QWidget,
)
from doctor_workstation.core import PermissionSet
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
@@ -18,16 +25,32 @@ from doctor_workstation.ui.pages.consultations import (
_video_payload,
appointment_rows,
is_diagnosis_confirmed,
is_valid_id_card,
is_video_available,
prescription_action_label,
)
class _ListDiagnosisDialog(QWidget):
saved = Signal()
def __init__(self, _repository: Any, parent: QWidget | None = None) -> None:
super().__init__(parent)
def open_for(self, *_args: Any, **_kwargs: Any) -> None:
return None
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
@pytest.fixture(autouse=True)
def isolate_list_from_detail_dialog(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(consultations_module, "DiagnosisDialog", _ListDiagnosisDialog)
@pytest.fixture
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
def run_immediately(
@@ -390,7 +413,20 @@ def test_switching_rows_invalidates_prescription_worker_and_clears_busy(
def test_native_call_does_not_reuse_video_qr_permission(
application: QApplication,
) -> None:
page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet([]))
class VideoRepository:
def get_call_ticket(self, patient_id: int, diagnosis_id: int) -> None:
pass
def start_call(self, diagnosis_id: int, patient_id: int, *, call_type: int = 2) -> None:
pass
def bind_call_room(self, diagnosis_id: int, room_id: str) -> None:
pass
def end_call(self, diagnosis_id: int) -> None:
pass
page = ConsultationsPage(VideoRepository(), permissions=PermissionSet([]))
emitted: list[dict[str, Any]] = []
page.video_requested.connect(emitted.append)
page.table.set_rows([_row()])
@@ -399,7 +435,849 @@ def test_native_call_does_not_reuse_video_qr_permission(
assert not page.video_button.isHidden()
assert page.video_button.isEnabled()
video_cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 10))
video_action = next(button for button in video_cell.findChildren(QToolButton))
assert video_action.text() == "进入视频问诊"
assert "摄像头和麦克风" in video_action.toolTip()
page._request_video()
assert emitted == [_video_payload(_row())]
page.close()
application.processEvents()
@pytest.mark.parametrize(
"value",
[
"11010519491231002X",
"110105491231002",
"44052420000101001x",
],
)
def test_id_card_preflight_accepts_admin_15_and_18_digit_shapes(value: str) -> None:
assert is_valid_id_card(value)
@pytest.mark.parametrize(
"value",
[
"",
"11010519491331002X",
"11010519491232002X",
"01010519491231002X",
"110105491331002",
"not-an-id",
],
)
def test_id_card_preflight_rejects_invalid_shapes(value: str) -> None:
assert not is_valid_id_card(value)
def test_invalid_id_card_never_reaches_mutation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["tcm.diagnosis/edit"]))
page.table.set_rows([_row()])
page.table.selectRow(0)
mutations: list[tuple[Any, str]] = []
messages: list[str] = []
monkeypatch.setattr(QInputDialog, "getText", lambda *_args, **_kwargs: ("123456", True))
monkeypatch.setattr(
page, "_run_mutation", lambda operation, message: mutations.append((operation, message))
)
monkeypatch.setattr(
consultations_module,
"show_toast",
lambda _owner, message, *_args, **_kwargs: messages.append(message),
)
page._fill_selected_id_card()
assert mutations == []
assert messages == ["请输入15或18位有效身份证号。"]
page.close()
application.processEvents()
def test_valid_id_card_reaches_the_existing_repository_dto(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[tuple[int, str]] = []
class Repository:
def fill_diagnosis_id_card(self, diagnosis_id: int, id_card: str) -> None:
calls.append((diagnosis_id, id_card))
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/edit"]),
)
page.table.set_rows([_row()])
page.table.selectRow(0)
monkeypatch.setattr(
QInputDialog,
"getText",
lambda *_args, **_kwargs: ("11010519491231002X", True),
)
monkeypatch.setattr(page, "_run_mutation", lambda operation, _message: operation())
page._fill_selected_id_card()
assert calls == [(501, "11010519491231002X")]
page.close()
application.processEvents()
def test_multi_appointment_cancel_is_revalidated_before_repository_mutation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
record = _row(
appointments=[
{"id": 101, "status": 1, "doctor_name": "陈医生"},
{"id": 102, "status": 4, "doctor_name": "李医生"},
]
)
page = ConsultationsPage(
SimpleNamespace(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
page.table.selectRow(0)
mutations: list[Any] = []
questions: list[Any] = []
messages: list[str] = []
monkeypatch.setattr(page, "_run_mutation", lambda *args: mutations.append(args))
monkeypatch.setattr(
QMessageBox,
"question",
lambda *args, **kwargs: questions.append((args, kwargs)),
)
monkeypatch.setattr(
consultations_module,
"show_toast",
lambda _owner, message, *_args, **_kwargs: messages.append(message),
)
page._cancel_selected_appointment()
assert mutations == []
assert questions == []
assert messages == ["该诊单有多条挂号,请在具体挂号记录中取消。"]
page.close()
application.processEvents()
def test_single_appointment_cancel_uses_the_visible_nested_appointment_id(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
cancelled: list[int] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int, **_kwargs: Any) -> None:
cancelled.append(appointment_id)
record = _row(
appointment_id=999,
appointments=[{"id": 101, "status": 1, "doctor_name": "陈医生"}],
)
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
page.table.selectRow(0)
monkeypatch.setattr(
QMessageBox,
"question",
lambda *_args, **_kwargs: QMessageBox.StandardButton.Yes,
)
monkeypatch.setattr(
page,
"_run_mutation",
lambda operation, _message: operation(),
)
page._cancel_selected_appointment()
assert cancelled == [101]
page.close()
application.processEvents()
def test_multi_appointment_card_cancel_mutates_only_the_signalled_exact_id(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
cancelled: list[int] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
cancelled.append(appointment_id)
record = _row(
appointment_id=999,
appointments=[
{"id": 101, "status": 1, "doctor_name": "陈医生", "time_text": "09:00"},
{"id": 102, "status": 4, "doctor_name": "李医生", "time_text": "10:00"},
],
)
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
monkeypatch.setattr(
QMessageBox,
"question",
lambda *_args, **_kwargs: QMessageBox.StandardButton.Yes,
)
monkeypatch.setattr(page, "_run_mutation", lambda operation, _message: operation())
page._cancel_appointment_item(record, 102)
assert cancelled == [102]
page.close()
application.processEvents()
@pytest.mark.parametrize("appointment_id", [0, 999, 103])
def test_exact_cancel_rejects_unknown_or_non_cancellable_appointment_before_confirmation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
appointment_id: int,
) -> None:
cancelled: list[int] = []
questions: list[Any] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
cancelled.append(appointment_id)
record = _row(
appointments=[
{"id": 101, "status": 1, "doctor_name": "陈医生"},
{"id": 103, "status": 3, "doctor_name": "李医生"},
]
)
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
monkeypatch.setattr(
QMessageBox,
"question",
lambda *args, **kwargs: questions.append((args, kwargs)),
)
page._cancel_appointment_item(record, appointment_id)
assert questions == []
assert cancelled == []
page.close()
application.processEvents()
def test_exact_cancel_revalidates_status_after_confirmation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
cancelled: list[int] = []
messages: list[str] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
cancelled.append(appointment_id)
record = _row(appointments=[{"id": 101, "status": 1, "doctor_name": "陈医生"}])
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
def finish_while_confirming(*_args: Any, **_kwargs: Any) -> Any:
record["appointments"][0]["status"] = 3
return QMessageBox.StandardButton.Yes
monkeypatch.setattr(QMessageBox, "question", finish_while_confirming)
monkeypatch.setattr(
consultations_module,
"show_toast",
lambda _owner, message, *_args, **_kwargs: messages.append(message),
)
page._cancel_appointment_item(record, 101)
assert cancelled == []
assert messages == ["挂号状态已刷新,本次未执行取消。"]
page.close()
application.processEvents()
def test_exact_cancel_confirmation_cancel_never_starts_mutation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[Any] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
calls.append(appointment_id)
record = _row(appointments=[{"id": 101, "status": 1, "doctor_name": "陈医生"}])
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table.set_rows([record])
monkeypatch.setattr(
QMessageBox,
"question",
lambda *_args, **_kwargs: QMessageBox.StandardButton.Cancel,
)
monkeypatch.setattr(page, "_run_mutation", lambda *args: calls.append(args))
page._cancel_appointment_item(record, 101)
assert calls == []
page.close()
application.processEvents()
def test_exact_cancel_direct_handler_rechecks_permission_before_confirmation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[Any] = []
class Repository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
calls.append(appointment_id)
record = _row(appointments=[{"id": 101, "status": 1}])
page = ConsultationsPage(Repository(), permissions=PermissionSet([]))
page.table.set_rows([record])
monkeypatch.setattr(
QMessageBox,
"question",
lambda *args, **kwargs: calls.append((args, kwargs)),
)
page._cancel_appointment_item(record, 101)
assert calls == []
page.close()
application.processEvents()
def test_menu_handlers_call_only_permission_gated_real_repository_methods(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[tuple[str, Any]] = []
class Repository:
def generate_video_qrcode(
self, doctor_id: int, patient_id: int, share_user_id: int
) -> dict[str, str]:
calls.append(
(
"video_qr",
{
"doctor_id": doctor_id,
"patient_id": patient_id,
"share_user_id": share_user_id,
},
)
)
return {"qrcode_url": "https://example.invalid/video.png"}
def generate_diagnosis_qrcode(
self,
diagnosis_id: int,
doctor_id: int,
patient_id: int,
share_user_id: int,
) -> dict[str, str]:
calls.append(
(
"confirm_qr",
{
"diagnosis_id": diagnosis_id,
"doctor_id": doctor_id,
"patient_id": patient_id,
"share_user_id": share_user_id,
},
)
)
return {"qrcode_url": "https://example.invalid/confirm.png"}
def list_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
calls.append(("logs", diagnosis_id))
return [{"id": 1, "action_desc": "挂号"}]
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
calls.append(
(
"order",
{
"patient_id": patient_id,
"order_type": order_type,
"amount": amount,
"remark": remark,
},
)
)
return {"order_no": "O-1"}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
calls.append(("order_qr", order_no))
return {"qrcode_url": "https://example.invalid/payment.png"}
record = _row(
appointment_doctor_id=77,
appointments=[
{
"id": 101,
"status": 1,
"doctor_id": 77,
"doctor_name": "陈医生",
}
],
)
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(
[
"tcm.diagnosis/videoQr",
"tcm.diagnosis/guahao",
"tcm.diagnosis/guahaoLogList",
"tcm.diagnosis/order",
]
),
current_user={"id": 66},
)
page.table.set_rows([record])
page.table.selectRow(0)
shown_qr: list[tuple[str, Any]] = []
shown_logs: list[Any] = []
monkeypatch.setattr(
page,
"_show_qr_result",
lambda title, _record, result: shown_qr.append((title, result)),
)
monkeypatch.setattr(
page,
"_show_appointment_logs",
lambda _record, result: shown_logs.append(result),
)
class AcceptedOrderDialog:
def __init__(self, _record: Any, _parent: Any) -> None:
pass
def exec(self) -> Any:
return QDialog.DialogCode.Accepted
def payload(self) -> dict[str, Any]:
return {"patient_id": 301, "order_type": 2, "amount": 88.5, "remark": "复诊"}
monkeypatch.setattr(consultations_module, "_DiagnosisOrderDialog", AcceptedOrderDialog)
monkeypatch.setattr(consultations_module._QrImagePreview, "load_url", lambda *_args: None)
monkeypatch.setattr(page, "refresh", lambda *_args, **_kwargs: None)
page._request_video_qr()
page._request_confirm_qr()
page._request_appointment_logs()
page._create_diagnosis_order()
qr_payload = {
"patient_id": 301,
"doctor_id": 77,
"share_user_id": 66,
}
assert calls == [
("video_qr", qr_payload),
("confirm_qr", {**qr_payload, "diagnosis_id": 501}),
("logs", 501),
(
"order",
{"patient_id": 301, "order_type": 2, "amount": 88.5, "remark": "复诊"},
),
("order_qr", "O-1"),
]
assert [title for title, _result in shown_qr] == ["视频二维码", "诊单二维码"]
assert shown_logs == [[{"id": 1, "action_desc": "挂号"}]]
assert page._order_qr_dialog is not None
assert page._order_qr_dialog.order_no == "O-1"
assert page._order_qr_dialog.qrcode_url == "https://example.invalid/payment.png"
page._order_qr_dialog.reject()
page.close()
application.processEvents()
def test_menu_handler_direct_calls_fail_closed_without_permission_or_active_state(
application: QApplication,
immediate_async: None,
) -> None:
calls: list[str] = []
class Repository:
def generate_video_qrcode(self, payload: dict[str, Any]) -> dict[str, str]:
calls.append(str(payload))
return {"qrcode_url": "https://example.invalid/video.png"}
denied = ConsultationsPage(Repository(), permissions=PermissionSet([]))
denied.table.set_rows([_row(appointment_doctor_id=77)])
denied.table.selectRow(0)
denied._request_video_qr()
assert calls == []
denied.close()
inactive = ConsultationsPage(Repository(), permissions=PermissionSet(["tcm.diagnosis/videoQr"]))
inactive.table.set_rows(
[_row(appointment_status=4, appointments=[{"id": 101, "status": 4, "doctor_id": 77}])]
)
inactive.table.selectRow(0)
inactive._request_video_qr()
assert calls == []
inactive.close()
application.processEvents()
def test_menu_request_generation_ignores_result_after_row_selection_changes(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
class Repository:
def generate_video_qrcode(
self, doctor_id: int, patient_id: int, share_user_id: int
) -> dict[str, str]:
del doctor_id, patient_id, share_user_id
return {"qrcode_url": "https://example.invalid/video.png"}
callbacks: list[Any] = []
def hold_request(
_operation: Any,
*,
on_success: Any = None,
on_error: Any = None,
**_kwargs: Any,
) -> object:
callbacks.append((on_success, on_error))
return object()
monkeypatch.setattr(consultations_module, "run_async", hold_request)
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/videoQr"]),
current_user={"id": 66},
)
rows = [
_row(appointment_doctor_id=77),
_row(
id=502,
diagnosis_id=502,
patient_id=302,
appointment_id=102,
appointment_doctor_id=78,
appointments=[{"id": 102, "status": 1, "doctor_id": 78}],
),
]
page.table.set_rows(rows)
page.table.selectRow(0)
shown: list[Any] = []
monkeypatch.setattr(page, "_show_qr_result", lambda *args: shown.append(args))
page._request_video_qr()
assert len(callbacks) == 1
page.table.selectRow(1)
callbacks[0][0]({"qrcode_url": "https://example.invalid/video.png"})
assert shown == []
page.close()
application.processEvents()
def test_diagnosis_order_qr_failure_retries_without_creating_a_second_order(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls: list[tuple[str, Any]] = []
class Repository:
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
calls.append(("create", (patient_id, order_type, amount, remark)))
return {"order_no": "PAY-RETRY-1"}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
calls.append(("qr", order_no))
if sum(kind == "qr" for kind, _payload in calls) == 1:
raise RuntimeError("二维码服务暂不可用")
return {"qrcode_url": "https://example.invalid/payment-retry.png"}
class AcceptedOrderDialog:
def __init__(self, _record: Any, _parent: Any) -> None:
pass
def exec(self) -> Any:
return QDialog.DialogCode.Accepted
def payload(self) -> dict[str, Any]:
return {"patient_id": 301, "order_type": 2, "amount": 88.5, "remark": "复诊"}
page = ConsultationsPage(
Repository(),
permissions=PermissionSet(["tcm.diagnosis/order"]),
)
page.table.set_rows([_row()])
page.table.selectRow(0)
monkeypatch.setattr(consultations_module, "_DiagnosisOrderDialog", AcceptedOrderDialog)
monkeypatch.setattr(consultations_module._QrImagePreview, "load_url", lambda *_args: None)
monkeypatch.setattr(page, "refresh", lambda *_args, **_kwargs: None)
page._create_diagnosis_order()
dialog = page._order_qr_dialog
assert dialog is not None
assert dialog.order_no == "PAY-RETRY-1"
assert dialog.retry_button.isEnabled()
assert "生成失败" in dialog.status_label.text()
assert calls == [
("create", (301, 2, 88.5, "复诊")),
("qr", "PAY-RETRY-1"),
]
dialog.retry_button.click()
assert calls == [
("create", (301, 2, 88.5, "复诊")),
("qr", "PAY-RETRY-1"),
("qr", "PAY-RETRY-1"),
]
assert dialog.qrcode_url == "https://example.invalid/payment-retry.png"
assert dialog.url_edit.text() == dialog.qrcode_url
dialog.reject()
assert not page._mutation_pending
page.close()
application.processEvents()
def test_diagnosis_order_without_order_no_never_requests_payment_qr(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
qr_calls: list[str] = []
class Repository:
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
del patient_id, order_type, amount, remark
return {"id": 99}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
qr_calls.append(order_no)
return {"qrcode_url": "https://example.invalid/should-not-open.png"}
class AcceptedOrderDialog:
def __init__(self, _record: Any, _parent: Any) -> None:
pass
def exec(self) -> Any:
return QDialog.DialogCode.Accepted
def payload(self) -> dict[str, Any]:
return {"patient_id": 301, "order_type": 2, "amount": 10.0, "remark": ""}
page = ConsultationsPage(Repository(), permissions=PermissionSet(["tcm.diagnosis/order"]))
page.table.set_rows([_row()])
page.table.selectRow(0)
monkeypatch.setattr(consultations_module, "_DiagnosisOrderDialog", AcceptedOrderDialog)
monkeypatch.setattr(page, "refresh", lambda *_args, **_kwargs: None)
page._create_diagnosis_order()
assert qr_calls == []
assert page._order_qr_dialog is None
assert not page._mutation_pending
page.close()
application.processEvents()
@pytest.mark.parametrize("invalidate", ["selection", "permission"])
def test_diagnosis_order_create_result_is_rejected_when_active_context_changes(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
invalidate: str,
) -> None:
qr_calls: list[str] = []
class Repository:
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
del patient_id, order_type, amount, remark
return {"order_no": "STALE-1"}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
qr_calls.append(order_no)
return {"qrcode_url": "https://example.invalid/stale.png"}
callbacks: list[tuple[Any, Any, Any]] = []
def hold_request(
operation: Any,
*,
on_success: Any = None,
on_error: Any = None,
on_finished: Any = None,
**_kwargs: Any,
) -> object:
callbacks.append((operation, on_success, on_finished))
return object()
class AcceptedOrderDialog:
def __init__(self, _record: Any, _parent: Any) -> None:
pass
def exec(self) -> Any:
return QDialog.DialogCode.Accepted
def payload(self) -> dict[str, Any]:
return {"patient_id": 301, "order_type": 2, "amount": 10.0, "remark": ""}
page = ConsultationsPage(Repository(), permissions=PermissionSet(["tcm.diagnosis/order"]))
rows = [
_row(),
_row(id=502, diagnosis_id=502, patient_id=302, appointment_id=102),
]
page.table.set_rows(rows)
page.table.selectRow(0)
monkeypatch.setattr(consultations_module, "_DiagnosisOrderDialog", AcceptedOrderDialog)
monkeypatch.setattr(consultations_module, "run_async", hold_request)
monkeypatch.setattr(page, "refresh", lambda *_args, **_kwargs: None)
page._create_diagnosis_order()
assert len(callbacks) == 1
operation, on_success, _on_finished = callbacks[0]
result = operation()
if invalidate == "selection":
page.table.selectRow(1)
else:
page.permissions = PermissionSet([])
on_success(result)
assert qr_calls == []
assert page._order_qr_dialog is None
assert not page._mutation_pending
page.close()
application.processEvents()
def test_closing_payment_dialog_invalidates_inflight_qr_result(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
class Repository:
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
del patient_id, order_type, amount, remark
return {"order_no": "CLOSE-1"}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
return {"qrcode_url": f"https://example.invalid/{order_no}.png"}
staged: list[tuple[Any, Any, Any]] = []
call_count = 0
def stage_request(
operation: Any,
*,
on_success: Any = None,
on_error: Any = None,
on_finished: Any = None,
**_kwargs: Any,
) -> object:
nonlocal call_count
call_count += 1
if call_count == 1:
on_success(operation())
if on_finished:
on_finished()
else:
staged.append((operation, on_success, on_finished))
return object()
class AcceptedOrderDialog:
def __init__(self, _record: Any, _parent: Any) -> None:
pass
def exec(self) -> Any:
return QDialog.DialogCode.Accepted
def payload(self) -> dict[str, Any]:
return {"patient_id": 301, "order_type": 2, "amount": 10.0, "remark": ""}
page = ConsultationsPage(Repository(), permissions=PermissionSet(["tcm.diagnosis/order"]))
page.table.set_rows([_row()])
page.table.selectRow(0)
monkeypatch.setattr(consultations_module, "_DiagnosisOrderDialog", AcceptedOrderDialog)
monkeypatch.setattr(consultations_module, "run_async", stage_request)
monkeypatch.setattr(page, "refresh", lambda *_args, **_kwargs: None)
page._create_diagnosis_order()
dialog = page._order_qr_dialog
assert dialog is not None
assert len(staged) == 1
dialog.reject()
assert page._order_qr_dialog is None
assert not page._mutation_pending
operation, on_success, on_finished = staged[0]
on_success(operation())
if on_finished:
on_finished()
assert page._order_qr_dialog is None
assert not page._mutation_pending
page.close()
application.processEvents()
+187
View File
@@ -0,0 +1,187 @@
"""Exact endpoint and fail-closed tests for diagnosis-detail mutations."""
from __future__ import annotations
import time
from datetime import date
from pathlib import Path
from typing import Any
import pytest
from doctor_workstation.services.mock_repository import DemoDoctorRepository
from doctor_workstation.services.repository import RemoteDoctorRepository
class RecordingClient:
"""Minimal no-network client that preserves exact endpoint DTOs."""
def __init__(self) -> None:
self.get_calls: list[tuple[str, dict[str, Any]]] = []
self.post_calls: list[tuple[str, dict[str, Any]]] = []
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
body = dict(params or {})
self.get_calls.append((endpoint, body))
if endpoint == "tcm.diagnosis/getImChatMessages":
return {"lists": [], "patient_im_id": "patient_301"}
return {"lists": [], "count": 0}
def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any:
body = dict(payload or {})
self.post_calls.append((endpoint, body))
if endpoint == "tcm.diagnosis/createManualCallRecord":
return {"id": 88}
if endpoint == "tcm.diagnosis/generateMiniProgramQrcode":
return {"qrcode_url": "https://example.invalid/mini.png"}
if endpoint == "tcm.diagnosis/generateOrderQrcode":
return {"qrcode_url": "https://example.invalid/order.png"}
if endpoint == "order.order/create":
return {"id": 99, "order_no": "ORDER99"}
return {"ok": True}
def test_remote_detail_actions_use_exact_confirmed_endpoints() -> None:
client = RecordingClient()
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
today = date.today().isoformat()
repository.set_revisit_slot_start_offset(501, 4)
repository.add_blood_record(
diagnosis_id=501,
patient_id=301,
record_date=today,
record_time="09:30",
fasting_blood_sugar=6.2,
)
repository.add_diet_record(
diagnosis_id=501,
patient_id=301,
record_date=today,
breakfast_foods="燕麦",
breakfast_images=[],
lunch_images=[],
dinner_images=[],
)
repository.add_exercise_record(
diagnosis_id=501,
patient_id=301,
record_date=today,
exercise_type="步行",
duration=30,
intensity=2,
images=[],
)
repository.add_tracking_note(501, "继续观察")
repository.add_diagnosis_todo(501, "复测餐后血糖", int(time.time()) + 120)
repository.cancel_diagnosis_todo(77)
repository.list_call_records(501)
repository.create_manual_call_record(501)
repository.attach_local_call_recording(
501, "https://media.example.invalid/replay.mp4", call_record_id=88
)
assert repository.list_im_chat_messages(501)["lists"] == []
repository.sync_im_chat_messages(501)
repository.list_appointment_logs(501)
repository.generate_video_qrcode(1001, 301, 1001)
repository.generate_diagnosis_qrcode(501, 1001, 301, 1001)
repository.create_diagnosis_order(301, 2, 88.6, remark="检查费")
repository.generate_order_qrcode("ORDER99")
repository.cancel_diagnosis_appointment(101)
assert ("tcm.diagnosis/getCallRecords", {"diagnosis_id": 501}) in client.get_calls
assert (
"tcm.diagnosis/getImChatMessages",
{"diagnosis_id": 501, "only_archived": 1},
) in client.get_calls
assert ("tcm.diagnosis/guahaoLogList", {"id": 501}) in client.get_calls
assert (
"doctor.appointment/cancel",
{"id": 101},
) in client.post_calls
assert (
"tcm.diagnosis/setRevisitSlotStartOffset",
{"id": 501, "revisit_slot_start_offset": 4},
) in client.post_calls
assert (
"order.order/create",
{"patient_id": 301, "order_type": 2, "amount": 88.6, "remark": "检查费"},
) in client.post_calls
qr_payloads = [
body
for endpoint, body in client.post_calls
if endpoint == "tcm.diagnosis/generateMiniProgramQrcode"
]
assert qr_payloads[0] == {
"diagnosis_id": 1001,
"doctor_id": 1001,
"patient_id": 301,
"share_user_id": 1001,
"mini_program_path": "pages/login/login",
}
assert qr_payloads[1] == {
"diagnosis_id": 501,
"doctor_id": 1001,
"patient_id": 301,
"share_user_id": 1001,
}
@pytest.mark.parametrize(
("call", "message"),
[
(lambda repo: repo.set_revisit_slot_start_offset(501, 21), "between 0 and 20"),
(lambda repo: repo.add_tracking_note(501, ""), "1 to 1000"),
(
lambda repo: repo.add_diagnosis_todo(501, "稍后", int(time.time()) + 5),
"30 seconds",
),
(lambda repo: repo.cancel_diagnosis_todo(0), "positive"),
(lambda repo: repo.create_diagnosis_order(301, 9, 1), "between 1 and 8"),
(lambda repo: repo.cancel_diagnosis_appointment(0), "positive"),
],
)
def test_remote_detail_validation_fails_before_transport(call: Any, message: str) -> None:
client = RecordingClient()
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
with pytest.raises(ValueError, match=message):
call(repository)
assert client.get_calls == []
assert client.post_calls == []
def test_demo_detail_mutations_round_trip(tmp_path: Path) -> None:
repository = DemoDoctorRepository(today=date(2026, 8, 10))
blood = repository.add_blood_record(
diagnosis_id=501,
patient_id=301,
record_date="2026-08-10",
fasting_blood_sugar=5.8,
)
assert blood["id"] > 0
assert (
repository.get_tracking_window(501, start_date="2026-08-10", end_date="2026-08-10")[
"blood_records"
][-1]["fasting_blood_sugar"]
== 5.8
)
todo = repository.add_diagnosis_todo(501, "今晚回访", int(time.time()) + 120)
cancelled = repository.cancel_diagnosis_todo(todo["id"])
assert cancelled["status_text"] == "已取消"
replay = tmp_path / "replay.mp4"
replay.write_bytes(b"demo-video")
uploaded = repository.upload_call_recording(replay, 501)
assert uploaded["file_url"].startswith("/demo/uploads/video/")
assert uploaded["file_url"] in repository.list_call_records(501)[0]["recording_urls_list"]
archive = repository.list_im_chat_messages(501, only_archived=True)
assert archive["only_archived"] is True
assert {row["msg_type"] for row in archive["lists"]} >= {"text", "image", "file"}
assert repository.sync_im_chat_messages(501)["queued"] is True
repository.set_revisit_slot_start_offset(501, 7)
assert repository.get_diagnosis_detail(501)["diagnosis"]["revisit_slot_start_offset"] == 7
assert repository.cancel_diagnosis_appointment(101)["status"] == 2
File diff suppressed because it is too large Load Diff
+682
View File
@@ -0,0 +1,682 @@
from __future__ import annotations
import os
from pathlib import Path
from types import SimpleNamespace
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QAbstractTableModel, QRect, Qt, Signal
from PySide6.QtGui import QColor, QImage, QPainter
from PySide6.QtWidgets import QApplication, QFrame, QToolButton, QWidget
from doctor_workstation.core import PermissionSet
from doctor_workstation.ui.diagnosis_index_widgets import (
DIAGNOSIS_INDEX_QSS,
DiagnosisItemDelegate,
DiagnosisTableModel,
)
from doctor_workstation.ui.pages import consultations as consultations_module
from doctor_workstation.ui.pages.consultations import ConsultationsPage
class _ListDiagnosisDialog(QWidget):
saved = Signal()
def __init__(self, _repository: Any, parent: QWidget | None = None) -> None:
super().__init__(parent)
def open_for(self, *_args: Any, **_kwargs: Any) -> None:
return None
class _CancellationRepository:
def cancel_diagnosis_appointment(self, appointment_id: int) -> None:
del appointment_id
class _FullMenuRepository(_CancellationRepository):
def generate_video_qrcode(
self, doctor_id: int, patient_id: int, share_user_id: int
) -> dict[str, str]:
del doctor_id, patient_id, share_user_id
return {"qrcode_url": "https://example.invalid/video.png"}
def generate_diagnosis_qrcode(
self,
diagnosis_id: int,
doctor_id: int,
patient_id: int,
share_user_id: int,
) -> dict[str, str]:
del diagnosis_id, doctor_id, patient_id, share_user_id
return {"qrcode_url": "https://example.invalid/confirm.png"}
def list_appointment_logs(self, diagnosis_id: int) -> list[dict[str, Any]]:
del diagnosis_id
return []
def create_diagnosis_order(
self,
patient_id: int,
order_type: int,
amount: float,
*,
remark: str = "",
) -> dict[str, Any]:
del patient_id, order_type, amount, remark
return {"order_no": "ORDER-715"}
def generate_order_qrcode(self, order_no: str) -> dict[str, Any]:
del order_no
return {"qrcode_url": "https://example.invalid/order.png"}
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
@pytest.fixture(autouse=True)
def isolate_list_from_detail_dialog(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(consultations_module, "DiagnosisDialog", _ListDiagnosisDialog)
@pytest.fixture
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
def run_immediately(
function: Any,
*args: Any,
on_success: Any = None,
on_error: Any = None,
on_finished: Any = None,
**_kwargs: Any,
) -> object:
try:
result = function(*args)
except Exception as error:
if on_error:
on_error(error)
else:
if on_success:
on_success(result)
finally:
if on_finished:
on_finished()
return object()
monkeypatch.setattr(consultations_module, "run_async", run_immediately)
def _row(identifier: int, **changes: Any) -> dict[str, Any]:
row: dict[str, Any] = {
"id": identifier,
"diagnosis_id": identifier,
"patient_id": identifier + 1000,
"patient_name": f"患者{identifier}",
"gender": 1,
"age": 38,
"has_appointment": 1,
"appointment_id": identifier + 2000,
"appointment_status": 1,
"appointments": [
{
"id": identifier + 2000,
"status": 1,
"doctor_name": "陈医生",
"time_text": "今天 09:00-09:30",
}
],
"DiagnosisViewRecord": [{"is_confirmed": 1}],
"assistant_id": 8,
"assistant_name": "赵医助",
"assign_read_at": None,
"has_prescription": 1,
"followup_time_text": "2026-08-17 09:00",
"followup_doctor_name": "陈医生",
"unserved_days": 2,
"last_blood_record_at": "2026-08-09 20:10",
}
row.update(changes)
return row
def _page() -> ConsultationsPage:
return ConsultationsPage(_CancellationRepository(), permissions=PermissionSet(["*"]))
def test_visual_hierarchy_and_filter_contract(
application: QApplication,
) -> None:
page = _page()
content_layout = page.page_scroll.widget().layout()
margins = content_layout.contentsMargins()
assert (margins.left(), margins.top(), margins.right(), margins.bottom()) == (8, 16, 8, 16)
assert content_layout.spacing() == 12
assert page.findChild(QFrame, "DiagnosisFilterCard") is not None
assert page.findChild(QFrame, "DiagnosisListCard") is not None
assert page.keyword_edit.width() == 160
assert not page.advanced_filters.isVisible()
assert page.more_filter_button.text() == "更多筛选"
assert page.more_filter_button.arrowType() == Qt.ArrowType.DownArrow
assert [page._date_button_labels[key] for key in page.date_buttons] == [
"昨天挂号",
"前天挂号",
"当天挂号",
"明天挂号",
"后天挂号",
"全部",
]
assert page.date_buttons[page._appointment_date].isChecked()
assert page.pending_assign_month.width() == 128
assert page.pending_assign_keyword.width() == 220
assert page.pending_assign_keyword.placeholderText() == "搜身份证/诊单ID/患者号/备注…"
assert page.pending_assign_button.parentWidget() is page.pending_assign_wrap
assert page.pending_assign_filters.parentWidget() is page.pending_assign_wrap
assert page.batch_assign_button.text() == "批量指派医助"
assert "PageHeader" not in {type(widget).__name__ for widget in page.findChildren(QFrame)}
page.close()
application.processEvents()
def test_pending_assign_and_secondary_chip_semantics(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
page = _page()
monkeypatch.setattr(page, "refresh", lambda silent=False: None)
page._choose_pending_assign()
assert page.pending_assign_button.isChecked()
assert not page.date_buttons[page._appointment_date].isChecked()
assert not page.pending_assign_filters.isHidden()
page.pending_assign_keyword.setText("订单-88")
filters = page._filters()
assert filters == {
"pending_assign": "1",
"pending_assign_keyword": "订单-88",
}
page.pending_assign_keyword.clear()
filters = page._filters()
assert filters["pending_assign_order_month"] == page.pending_assign_month.text().strip()
assert filters["pending_assign_keyword"] == ""
page._choose_appointment_filter("0")
assert page.appointment_filter_buttons["0"].isChecked()
assert page._combo_value(page.has_appointment_combo) == "0"
page._toggle_advanced_filters(True)
assert not page.advanced_filters.isHidden()
assert page.more_filter_button.text() == "收起"
assert page.more_filter_button.arrowType() == Qt.ArrowType.UpArrow
assert page.unserved_sort_combo.isHidden()
date_ranges = page.advanced_filters.findChildren(QFrame, "DiagnosisDateRange")
assert len(date_ranges) == 2
assert all(field.width() == 260 for field in date_ranges)
page.close()
application.processEvents()
def test_dedicated_model_fixed_columns_selection_and_sort(
application: QApplication,
) -> None:
page = _page()
assert isinstance(page.table.model(), QAbstractTableModel)
assert isinstance(page.table.model(), DiagnosisTableModel)
assert page.table_host.LEFT_WIDTHS == (48, 70, 60, 100, 175, 88, 120, 100, 72, 110)
assert page.table_host.FIXED_WIDTHS == (120, 340)
assert page.table_host.fixed.width() == 462
assert page.table.isColumnHidden(10)
assert page.table_host.fixed.isColumnHidden(9)
assert not page.table_host.fixed.isColumnHidden(10)
assert page.table.verticalScrollBarPolicy() == Qt.ScrollBarPolicy.ScrollBarAlwaysOff
rows = [_row(501), _row(502, has_appointment=0, appointments=[])]
page.table_host.set_rows(rows)
model = page.table_host.model
assert model.setData(model.index(0, 0), Qt.CheckState.Checked, Qt.ItemDataRole.CheckStateRole)
assert page.table_host.selected_records() == [rows[0]]
assert page.selected_label.text() == "已选 1 条"
requested: list[str] = []
page.table_host.sort_unserved_requested.connect(requested.append)
assert page.table_host.model.headerData(9, Qt.Orientation.Horizontal) == "未服务天数"
assert page.table_host.model._sort_direction == ""
page.table_host.main.horizontalHeader().sectionClicked.emit(9)
assert requested == ["desc"]
assert page.table_host.model._sort_direction == "desc"
page.table_host.action_requested.disconnect(page._row_action)
actions: list[tuple[str, Any]] = []
page.table_host.action_requested.connect(
lambda action, record: actions.append((action, record))
)
action_cell = page.table_host.fixed.indexWidget(model.index(0, 11))
second_fixed_cell = page.table_host.fixed.indexWidget(model.index(1, 11))
second_fixed_cell.hovered_row.emit(1)
assert model.hover_row == 1
second_fixed_cell.hovered_row.emit(-1)
assert model.hover_row == -1
direct_links = {
button.text() for button in action_cell.findChildren(QToolButton) if button.menu() is None
}
assert {"查看", "诊单", "开方", "预约", "补全身份证"} <= direct_links
more = next(button for button in action_cell.findChildren(QToolButton) if button.menu())
menu_texts = [action.text() for action in more.menu().actions() if not action.isSeparator()]
assert "指派" in menu_texts
assert "取消挂号" in menu_texts
assert {"视频二维码", "二维码", "挂号日志", "创建订单"}.isdisjoint(menu_texts)
next(action for action in more.menu().actions() if action.text() == "指派").trigger()
assert actions == [("assign", rows[0])]
page.close()
application.processEvents()
def test_empty_loading_and_full_pager_keep_the_table_shell(
application: QApplication,
) -> None:
page = _page()
page.table_host.set_rows([])
page._loading = True
page.resize(1024, 640)
page.show()
application.processEvents()
page._loading = False
assert page.table_host.empty_label.isVisible()
assert page.table.horizontalHeader().isVisible()
assert page.table_host.height() >= 39 + 60
page.loading_overlay.start()
application.processEvents()
assert page.loading_overlay.isVisible()
assert page.loading_overlay.geometry() == page.table_host.rect()
page.loading_overlay.stop()
page.pager.update_state(3, 97)
assert [page.pager.size_combo.itemData(index) for index in range(4)] == [15, 20, 30, 40]
assert len([button for button in page.pager._page_buttons if not button.isHidden()]) == 5
assert page.pager.jumper.maximum() == 7
page.close()
application.processEvents()
@pytest.mark.parametrize(
("record", "stripe", "channel"),
[
(_row(701, DiagnosisViewRecord=[{"is_confirmed": 0}]), "#e6a23c", "warning"),
(_row(702, has_appointment=0, appointments=[]), "#909399", "info"),
],
)
def test_semantic_hover_preserves_gradient_and_three_pixel_stripe(
application: QApplication,
record: dict[str, Any],
stripe: str,
channel: str,
) -> None:
model = DiagnosisTableModel([record])
model.set_hover_row(0)
image = QImage(48, 52, QImage.Format.Format_ARGB32_Premultiplied)
image.fill(QColor("#FFFFFF"))
painter = QPainter(image)
DiagnosisItemDelegate._paint_row_background(
painter,
QRect(0, 0, 48, 52),
record,
0,
0,
model,
)
painter.end()
assert image.pixelColor(0, 20).name() == stripe
assert image.pixelColor(1, 20).name() == stripe
assert image.pixelColor(2, 20).name() == stripe
gradient = image.pixelColor(6, 20)
assert gradient.name() != "#f8f8f8", f"{channel} hover collapsed to a neutral row"
def test_fixed_shadow_focus_and_admin_token_contract(application: QApplication) -> None:
page = _page()
page.resize(1024, 640)
page.show()
application.processEvents()
shadow = page.table_host.fixed_shadow
assert shadow.isVisible()
assert shadow.width() == 12
assert shadow.geometry().right() == page.table_host.fixed.geometry().left() - 1
assert 'font-family: "PingFang SC", Arial, "Hiragino Sans GB", "Microsoft YaHei"' in (
DIAGNOSIS_INDEX_QSS
)
assert "QTableView:focus" in DIAGNOSIS_INDEX_QSS
assert "QToolButton[rowLink]:focus" in DIAGNOSIS_INDEX_QSS
assert '#DiagnosisIndex QToolButton[diagnosisChip="true"][semantic="warning"]' in (
DIAGNOSIS_INDEX_QSS
)
page.close()
application.processEvents()
def test_multi_appointment_never_exposes_ambiguous_row_cancel(
application: QApplication,
) -> None:
page = _page()
single = _row(710)
multiple = _row(
711,
appointments=[
{"id": 2711, "status": 1, "doctor_name": "陈医生", "time_text": "09:00"},
{
"id": 4_294_967_302,
"status": 4,
"doctor_name": "李医生",
"time_text": "10:00",
},
],
)
page.table_host.set_rows([single, multiple])
def menu_texts(row: int) -> set[str]:
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(row, 11))
more = next(button for button in cell.findChildren(QToolButton) if button.menu())
return {action.text() for action in more.menu().actions() if not action.isSeparator()}
assert "取消挂号" in menu_texts(0)
assert "取消挂号" not in menu_texts(1)
appointment_cell = page.table_host.main.indexWidget(page.table_host.model.index(1, 4))
appointment_buttons = appointment_cell.findChildren(QToolButton)
assert [button.text() for button in appointment_buttons] == ["取消", "取消"]
assert [button.accessibleName() for button in appointment_buttons] == [
"取消挂号 2711",
"取消挂号 4294967302",
]
page.table_host.appointment_cancel_requested.disconnect(page._cancel_appointment_item)
requested: list[tuple[Any, int]] = []
page.table_host.appointment_cancel_requested.connect(
lambda record, appointment_id: requested.append((record, appointment_id))
)
appointment_buttons[1].click()
assert requested == [(multiple, 4_294_967_302)]
page.close()
application.processEvents()
def test_full_more_menu_requires_each_real_repository_capability(
application: QApplication,
) -> None:
page = ConsultationsPage(_FullMenuRepository(), permissions=PermissionSet(["*"]))
record = _row(
715,
appointment_doctor_id=9,
appointments=[
{
"id": 2715,
"status": 1,
"doctor_id": 9,
"doctor_name": "陈医生",
"time_text": "09:00",
}
],
)
page.table_host.set_rows([record])
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 11))
more = next(button for button in cell.findChildren(QToolButton) if button.menu())
assert [action.text() for action in more.menu().actions() if not action.isSeparator()] == [
"指派",
"取消指派",
"视频二维码",
"二维码",
"取消挂号",
"挂号日志",
"创建订单",
"删除",
]
actions = more.menu().actions()
menu_actions = [action for action in actions if not action.isSeparator()]
assert all(not action.icon().isNull() for action in menu_actions)
assert all(action.property("iconSource") == "qpaint" for action in menu_actions)
assert actions[-2].isSeparator()
assert actions[-1].text() == "删除"
assert actions[-1].property("danger") is True
more.menu().ensurePolished()
more.menu().adjustSize()
more.menu().show()
application.processEvents()
danger_rect = more.menu().actionGeometry(actions[-1])
danger_image = more.menu().grab().toImage()
red_text_pixels = 0
for y in range(max(0, danger_rect.top()), min(danger_image.height(), danger_rect.bottom() + 1)):
for x in range(
max(0, danger_rect.left() + 40),
min(danger_image.width(), danger_rect.right() + 1),
):
color = danger_image.pixelColor(x, y)
if color.red() > 190 and color.green() < 150 and color.blue() < 150:
red_text_pixels += 1
assert red_text_pixels > 8, "删除文案必须由 danger 色绘制,不能回退成原生黑色"
more.menu().hide()
assert page.table_host.action_policy == {
"view": True,
"edit": True,
"prescription": True,
"appointment": True,
"assign": True,
"delete": True,
"video_call": False,
"appointment_cancel": True,
"video_qr": True,
"confirm_qr": True,
"appointment_logs": True,
"create_order": True,
}
page.close()
application.processEvents()
def test_patient_workspace_cancel_endpoint_is_not_a_diagnosis_list_capability(
application: QApplication,
) -> None:
class PatientWorkspaceOnlyRepository:
def cancel_patient_appointment(self, appointment_id: int) -> None:
del appointment_id
page = ConsultationsPage(
PatientWorkspaceOnlyRepository(),
permissions=PermissionSet(["tcm.diagnosis/guahao"]),
)
page.table_host.set_rows([_row(716)])
assert not page.table_host.action_policy["appointment_cancel"]
assert page.table_host.main.indexWidget(page.table_host.model.index(0, 4)) is None
action_cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 11))
more_buttons = [button for button in action_cell.findChildren(QToolButton) if button.menu()]
assert all(button.isHidden() for button in more_buttons)
assert all(
"取消挂号" not in {action.text() for action in button.menu().actions()}
for button in more_buttons
)
page.close()
application.processEvents()
@pytest.mark.parametrize("missing_method", ["create_diagnosis_order", "generate_order_qrcode"])
def test_order_menu_requires_the_complete_create_and_payment_qr_capability(
application: QApplication,
missing_method: str,
) -> None:
repository = _FullMenuRepository()
setattr(repository, missing_method, None)
page = ConsultationsPage(
repository,
permissions=PermissionSet(["tcm.diagnosis/order"]),
)
page.table_host.set_rows([_row(717, appointment_doctor_id=9)])
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 11))
more_buttons = [button for button in cell.findChildren(QToolButton) if button.menu()]
assert not page.table_host.action_policy["create_order"]
assert all(
"创建订单" not in {action.text() for action in button.menu().actions()}
for button in more_buttons
)
page.close()
application.processEvents()
@pytest.mark.parametrize(
("permission", "missing_text"),
[
("tcm.diagnosis/videoQr", "视频二维码"),
("tcm.diagnosis/guahao", "二维码"),
("tcm.diagnosis/guahaoLogList", "挂号日志"),
("tcm.diagnosis/order", "创建订单"),
],
)
def test_menu_capability_without_its_exact_permission_is_hidden(
application: QApplication,
permission: str,
missing_text: str,
) -> None:
granted = {
"tcm.diagnosis/assign",
"tcm.diagnosis/delete",
"tcm.diagnosis/videoQr",
"tcm.diagnosis/guahao",
"tcm.diagnosis/guahaoLogList",
"tcm.diagnosis/order",
}
granted.remove(permission)
page = ConsultationsPage(_FullMenuRepository(), permissions=PermissionSet(granted))
page.table_host.set_rows([_row(718, appointment_doctor_id=9)])
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 11))
more = next(button for button in cell.findChildren(QToolButton) if button.menu())
texts = {action.text() for action in more.menu().actions() if not action.isSeparator()}
assert missing_text not in texts
page.close()
application.processEvents()
def test_permission_crop_removes_unavailable_row_actions(application: QApplication) -> None:
page = ConsultationsPage(
SimpleNamespace(),
permissions=PermissionSet(["tcm.diagnosis/readonlyDetail"]),
)
page.table_host.set_rows([_row(720)])
cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 11))
visible = {button.text() for button in cell.findChildren(QToolButton) if not button.isHidden()}
assert visible == {"查看"}
assert not page.table_host.action_policy["video_qr"]
assert not page.table_host.action_policy["confirm_qr"]
assert not page.table_host.action_policy["appointment_logs"]
assert not page.table_host.action_policy["create_order"]
page.close()
application.processEvents()
def test_video_entry_is_hidden_without_the_native_repository_lifecycle(
application: QApplication,
) -> None:
page = _page()
page.table_host.set_rows([_row(725)])
video_cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 10))
assert video_cell.findChildren(QToolButton) == []
assert page.video_button.isHidden()
page.close()
application.processEvents()
def test_error_state_is_persistent_until_rows_replace_it(application: QApplication) -> None:
page = _page()
page.show()
application.processEvents()
page.table_host.set_rows([])
page.table_host.show_error("列表加载失败,请重试")
assert page.table_host.empty_label.isVisible()
assert page.table_host.empty_label.property("stateKind") == "error"
assert "加载失败" in page.table_host.empty_label.text()
page.table_host.set_rows([_row(730)])
assert not page.table_host.empty_label.isVisible()
assert page.table_host.empty_label.property("stateKind") == "empty"
page.close()
application.processEvents()
@pytest.mark.parametrize("size", [(1024, 640), (1440, 900)])
def test_two_desktop_sizes_scroll_vertically_without_horizontal_page_clipping(
application: QApplication,
size: tuple[int, int],
) -> None:
page = _page()
rows = [_row(600 + index, patient_name=f"患者{index:02d}") for index in range(15)]
page.table_host.set_rows(rows)
page.resize(*size)
page.show()
application.processEvents()
assert page.page_scroll.horizontalScrollBar().maximum() == 0
assert page.page_scroll.verticalScrollBar().maximum() > 0
assert page.table_host.fixed.geometry().right() <= page.table_host.rect().right()
assert page.search_button.geometry().right() <= page.search_button.parentWidget().rect().right()
page.close()
application.processEvents()
def test_required_reference_artifacts_exist() -> None:
root = Path(__file__).resolve().parents[1]
expected = {
root / "artifacts" / "diagnosis_visual" / "diagnosis_1024x640.png": (1024, 640),
root / "artifacts" / "diagnosis_visual" / "diagnosis_1440x900.png": (1440, 900),
root / "artifacts" / "diagnosis_visual" / "diagnosis_loading_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_empty_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_error_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_hover_warning_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_focus_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_permissions_cropped_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_horizontal_scroll_1024x640.png": (
1024,
640,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_pending_assign_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_advanced_filters_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_full_menu_1280x800.png": (
1280,
800,
),
root / "artifacts" / "diagnosis_visual" / "diagnosis_order_qrcode_1280x800.png": (
1280,
800,
),
root
/ "artifacts"
/ "diagnosis_visual"
/ "diagnosis_double_appointment_cancel_1280x800.png": (
1280,
800,
),
}
for path, dimensions in expected.items():
assert path.is_file(), f"run scripts/render_diagnosis_visual.py to create {path.name}"
image = QImage(str(path))
assert not image.isNull()
assert (image.width(), image.height()) == dimensions
@@ -0,0 +1,326 @@
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QBuffer, QByteArray, QIODevice, QObject, QSize, Signal
from PySide6.QtGui import QColor, QImage
from PySide6.QtNetwork import QNetworkReply
from PySide6.QtWidgets import QApplication, QPushButton, QWidget
from doctor_workstation.ui.diagnosis_drawer import (
ChatPanel,
DailyRecordPanel,
NotesTimeline,
_RemoteImageButton,
)
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
def _png_bytes(width: int, height: int, color: str = "#0F766E") -> bytes:
image = QImage(width, height, QImage.Format.Format_ARGB32)
image.fill(QColor(color))
buffer = QBuffer()
assert buffer.open(QIODevice.OpenModeFlag.WriteOnly)
assert image.save(buffer, "PNG")
return bytes(buffer.data())
class _FakeReply(QObject):
finished = Signal()
def __init__(
self,
payload: bytes,
error: QNetworkReply.NetworkError = QNetworkReply.NetworkError.NoError,
parent: QObject | None = None,
) -> None:
super().__init__(parent)
self.payload = payload
self.network_error = error
self.aborted = False
def abort(self) -> None:
self.aborted = True
def error(self) -> QNetworkReply.NetworkError:
return self.network_error
def readAll(self) -> QByteArray: # noqa: N802 - mirrors QNetworkReply
return QByteArray(self.payload)
class _FakeManager(QObject):
def __init__(self, parent: QObject) -> None:
super().__init__(parent)
self.responses: list[tuple[bytes, QNetworkReply.NetworkError]] = []
self.requests: list[str] = []
self.replies: list[_FakeReply] = []
def queue(
self,
payload: bytes,
error: QNetworkReply.NetworkError = QNetworkReply.NetworkError.NoError,
) -> None:
self.responses.append((payload, error))
def get(self, request: object) -> _FakeReply:
payload, error = self.responses.pop(0)
self.requests.append(request.url().toString())
reply = _FakeReply(payload, error, self)
self.replies.append(reply)
return reply
class _RenderOwner(QWidget):
def __init__(self, generation: int) -> None:
super().__init__()
self._image_generation = generation
def _hold_remote_load(self: _RemoteImageButton, source: str) -> None:
"""Offline transport used by panel tests; payload completion stays explicit."""
self._source = str(source).strip()
self._invalidate_request()
self.setToolTip(self._source)
self._show_loading()
def test_remote_image_request_is_thread_owned_and_rejects_stale_results(
application: QApplication,
) -> None:
owner = _RenderOwner(7)
button = _RemoteImageButton(
"",
render_owner=owner,
owner_generation=7,
maximum_size=QSize(64, 64),
fallback_text="舌象\n查看",
cover=True,
object_name="DiagnosisTongueThumb",
parent=owner,
)
button._manager.deleteLater()
manager = _FakeManager(button)
button._manager = manager
manager.queue(_png_bytes(180, 90, "#DC2626"))
manager.queue(_png_bytes(180, 90, "#16A34A"))
button.load_url("https://media.example.invalid/old.png")
old_reply = manager.replies[-1]
assert button.text() == "舌象\n查看"
assert button.property("loadState") == "loading"
button.load_url("https://media.example.invalid/current.png")
current_reply = manager.replies[-1]
assert old_reply.aborted is True
old_reply.finished.emit()
assert button.property("loadState") == "loading"
current_reply.finished.emit()
assert button.property("loadState") == "ready"
assert button._rendered_pixmap.size() == QSize(64, 64)
assert button.text() == ""
assert manager.parent() is button
assert manager.thread() == button.thread() == application.thread()
assert manager.requests == [
"https://media.example.invalid/old.png",
"https://media.example.invalid/current.png",
]
manager.queue(_png_bytes(90, 180, "#2563EB"))
button.load_url("https://media.example.invalid/new-owner.png")
owner._image_generation += 1
manager.replies[-1].finished.emit()
assert button.property("loadState") == "loading"
def test_remote_image_uses_text_only_after_request_or_decode_failure(
application: QApplication,
) -> None:
owner = _RenderOwner(3)
button = _RemoteImageButton(
"",
render_owner=owner,
owner_generation=3,
maximum_size=QSize(240, 200),
fallback_text="查看图片",
cover=False,
object_name="DiagnosisChatImage",
parent=owner,
)
button._manager.deleteLater()
manager = _FakeManager(button)
button._manager = manager
manager.queue(b"not-an-image")
button.load_url("https://media.example.invalid/broken.jpg")
assert button.text() == "查看图片"
assert button.property("loadState") == "loading"
manager.replies[-1].finished.emit()
assert button.property("loadState") == "failed"
assert button.text() == "查看图片"
request_count = len(manager.requests)
button.load_url("file:///C:/private/image.png")
assert len(manager.requests) == request_count
assert button.property("loadState") == "failed"
assert application.thread() == button.thread()
def test_notes_render_cover_thumbnail_and_keep_safe_open_and_single_delete(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_RemoteImageButton, "load_url", _hold_remote_load)
timeline = NotesTimeline(editable=True)
opened: list[str] = []
deleted: list[tuple[int, str, str]] = []
timeline.open_attachment_requested.connect(opened.append)
timeline.delete_attachment_requested.connect(
lambda note_id, kind, path: deleted.append((note_id, kind, path))
)
tongue_url = "https://media.example.invalid/tongue-7001.jpg"
timeline.set_notes(
[
{
"id": 7001,
"note_date": "2026-08-10",
"content": "舌淡红,苔薄白。",
"tongue_images": [tongue_url],
"report_files": [
{
"name": "近期血糖趋势.pdf",
"url": "https://media.example.invalid/report-7001.pdf",
}
],
}
]
)
thumb = timeline.findChild(_RemoteImageButton, "DiagnosisTongueThumb")
assert thumb is not None
assert thumb.text() == "舌象\n查看"
assert thumb.property("loadState") == "loading"
assert thumb._apply_payload(_png_bytes(192, 96), thumb._generation)
assert thumb.property("loadState") == "ready"
assert thumb._rendered_pixmap.size() == QSize(64, 64)
thumb.click()
assert opened == [tongue_url]
remove_buttons = timeline.findChildren(QPushButton, "DiagnosisAttachmentRemove")
assert len(remove_buttons) == 2
tongue_remove = next(button for button in remove_buttons if "舌象" in button.toolTip())
tongue_remove.click()
assert deleted == [(7001, "tongue_images", tongue_url)]
stale_generation = thumb._generation
timeline.set_notes(
[
{
"id": 7002,
"note_date": "2026-08-11",
"content": "复诊舌象。",
"tongue_images": ["https://media.example.invalid/tongue-7002.jpg"],
}
]
)
assert not thumb._apply_payload(_png_bytes(96, 192), stale_generation)
current = next(
item
for item in timeline.findChildren(_RemoteImageButton, "DiagnosisTongueThumb")
if item is not thumb
)
assert current.text() == "舌象\n查看"
assert current.property("loadState") == "loading"
assert not current._apply_payload(b"invalid", current._generation)
assert current.text() == "舌象\n查看"
timeline.close()
application.processEvents()
def test_chat_image_is_previewable_bounded_and_owner_generation_safe(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(_RemoteImageButton, "load_url", _hold_remote_load)
panel = ChatPanel()
opened: list[str] = []
panel.open_attachment_requested.connect(opened.append)
first_url = "https://media.example.invalid/glucose-chart.jpg"
panel.set_messages(
[
{
"msg_id": "chat-image-1",
"msg_type": "image",
"image_url": first_url,
"is_from_doctor": False,
"time": "2026-08-10 08:31",
}
]
)
image = panel.findChild(_RemoteImageButton, "DiagnosisChatImage")
assert image is not None
assert image.text() == "查看图片"
assert image.property("loadState") == "loading"
assert image._apply_payload(_png_bytes(640, 480), image._generation)
assert image._rendered_pixmap.size() == QSize(240, 180)
assert image.width() <= 240 and image.height() <= 200
image.click()
assert opened == [first_url]
stale_generation = image._generation
second_url = "https://media.example.invalid/tall-photo.jpg"
panel.set_messages(
[
{
"msg_id": "chat-image-2",
"msg_type": "image",
"image_url": second_url,
"is_from_doctor": True,
"from_staff_name": "陈医生",
"time": "2026-08-10 08:42",
}
]
)
assert not image._apply_payload(_png_bytes(480, 640), stale_generation)
current = next(
item
for item in panel.findChildren(_RemoteImageButton, "DiagnosisChatImage")
if item is not image
)
assert current._apply_payload(_png_bytes(120, 480), current._generation)
assert current._rendered_pixmap.size() == QSize(50, 200)
assert current.property("loadState") == "ready"
panel.close()
application.processEvents()
def test_daily_todo_has_exact_local_toolbar_and_refresh_signal(
application: QApplication,
) -> None:
panel = DailyRecordPanel()
panel.set_editable(True)
refreshed: list[bool] = []
panel.refresh_requested.connect(lambda: refreshed.append(True))
assert panel.todo_add_button.text() == "+ 新增待办"
assert panel.todo_add_button.parentWidget() is panel.todo_toolbar
assert panel.todo_refresh_button.text() == "刷新"
assert panel.todo_refresh_button.parentWidget() is panel.todo_toolbar
assert panel.todo_toolbar.objectName() == "DiagnosisTodoToolbar"
assert all(button.parentWidget() is panel.todo_toolbar for button in panel.todo_group.buttons())
panel.todo_refresh_button.click()
assert refreshed == [True]
panel.set_loading(True)
assert not panel.todo_refresh_button.isEnabled()
panel.set_loading(False)
assert panel.todo_refresh_button.isEnabled()
panel.close()
application.processEvents()
@@ -0,0 +1,466 @@
from __future__ import annotations
import os
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtWidgets import QApplication, QFrame, QLabel, QPushButton, QVBoxLayout
from doctor_workstation.ui import diagnosis_media
from doctor_workstation.ui.diagnosis_drawer import RecordTable
from doctor_workstation.ui.diagnosis_media import (
InlineRecordingPlayer,
RecordingPlaybackCell,
normalize_recording_urls,
preferred_recording_url,
safe_http_url,
)
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
from doctor_workstation.ui.dialogs.diagnosis import DiagnosisDialog, OrderDetailDrawer
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
class _Repository:
def get_prescription_order(self, order_id: int) -> dict[str, Any]:
return {"id": order_id}
def set_revisit_slot_start_offset(self, diagnosis_id: int, offset: int) -> dict[str, Any]:
return {"diagnosis_id": diagnosis_id, "revisit_slot_start_offset": offset}
def upload_call_recording(
self,
path: str,
diagnosis_id: int,
*,
call_record_id: int | None = None,
) -> dict[str, Any]:
return {
"path": path,
"diagnosis_id": diagnosis_id,
"call_record_id": call_record_id,
}
def _rich_order() -> dict[str, Any]:
return {
"id": 801,
"order_no": "RX-20260811-0801",
"diagnosis_id": 501,
"prescription_id": 601,
"amount": 428.5,
"linked_pay_paid_total": 300,
"refund_amount": 20,
"agency_collect_amount": 128.5,
"fulfillment_status": 5,
"prescription_audit_status": 1,
"payment_slip_audit_status": 1,
"doctor_name": "陈医生",
"assistant_name": "赵医助",
"creator_name": "赵医助",
"creator_account": "assistant.zhao",
"create_time": "2026-08-11 09:26",
"recipient_name": "林晓岚",
"recipient_phone": "18600004218",
"shipping_province": "河南省",
"shipping_city": "洛阳市",
"shipping_district": "洛龙区",
"shipping_address": "开元大道 88 号",
"is_follow_up": 1,
"medication_days": 14,
"service_channel": "线上复诊",
"service_package": ["调理服务", "复诊随访"],
"fee_type": 3,
"tracking_number": "SF164208110801",
"express_company": "sf",
"remark_assistant": "工作日下午送达",
"prescription_audit_remark": "辨证与用量已复核",
"payment_slip_audit_remark": "收款凭证已核验",
"prescription": {
"id": 601,
"sn": "RX601",
"patient_name": "林晓岚",
"gender_desc": "",
"age": 34,
"phone": "18600004218",
"prescription_date": "2026-08-11",
"doctor_name": "陈医生",
"prescription_type": "饮片",
"clinical_diagnosis": "气阴两虚",
"dose_count": 14,
"dose_unit": "",
"usage_instruction": "水煎服",
"amount": 428.5,
"audit_status": 1,
"dosage_amount": 180,
"dosage_unit": "g",
"need_decoction": 1,
"times_per_day": 2,
"usage_days": 14,
"dietary_taboo": ["辛辣", "生冷"],
"void_status": 0,
},
"linked_pay_orders": [
{
"id": 9101,
"order_no": "PAY-9101",
"order_type_desc": "药品费用",
"amount": 300,
"status_desc": "已支付",
"creator_name": "赵医助",
"create_time": "2026-08-11 09:32",
}
],
"unlinked_pay_orders": [],
"logistics_trace": {
"state_text": "运输中",
"carrier_label": "顺丰速运",
"traces": [
{
"time": "2026-08-11 16:10",
"status": "运输中",
"context": "快件已离开洛阳集散中心",
},
{
"time": "2026-08-11 13:06",
"status": "已揽收",
"context": "顺丰速运已收取快件",
},
],
},
"logs": [
{
"id": 1,
"admin_name": "赵医助",
"action": "ship",
"summary": "确认发货并填写顺丰运单",
"create_time": "2026-08-11 13:08",
}
],
}
def test_order_offset_copy_tooltip_preview_and_list_columns_are_exact(
application: QApplication,
) -> None:
dialog = DiagnosisDialog(_Repository(), permissions=["*"])
label = dialog.findChild(QLabel, "DiagnosisOrderOffsetLabel")
assert label is not None
assert label.text() == "复诊统计起始偏移"
assert label.toolTip() == diagnosis_module._ORDER_OFFSET_HELP
assert dialog.order_offset_help.toolTip() == diagnosis_module._ORDER_OFFSET_HELP
assert dialog.order_offset_save.text() == "保存"
dialog._editable = True
dialog._can_offset = True
dialog._saved_order_offset = 0
dialog.order_offset.setValue(2)
assert dialog.order_offset_preview.text() == "第 1 笔实单计为三诊"
assert dialog.order_offset_save.isEnabled()
dialog._fill_orders(
[
{
"id": 801,
"order_no": "RX-801",
"global_visit_seq": 4,
"counts_for_revisit_rate": 0,
"amount": 286,
"fulfillment_status": 2,
}
]
)
table = dialog._table_registry["orders"][1]
assert table.horizontalHeaderItem(0).text() == "订单编号"
assert table.item(0, 1).text() == "4诊"
assert table.item(0, 2).text() == ""
assert table.item(0, 3).text() == "¥286.00"
assert table.item(0, 6).text() == "待发货"
dialog.close()
application.processEvents()
def test_order_detail_is_eighty_percent_readonly_drawer_with_real_sections(
application: QApplication,
) -> None:
host = DiagnosisDialog(_Repository(), permissions=["*"])
host.resize(1200, 760)
host.show()
drawer = host._build_order_detail_dialog(_rich_order(), 801)
drawer.show()
application.processEvents()
assert isinstance(drawer, OrderDetailDrawer)
assert drawer.size() == host.size()
assert abs(drawer.drawer_panel.width() - round(host.width() * 0.8)) <= 1
assert drawer.drawer_panel.property("readonly") is True
labels = [label.text() for label in drawer.findChildren(QLabel)]
for section in ("金额概览", "处方详情", "收款记录", "履约与收货信息", "物流轨迹", "操作日志"):
assert section in labels
assert "¥428.50" in labels
assert "¥300.00" in labels
assert "确认发货并填写顺丰运单" in labels
assert "快件已离开洛阳集散中心" in labels
linked = drawer.findChild(RecordTable, "DiagnosisOrderLinkedPayments")
assert linked is not None and linked.rowCount() == 1
drawer.close()
host.close()
application.processEvents()
def test_order_detail_missing_fields_use_explicit_empty_states_without_fake_zero(
application: QApplication,
) -> None:
host = DiagnosisDialog(_Repository(), permissions=["*"])
host.resize(1000, 680)
drawer = host._build_order_detail_dialog({"id": 809}, 809)
drawer.show()
application.processEvents()
labels = [label.text() for label in drawer.findChildren(QLabel)]
amount_values = [
label.text()
for label in drawer.findChildren(QLabel)
if label.property("orderAmountValue") is True
]
assert amount_values == ["", "", "", "", ""]
assert "¥0.00" not in labels
assert "无处方数据(详情接口未返回 prescription" in labels
assert "详情接口未返回关联收款记录字段" in labels
assert "详情接口未返回未关联收款记录字段" in labels
assert "订单详情未返回快递单号,暂无物流轨迹" in labels
assert "详情接口未返回操作日志数据" in labels
drawer.close()
host.close()
application.processEvents()
def test_order_logs_remain_permission_gated(application: QApplication) -> None:
host = DiagnosisDialog(
_Repository(),
permissions=["tcm.prescriptionOrder/detail"],
)
drawer = host._build_order_detail_dialog(_rich_order(), 801)
denied = next(
label for label in drawer.findChildren(QLabel) if label.property("permissionDenied") is True
)
assert denied.text() == "当前账号无操作日志查看权限"
assert "确认发货并填写顺丰运单" not in [label.text() for label in drawer.findChildren(QLabel)]
drawer.close()
host.close()
def test_recording_preference_inline_height_alternates_and_safe_external_open(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
urls = [
"https://bucket.cos.ap-shanghai.myqcloud.com/replay/index.m3u8",
"https://media.example.invalid/replay.mp4",
"https://media.example.invalid/replay.webm",
"file:///C:/private/replay.mp4",
"https://media.example.invalid/replay.mp4",
]
assert normalize_recording_urls(urls) == urls[:-1]
assert preferred_recording_url(urls) == "https://media.example.invalid/replay.mp4"
assert safe_http_url("file:///C:/private/replay.mp4") is None
opened: list[str] = []
monkeypatch.setattr(
diagnosis_media.QDesktopServices,
"openUrl",
lambda url: opened.append(url.toString()) or True,
)
cell = RecordingPlaybackCell(urls, record_id=48)
cell.show()
application.processEvents()
player = cell.findChild(InlineRecordingPlayer, "DiagnosisInlineRecordingPlayer")
assert player is not None
assert player.target == "https://media.example.invalid/replay.mp4"
assert player.maximumHeight() == 180
assert player.property("maximumPlaybackHeight") == 180
assert player._source_attached is False
alternate_buttons = cell.findChildren(QPushButton, "DiagnosisRecordingAlternateLink")
assert [button.text() for button in alternate_buttons] == [
"COS HLS 1",
"链接 2",
"MP4 3",
]
assert alternate_buttons[-1].isEnabled() is False
alternate_buttons[0].click()
assert opened == [urls[0]]
empty = RecordingPlaybackCell([], record_id=49)
empty_state = empty.findChild(QLabel, "DiagnosisEmptyState")
assert empty_state is not None and empty_state.text() == "暂无录制回放"
invalid = RecordingPlaybackCell(["file:///C:/private/replay.mp4"], record_id=50)
invalid_state = invalid.findChild(QLabel, "DiagnosisUnsupportedState")
assert invalid_state is not None and "回放地址无效" in invalid_state.text()
empty.close()
invalid.close()
cell.close()
application.processEvents()
def test_video_table_embeds_player_and_preserves_row_bound_upload(
application: QApplication,
) -> None:
dialog = DiagnosisDialog(_Repository(), permissions=["*"])
dialog._editable = True
dialog._can_video_upload = True
dialog._diagnosis_id = 501
dialog._tab_generations["video"] = 7
uploaded: list[int | None] = []
dialog._upload_call_recording = lambda call_record_id=None: uploaded.append(call_record_id) # type: ignore[method-assign]
dialog._fill_video(
[
{
"id": 48,
"recording_urls_list": [
"https://media.example.invalid/replay.mp4",
"https://media.example.invalid/replay-backup.m3u8",
],
"call_type": 2,
"status": 2,
"recording_status_text": "录制完成",
},
{
"id": 47,
"recording_urls_list": [],
"call_type": 1,
"status": 3,
"recording_status_text": "暂无录制",
},
]
)
preview = QFrame()
preview_layout = QVBoxLayout(preview)
page = dialog._tab_pages["video"]
page.setParent(preview)
page.show()
preview_layout.addWidget(page)
preview.resize(1200, 560)
preview.show()
application.processEvents()
table = dialog._table_registry["video"][1]
playback = table.cellWidget(0, 0)
assert isinstance(playback, RecordingPlaybackCell)
assert playback.property("callRecordId") == 48
player = playback.findChild(InlineRecordingPlayer)
assert player is not None
surface = player.findChild(QFrame, "DiagnosisInlineRecordingSurface")
external = player.findChild(QPushButton, "DiagnosisInlineRecordingExternal")
fallback = player.findChild(QPushButton, "DiagnosisInlineRecordingFallback")
assert surface is not None and external is not None and fallback is not None
assert table.rowHeight(0) >= playback.required_table_row_height()
assert playback.height() >= playback.minimumSizeHint().height()
assert 158 <= player.height() <= 180
assert surface.height() >= 122
assert player.position.width() >= 50
assert player.position.height() >= 12
assert player.time_label.height() >= 15
for control in (player.play_button, external, fallback):
assert control.isVisibleTo(player)
assert control.height() >= 24
for control in (player.play_button, player.position, player.time_label, external, fallback):
assert player.rect().contains(control.geometry())
# Guard the actual rendered pixels: the previous regression produced only
# a 16 px dark strip despite the class-level maximumHeight declaration.
playback_image = playback.grab().toImage()
dark_rows = []
for y in range(playback_image.height()):
dark_pixels = sum(
1
for x in range(playback_image.width())
if max(
playback_image.pixelColor(x, y).red(),
playback_image.pixelColor(x, y).green(),
playback_image.pixelColor(x, y).blue(),
)
<= 55
)
if dark_pixels >= round(playback_image.width() * 0.65):
dark_rows.append(y)
assert dark_rows and dark_rows[-1] - dark_rows[0] + 1 >= 120
assert table.item(1, 0).text() == "暂无录制回放"
assert dialog._recording_players == []
upload_host = table.cellWidget(0, 8)
assert upload_host is not None
upload = upload_host.findChild(QPushButton, "DiagnosisVideoRowUpload")
assert upload is not None
assert upload.property("callRecordId") == 48
assert table.item(0, 8).text() == ""
assert abs(upload_host.rect().center().y() - upload.geometry().center().y()) <= 1
upload.click()
assert uploaded == [48]
preview.close()
dialog.close()
application.processEvents()
def test_inline_player_rejects_stale_owner_generation(application: QApplication) -> None:
class _Owner:
_tab_generations = {"video": 4}
owner = _Owner()
player = InlineRecordingPlayer(
"https://media.example.invalid/replay.mp4",
render_owner=owner,
owner_generation=4,
)
owner._tab_generations["video"] = 5
if player.player is not None:
assert player._attach_source() is False
assert player.play_button.isEnabled() is False
assert "已刷新" in player.placeholder.text()
player.close()
application.processEvents()
def test_drawer_and_inline_player_render_non_empty_images(
application: QApplication,
tmp_path: Any,
) -> None:
host = DiagnosisDialog(_Repository(), permissions=["*"])
host.resize(1100, 720)
host.show()
drawer = host._build_order_detail_dialog(_rich_order(), 801)
drawer.show()
application.processEvents()
order_image = drawer.grab().toImage()
order_path = tmp_path / "order.png"
assert order_image.width() == 1100 and order_image.height() == 720
assert order_image.save(str(order_path), "PNG")
assert order_path.stat().st_size > 10_000
cell = RecordingPlaybackCell(
[
"https://media.example.invalid/replay.mp4",
"https://media.example.invalid/replay.m3u8",
],
record_id=48,
)
cell.resize(520, 235)
cell.show()
application.processEvents()
video_image = cell.grab().toImage()
video_path = tmp_path / "video.png"
assert video_image.save(str(video_path), "PNG")
assert video_path.stat().st_size > 2_000
cell.close()
drawer.close()
host.close()
application.processEvents()
+4
View File
@@ -24,7 +24,11 @@ def test_windows_one_click_entrypoints_and_release_pipeline() -> None:
assert run_script.index("$FrozenExecutable") < run_script.index("Find-Uv")
assert "& $Uv sync --frozen" in run_script
assert '"UV_CACHE_DIR"' in run_script
assert "$ProjectUvCache" in run_script
assert "& $Uv sync --frozen --extra build" in package_script
assert '"UV_CACHE_DIR"' in package_script
assert "$ProjectUvCache" in package_script
assert "& $Npm ci --prefix" in package_script
assert "build_windows.ps1" in package_script
assert "DoctorWorkstation-Windows-x64-$ProjectVersion.zip" in package_script
+118
View File
@@ -0,0 +1,118 @@
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
MEDIA_HOOK = PROJECT_ROOT / "packaging" / "runtime_media_smoke.py"
def read(relative_path: str) -> str:
return (PROJECT_ROOT / relative_path).read_text(encoding="utf-8")
def test_spec_explicitly_collects_qt_multimedia_and_installs_frozen_gate() -> None:
spec = read("packaging/doctor_workstation.spec")
assert '"PySide6.QtMultimedia"' in spec
assert '"PySide6.QtMultimediaWidgets"' in spec
assert "qt_multimedia_hiddenimports" in spec
assert "runtime_hooks=[str(MEDIA_SMOKE_HOOK)]" in spec
assert "Qt6Multimedia*.dll/.dylib/framework" in spec
assert "plugins/multimedia" in spec
def test_media_runtime_hook_is_inert_without_gate_argument() -> None:
result = subprocess.run(
[sys.executable, "-I", "-S", str(MEDIA_HOOK)],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
timeout=10,
check=False,
)
assert result.returncode == 0, result.stderr
def test_media_runtime_hook_fails_when_frozen_components_cannot_import() -> None:
result = subprocess.run(
[sys.executable, "-I", "-S", str(MEDIA_HOOK), "--media-smoke-test"],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
timeout=10,
check=False,
)
assert result.returncode == 70
assert "Qt multimedia smoke gate failed" in result.stderr
def test_media_runtime_hook_constructs_player_and_video_widget_offscreen(
tmp_path: Path,
) -> None:
environment = os.environ.copy()
environment.update(
{
"QT_QPA_PLATFORM": "offscreen",
"QT_LOGGING_RULES": "qt.multimedia.*=false",
"XDG_CONFIG_HOME": str(tmp_path / "config"),
"XDG_CACHE_HOME": str(tmp_path / "cache"),
}
)
result = subprocess.run(
[sys.executable, str(MEDIA_HOOK), "--media-smoke-test"],
cwd=tmp_path,
env=environment,
capture_output=True,
text=True,
timeout=30,
check=False,
)
assert result.returncode == 0, result.stdout + result.stderr
assert "Frozen Qt multimedia smoke gate passed" in result.stdout
def test_windows_build_checks_files_and_runs_media_gate_before_release_archive() -> None:
build = read("scripts/build_windows.ps1")
package = read("scripts/package_windows.ps1")
for required_name in (
"QtMultimedia.pyd",
"QtMultimediaWidgets.pyd",
"Qt6Multimedia.dll",
"Qt6MultimediaWidgets.dll",
"ffmpegmediaplugin.dll",
"windowsmediaplugin.dll",
):
assert required_name in build
assert "Assert-FrozenMultimedia -Artifact $Artifact" in build
assert build.index('Argument "--media-smoke-test"') < build.index('Write-Host "Build complete')
assert "packaging\\runtime_media_smoke.py" in package
assert package.index("& $BuildScript") < package.index("Compress-Archive")
def test_macos_build_checks_modules_frameworks_plugins_and_runs_gate_before_zip() -> None:
build = read("scripts/build_macos.sh")
package = read("scripts/package_macos.sh")
entry_check = read("scripts/check_macos_entrypoints.sh")
for contract in (
"QtMultimedia*.so",
"QtMultimediaWidgets*.so",
"QtMultimedia.framework",
"QtMultimediaWidgets.framework",
"*Qt6Multimedia*.dylib",
"*/plugins/multimedia",
"*mediaplugin*.dylib",
"--media-smoke-test",
):
assert contract in build
assert build.index('"--media-smoke-test"') < build.index('echo "Build complete')
assert package.index('build_macos.sh"') < package.index("/usr/bin/ditto -c -k")
assert 'check_macos_entrypoints.sh"' in package
assert 'index_mode" == "100755"' in entry_check
@@ -0,0 +1,177 @@
from __future__ import annotations
import os
from types import SimpleNamespace
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QPoint, Qt
from PySide6.QtWidgets import QApplication, QTabWidget, QWidget
from doctor_workstation.ui.dialogs.prescription import PrescriptionEditorDialog
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
def _repository() -> SimpleNamespace:
return SimpleNamespace(list_medicines=lambda **_kwargs: {"lists": [], "count": 0})
def _seed() -> dict[str, object]:
return {
"id": 502,
"diagnosis_id": 301,
"appointment_id": 401,
"patient_name": "林晓岚",
"phone": "13800138000",
"gender": 0,
"age": 34,
"visit_no": "1K00000401",
"tongue": "舌淡红、苔薄白",
"pulse": "面色少华",
"pulse_condition": "脉细",
"clinical_diagnosis": "气阴两虚",
"doctor_name": "陈医生",
"herbs": [
{"medicine_id": 11, "name": "黄芪", "dosage": 15, "formula_type": "主方"},
{"medicine_id": 12, "name": "党参", "dosage": 12, "formula_type": "主方"},
{"medicine_id": 13, "name": "白术", "dosage": 10, "formula_type": "主方"},
{"medicine_id": 14, "name": "茯苓", "dosage": 12, "formula_type": "主方"},
{"medicine_id": 15, "name": "酸枣仁", "dosage": 9, "formula_type": "辅方"},
],
}
def _show(editor: PrescriptionEditorDialog, application: QApplication) -> None:
editor.show()
for _ in range(5):
application.processEvents()
def test_editor_is_a_fixed_header_body_footer_right_drawer(
application: QApplication,
) -> None:
host = QWidget()
host.resize(1440, 900)
host.show()
editor = PrescriptionEditorDialog(
_repository(),
_seed(),
current_user=SimpleNamespace(id=7, name="陈医生"),
parent=host,
)
_show(editor, application)
origin = host.mapToGlobal(QPoint(0, 0))
assert editor.windowFlags() & Qt.WindowType.FramelessWindowHint
assert editor.width() == 1200
assert editor.height() == host.height()
assert editor.x() + editor.width() == origin.x() + host.width()
assert editor.y() == origin.y()
assert not editor.findChildren(QTabWidget)
assert editor.tabs.count() == 4
assert editor.tabs.tabText(2) == "剂型与用法"
assert editor.body_scroll.verticalScrollBar().maximum() > 0
assert editor.body_scroll.horizontalScrollBar().maximum() == 0
section_tops = [section.geometry().top() for section in editor.section_widgets]
assert section_tops == sorted(section_tops)
assert [section.parentWidget() for section in editor.section_widgets] == [
editor.body_content
] * 5
header_y = editor.header.mapToGlobal(QPoint(0, 0)).y()
footer_y = editor.footer.mapToGlobal(QPoint(0, 0)).y()
editor.body_scroll.verticalScrollBar().setValue(
editor.body_scroll.verticalScrollBar().maximum()
)
application.processEvents()
assert editor.header.mapToGlobal(QPoint(0, 0)).y() == header_y
assert editor.footer.mapToGlobal(QPoint(0, 0)).y() == footer_y
editor.close()
host.close()
application.processEvents()
def test_drawer_uses_full_width_on_a_narrow_host(application: QApplication) -> None:
host = QWidget()
host.resize(760, 720)
host.show()
editor = PrescriptionEditorDialog(_repository(), _seed(), parent=host)
_show(editor, application)
origin = host.mapToGlobal(QPoint(0, 0))
assert editor.width() == host.width()
assert editor.x() == origin.x()
assert editor.body_scroll.horizontalScrollBar().maximum() == 0
assert editor.herbs._grid_columns == 3
editor.close()
host.close()
application.processEvents()
def test_duplicate_guard_and_admin_dto_survive_the_visual_restructure(
application: QApplication,
) -> None:
source = _seed()
source["herbs"] = [
{"medicine_id": 11, "name": "黄芪", "dosage": 15, "formula_type": "主方"},
{"medicine_id": 11, "name": " 黄芪 ", "dosage": 9, "formula_type": "辅方"},
]
editor = PrescriptionEditorDialog(
_repository(),
source,
mode="edit",
current_user=SimpleNamespace(id=7, name="陈医生"),
)
editor.signature._has_strokes = True
payload = editor.payload()
assert payload["id"] == 502
assert payload["diagnosis_id"] == 301
assert payload["herbs"][0]["formula_type"] == "主方"
assert payload["herbs"][1]["formula_type"] == "辅方"
assert all(bool(row.property("duplicate")) for row in editor.herbs.rows)
editor.accept()
assert editor.result() == 0
assert editor.tabs.currentIndex() == 1
assert "药材不可重复:黄芪" in editor.validation.label.text()
editor.close()
application.processEvents()
def test_locked_template_rows_keep_the_existing_mutation_contract(
application: QApplication,
) -> None:
source = _seed()
source["herbs"] = [
{
"medicine_id": 11,
"name": "黄芪",
"dosage": 15,
"formula_type": "主方",
"locked": True,
}
]
editor = PrescriptionEditorDialog(_repository(), source)
assert editor.herbs.locked
assert not editor.add_main_button.isEnabled()
assert not editor.add_aux_button.isEnabled()
assert editor.import_library_button.isEnabled()
assert editor.paste_button.isEnabled()
assert not editor.herbs.rows[0].medicine.isEnabled()
assert not editor.herbs.rows[0].dosage.isEnabled()
assert editor.herbs.rows[0].remove_button.isHidden()
assert editor.payload()["herbs"][0]["locked"] is True
editor.close()
application.processEvents()
+176
View File
@@ -0,0 +1,176 @@
from __future__ import annotations
import os
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtWidgets import QApplication, QWidget
from doctor_workstation.ui import shell as shell_module
from doctor_workstation.ui.shell import NavigationItem, ShellWindow
class _ShellPageDouble(QWidget):
def __init__(
self,
_repository: Any,
*,
permissions: Any,
current_user: Any,
) -> None:
super().__init__()
self.permissions = permissions
self.current_user = current_user
self.refresh_count = 0
def refresh(self) -> None:
self.refresh_count += 1
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
@pytest.fixture
def shell_window(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> ShellWindow:
navigation = [
NavigationItem(key, title, glyph, _ShellPageDouble, (permission,))
for key, title, glyph, permission in (
("reception", "接诊台", "", "doctor.appointment/lists"),
("prescription_library", "我的处方库", "", "tcm.prescriptionLibrary/lists"),
("prescriptions", "已开处方", "", "tcm.prescription/lists"),
("patients", "我的患者", "", "firstvisit.myPatient/lists"),
("consultations", "问诊列表", "", "tcm.diagnosis/lists"),
)
]
monkeypatch.setattr(
shell_module,
"_resolve_navigation",
lambda _menu, _permissions, *, demo_mode: [(item, item.title) for item in navigation],
)
window = ShellWindow(
object(),
{
"user": {"name": "陈医生", "department_name": "中医门诊", "role_ids": [1]},
"demo_mode": True,
},
permissions={item.permissions[0] for item in navigation},
)
window.show()
application.processEvents()
yield window
window.close()
application.processEvents()
def test_shell_matches_admin_geometry_at_both_acceptance_sizes(
application: QApplication,
shell_window: ShellWindow,
) -> None:
for width, height in ((1024, 640), (1440, 900)):
shell_window.resize(width, height)
application.processEvents()
assert shell_window.sidebar.width() == 183
assert shell_window.topbar.height() == 50
assert shell_window.tabs_host.height() == 40
assert shell_window.workspace.width() == width - 183
assert shell_window.stack.width() == width - 183
assert shell_window.stack.height() == height - 90
assert shell_window.stack.geometry().right() < shell_window.workspace.width()
assert shell_window.stack.geometry().bottom() < shell_window.workspace.height()
image = shell_window.grab().toImage()
assert image.pixelColor(100, 100).name().lower() == "#1d2124"
assert image.pixelColor(200, 10).name().lower() == "#ffffff"
assert image.pixelColor(190, 70).name().lower() == "#ffffff"
def test_every_visible_page_navigates_and_visited_tabs_track_active_page(
shell_window: ShellWindow,
) -> None:
changed: list[str] = []
shell_window.page_changed.connect(changed.append)
for key in (
"reception",
"prescription_library",
"prescriptions",
"patients",
"consultations",
):
assert shell_window.navigate(key)
assert shell_window.stack.currentWidget() is shell_window.pages[key]
assert shell_window.nav_buttons[key].isChecked()
assert shell_window.tab_bar.tabData(shell_window.tab_bar.currentIndex()) == key
assert shell_window.visited_tab_keys() == (
"reception",
"prescription_library",
"prescriptions",
"patients",
"consultations",
)
assert changed[-5:] == [
"reception",
"prescription_library",
"prescriptions",
"patients",
"consultations",
]
def test_non_fixed_tabs_close_and_active_close_renavigates(
shell_window: ShellWindow,
) -> None:
for key in ("patients", "consultations"):
assert shell_window.navigate(key)
assert not shell_window.close_tab("reception")
assert shell_window.close_tab("patients")
assert "patients" not in shell_window.visited_tab_keys()
assert shell_window.tab_bar.tabData(shell_window.tab_bar.currentIndex()) == "consultations"
assert shell_window.close_current_tab()
assert shell_window.tab_bar.tabData(shell_window.tab_bar.currentIndex()) == "reception"
assert shell_window.stack.currentWidget() is shell_window.pages["reception"]
assert shell_window.nav_buttons["reception"].isChecked()
shell_window.close_all_tabs()
assert shell_window.visited_tab_keys() == ("reception",)
assert shell_window.stack.currentWidget() is shell_window.pages["reception"]
def test_sidebar_collapse_preserves_active_navigation(shell_window: ShellWindow) -> None:
assert shell_window.navigate("consultations")
shell_window.toggle_sidebar()
assert shell_window.sidebar.width() == 64
assert shell_window.nav_buttons["consultations"].text() == ""
assert shell_window.nav_buttons["consultations"].isChecked()
shell_window.toggle_sidebar()
assert shell_window.sidebar.width() == 183
assert shell_window.nav_buttons["consultations"].text().endswith("问诊列表")
assert shell_window.nav_buttons["consultations"].isChecked()
def test_shell_directional_controls_have_no_unicode_arrow_text(
shell_window: ShellWindow,
) -> None:
assert shell_window.fold_button.text() == ""
assert shell_window.refresh_button.text() == ""
assert shell_window.fullscreen_button.text() == ""
assert shell_window.tabs_menu_button.text() == ""
assert all(
arrow not in button.text()
for button in shell_window.findChildren(QWidget)
if hasattr(button, "text") and callable(button.text)
for arrow in ("", "", "", "", "", "", "", "")
)