Files
zyt/app/tests/test_patients_ui.py
T
2026-08-14 14:37:30 +08:00

644 lines
21 KiB
Python

from __future__ import annotations
import os
from types import SimpleNamespace
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QDate
from PySide6.QtWidgets import QApplication, QDialogButtonBox, QInputDialog, QLabel
from doctor_workstation.core import PermissionSet
from doctor_workstation.services import DemoDoctorRepository
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
from doctor_workstation.ui.dialogs.diagnosis import DiagnosisDialog
from doctor_workstation.ui.pages import patients as patients_module
from doctor_workstation.ui.pages.patients import (
PatientListWorkspace,
PatientOrdersWorkspace,
PatientProgressWorkspace,
PatientsPage,
_AppointmentDialog,
_AssignDialog,
_OrderDetailDialog,
_OrderEditDialog,
_PaymentDialog,
_RefundDialog,
)
from doctor_workstation.ui.shell import NAVIGATION, ShellWindow, _resolve_navigation
@pytest.fixture(scope="module")
def application() -> QApplication:
return QApplication.instance() or QApplication([])
@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, **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()
monkeypatch.setattr(patients_module, "run_async", run_immediately)
monkeypatch.setattr(diagnosis_module, "run_async", run_immediately)
def test_shell_resolves_dynamic_menu_order_visibility_and_canonical_permissions() -> None:
permissions = PermissionSet(
[
"firstvisit.myPatient/lists",
"tcm.diagnosis/lists",
"tcm.prescription/lists",
"doctor.appointment/lists",
]
)
menu = [
{
"name": "隐藏接诊",
"perms": "doctor.appointment/lists",
"sort": 999,
"is_show": 0,
},
{
"name": "诊疗中心",
"sort": 20,
"children": [
{
"name": "患者工作区",
"component": "first_visit/my_patients/index",
"sort": 80,
},
{
"name": "问诊工作区",
"perms": "tcm.diagnosis/lists",
"sort": 60,
},
],
},
{
"name": "停用处方",
"perms": "tcm.prescription/lists",
"sort": 30,
"is_disable": 1,
},
]
resolved = _resolve_navigation(menu, permissions, demo_mode=False)
assert [(item.key, title) for item, title in resolved] == [
("patients", "接诊台"),
("consultations", "问诊工作区"),
]
assert _resolve_navigation([], permissions, demo_mode=False) == []
assert [item.key for item, _title in _resolve_navigation([], permissions, demo_mode=True)] == [
"reception",
"appointments",
"prescriptions",
"patients",
"consultations",
]
assert (
_resolve_navigation(
[{"perms": "firstvisit.myPatient/lists"}],
PermissionSet(["firstvisit.myPatient.lists"]),
demo_mode=False,
)
== []
)
def test_patient_page_runs_all_three_demo_workspaces_and_progress_timer(
application: QApplication,
immediate_async: None,
) -> None:
repository = DemoDoctorRepository()
session = repository.login(repository.DEMO_ACCOUNT, repository.DEMO_PASSWORD)
page = PatientsPage(repository, permissions=session.permissions, current_user=session.user)
page.resize(808, 560)
page.show()
application.processEvents()
assert [page.tabs.tabText(index) for index in range(page.tabs.count())] == [
"患者列表",
"订单管理",
"面诊进度",
]
assert page.patient_workspace.table.rowCount() > 0
assert page.patient_workspace.scope_label.text() != ""
page.tabs.setCurrentIndex(1)
application.processEvents()
assert page.order_workspace.table.rowCount() > 0
assert page.order_workspace.metrics["orders"].text() == "1"
assert page.order_workspace.metrics["amount"].text() == "¥368.00"
page.tabs.setCurrentIndex(2)
application.processEvents()
assert page.progress_workspace.timer.isActive()
assert page.progress_workspace.schedule_table.rowCount() == 7
page.tabs.setCurrentIndex(0)
assert not page.progress_workspace.timer.isActive()
page.close()
application.processEvents()
def test_patient_refresh_generation_ignores_late_results(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
callbacks: list[dict[str, Any]] = []
def queue_async(_function: Any, **options: Any) -> object:
callbacks.append(options)
return object()
monkeypatch.setattr(patients_module, "run_async", queue_async)
workspace = PatientListWorkspace(SimpleNamespace(), PermissionSet(["*"]))
workspace.refresh()
workspace.refresh()
newer = {
"lists": [{"id": 2, "diagnosis_id": 2, "patient_name": "新结果"}],
"count": 1,
}
stale = {
"lists": [{"id": 1, "diagnosis_id": 1, "patient_name": "旧结果"}],
"count": 1,
}
callbacks[1]["on_success"](newer)
callbacks[0]["on_success"](stale)
application.processEvents()
assert workspace.table.rowCount() == 1
assert workspace.table.item(0, 1).text().startswith("新结果")
workspace.close()
def test_workspace_queries_use_frozen_page_no_contract(
application: QApplication,
immediate_async: None,
) -> None:
calls: list[tuple[str, dict[str, Any]]] = []
class Repository:
def list_patients(self, **kwargs: Any) -> dict[str, Any]:
calls.append(("patients", kwargs))
return {"lists": [], "count": 0}
def patient_orders(self, **kwargs: Any) -> dict[str, Any]:
calls.append(("orders", kwargs))
return {"lists": [], "count": 0}
def patient_progress(self, **kwargs: Any) -> dict[str, Any]:
calls.append(("progress", kwargs))
return {"lists": [], "count": 0}
repository = Repository()
patient = PatientListWorkspace(repository, PermissionSet(["*"]))
orders = PatientOrdersWorkspace(repository, PermissionSet(["*"]))
progress = PatientProgressWorkspace(repository)
patient.refresh()
orders.refresh()
progress.refresh()
assert [name for name, _kwargs in calls] == ["patients", "orders", "progress"]
for _name, kwargs in calls:
assert kwargs["page_no"] == 1
assert kwargs["page_size"] == 15
assert "page" not in kwargs
patient.close()
orders.close()
progress.close()
application.processEvents()
def test_workspace_workers_use_gui_thread_query_snapshots(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
queued: list[tuple[Any, tuple[Any, ...], dict[str, Any]]] = []
patient_calls: list[dict[str, Any]] = []
order_calls: list[dict[str, Any]] = []
def queue_async(function: Any, *args: Any, **options: Any) -> object:
queued.append((function, args, options))
return object()
class Repository:
def list_patients(self, **kwargs: Any) -> dict[str, Any]:
patient_calls.append(kwargs)
return {"lists": [], "count": 0}
def patient_orders(self, **kwargs: Any) -> dict[str, Any]:
order_calls.append(kwargs)
return {"lists": [], "count": 0}
repository = Repository()
patient = PatientListWorkspace(repository, PermissionSet(["*"]))
orders = PatientOrdersWorkspace(repository, PermissionSet(["*"]))
monkeypatch.setattr(patients_module, "run_async", queue_async)
patient.keyword_edit.setText("captured patient")
patient.refresh()
patient.keyword_edit.setText("changed patient")
function, args, _options = queued[0]
function(*args)
orders.keyword_edit.setText("captured order")
orders.refresh()
orders.keyword_edit.setText("changed order")
function, args, _options = queued[1]
function(*args)
assert patient_calls[0]["keyword"] == "captured patient"
assert order_calls[0]["keyword"] == "captured order"
patient.close()
orders.close()
application.processEvents()
def test_appointment_form_uses_rosters_slots_and_diagnosis_id_contract(
application: QApplication,
immediate_async: None,
) -> None:
tomorrow = QDate.currentDate().addDays(1).toString("yyyy-MM-dd")
appointment_queries: list[dict[str, Any]] = []
roster_queries: list[dict[str, Any]] = []
slot_queries: list[dict[str, Any]] = []
class Repository:
def list_diagnosis_doctors(self) -> list[dict[str, Any]]:
return [{"id": 77, "name": "陈医生", "department_name": "中医科"}]
def get_dictionary(self, dictionary_type: str) -> list[dict[str, Any]]:
assert dictionary_type == "channels"
return [{"id": 1, "name": "线上复诊", "value": "online", "status": 1, "sort": 10}]
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
appointment_queries.append(kwargs)
return {"lists": [], "count": 0}
def list_appointment_rosters(self, **kwargs: Any) -> dict[str, Any]:
roster_queries.append(kwargs)
return {"lists": [{"date": tomorrow}], "count": 1}
def get_available_appointment_slots(self, **kwargs: Any) -> dict[str, Any]:
slot_queries.append(kwargs)
return {"slots": [{"time": "09:30-10:00", "available": True, "quota": 2}]}
row = {
"id": 501,
"diagnosis_id": 501,
"patient_id": 999,
"source_patient_id": 999,
"patient_name": "林晓岚",
"doctor_id": 77,
}
dialog = _AppointmentDialog(row, repository=Repository())
application.processEvents()
dialog.channel_source.setCurrentIndex(dialog.channel_source.findData("online"))
dialog.slot_combo.setCurrentIndex(dialog.slot_combo.findData("09:30-10:00"))
dialog.remark.setPlainText("复诊预约")
application.processEvents()
payload = dialog.payload()
assert appointment_queries[0]["patient_id"] == 501
assert roster_queries[0]["doctor_id"] == 77
assert slot_queries[0] == {
"doctor_id": 77,
"appointment_date": tomorrow,
"period": "all",
}
assert dialog.ok_button.isEnabled()
assert 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 payload["patient_id"] != row["source_patient_id"]
dialog.close()
application.processEvents()
def test_payment_and_refund_forms_expose_full_contract(
application: QApplication,
) -> None:
payment = _PaymentDialog({"amount": 368, "linked_pay_paid_total": 100})
payment.pay_create_type.setCurrentIndex(payment.pay_create_type.findData("express_cod"))
payment.pay_amount.setValue(268)
payment.pay_remark.setPlainText("货到代收")
payment.completion_request.setChecked(True)
assert payment.payload() == {
"order_type": 3,
"pay_amount": 268.0,
"pay_remark": "货到代收",
"completion_request": 1,
"pay_create_type": "express_cod",
}
refund = _RefundDialog({"amount": 368})
refund.reason.setPlainText("患者取消")
assert refund.payload() == {"reason": "患者取消", "refund_amount": None}
refund.specify_amount.setChecked(True)
refund.refund_amount.setValue(88.5)
assert refund.payload() == {"reason": "患者取消", "refund_amount": 88.5}
payment.close()
refund.close()
application.processEvents()
def test_patient_business_dialogs_expose_shared_visual_contract(
application: QApplication,
) -> None:
detail = {
"id": 81,
"order_no": "ORDER-81",
"patient_name": "鹿立核",
"recipient_name": "鹿立核",
"recipient_phone": "13800138000",
"shipping_address": "北京市朝阳区测试路 8 号",
"amount": 368,
}
dialogs = (
_PaymentDialog(detail),
_RefundDialog(detail),
_AssignDialog(
detail,
[{"id": 7, "name": "陈医助", "department_name": "中医门诊"}],
),
_OrderDetailDialog(detail),
_OrderEditDialog(detail),
)
assert [dialog.objectName() for dialog in dialogs] == [
"PatientPaymentDialog",
"PatientRefundDialog",
"PatientAssignDialog",
"PatientOrderDetailDialog",
"PatientOrderEditDialog",
]
assert all(dialog.property("businessDialog") is True for dialog in dialogs)
assert all(
any(label.property("dialogRole") == "title" for label in dialog.findChildren(QLabel))
for dialog in dialogs
)
assert dialogs[0].findChild(QDialogButtonBox).button(
QDialogButtonBox.StandardButton.Ok
).property("variant") == "primary"
assert dialogs[1].findChild(QDialogButtonBox).button(
QDialogButtonBox.StandardButton.Ok
).property("variant") == "danger"
assert dialogs[2].findChild(QDialogButtonBox).button(
QDialogButtonBox.StandardButton.Ok
).property("variant") == "primary"
assert dialogs[4].findChild(QDialogButtonBox).button(
QDialogButtonBox.StandardButton.Ok
).property("variant") == "primary"
for dialog in dialogs:
dialog.close()
application.processEvents()
def test_out_of_order_mutation_successes_each_trigger_reconciliation(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
) -> None:
page = PatientsPage(SimpleNamespace(), permissions=PermissionSet(["*"]))
callbacks: list[dict[str, Any]] = []
reconciliations: list[str] = []
def queue_async(_function: Any, **options: Any) -> object:
callbacks.append(options)
return object()
monkeypatch.setattr(patients_module, "run_async", queue_async)
monkeypatch.setattr(page, "_after_mutation", lambda: reconciliations.append("refresh"))
page._run_action("first", lambda: None, success="first done")
page._run_action("second", lambda: None, success="second done")
callbacks[1]["on_success"](None)
callbacks[0]["on_success"](None)
callbacks[1]["on_finished"]()
callbacks[0]["on_finished"]()
assert reconciliations == ["refresh", "refresh"]
assert page._pending_action_tokens == set()
page.close()
application.processEvents()
def test_diagnosis_dialog_loads_histories_and_saves_canonical_fields(
application: QApplication,
immediate_async: None,
) -> None:
repository = DemoDoctorRepository()
repository.login(repository.DEMO_ACCOUNT, repository.DEMO_PASSWORD)
diagnosis_id = repository.list_patients().items[0].diagnosis_id
before = repository.get_diagnosis_detail(diagnosis_id)
dialog = DiagnosisDialog(repository)
dialog.open_for(diagnosis_id, editable=True)
application.processEvents()
assert (
dialog.edit_fields["chief_complaint"].toPlainText()
== before["diagnosis"]["chief_complaint"]
)
assert dialog.appointment_table.rowCount() >= 1
dialog.edit_fields["chief_complaint"].setPlainText("离屏回归主诉")
dialog._save()
assert (
repository.get_diagnosis_detail(diagnosis_id)["diagnosis"]["chief_complaint"]
== "离屏回归主诉"
)
dialog.close()
application.processEvents()
def test_order_action_matrix_requires_exact_permissions_and_states(
application: QApplication,
) -> None:
codes = {
"tcm.prescriptionOrder/detail",
"tcm.prescriptionOrder/edit",
"tcm.prescriptionOrder/auditPrescription",
"tcm.prescriptionOrder/auditPayment",
"tcm.prescriptionOrder/ddcode",
"tcm.prescriptionOrder/ship",
"tcm.prescriptionOrder/addPayOrder",
"tcm.prescriptionOrder/complete",
"tcm.prescriptionOrder/refund",
"tcm.prescriptionOrder/withdraw",
"tcm.prescriptionOrder/uploadToPharmacy",
}
workspace = PatientOrdersWorkspace(SimpleNamespace(), PermissionSet(codes))
pending = {
"id": 1,
"fulfillment_status": 1,
"prescription_audit_status": 0,
"payment_slip_audit_status": 0,
"amount": 100,
"linked_pay_paid_total": 0,
}
shipped = {
**pending,
"fulfillment_status": 5,
"prescription_audit_status": 1,
"payment_slip_audit_status": 1,
"linked_pay_paid_total": 80,
}
assert [key for key, _label, _danger in workspace._available_actions(pending)] == [
"edit",
"audit_prescription",
"ddcode",
"withdraw",
]
assert [key for key, _label, _danger in workspace._available_actions(shipped)] == [
"revoke_pay_audit",
"ddcode",
"add_pay_order",
"complete",
"refund",
"upload_pharmacy",
]
remote_locked = {**pending, "gancao_reciperl_order_no": "GC-REMOTE-1"}
assert [key for key, _label, _danger in workspace._available_actions(remote_locked)] == [
"audit_prescription",
"ddcode",
]
alias_only = PatientOrdersWorkspace(
SimpleNamespace(), PermissionSet(["tcm.prescriptionOrder.edit"])
)
assert alias_only._available_actions(pending) == []
workspace.close()
alias_only.close()
application.processEvents()
def test_order_audit_action_uses_repository_contract_values(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
repository = DemoDoctorRepository()
session = repository.login(repository.DEMO_ACCOUNT, repository.DEMO_PASSWORD)
page = PatientsPage(repository, permissions=session.permissions, current_user=session.user)
order_id = repository.patient_orders().items[0]["id"]
repository.revoke_patient_order_payment_audit(order_id)
repository.revoke_patient_order_prescription_audit(order_id)
row = repository.get_patient_order(order_id)
monkeypatch.setattr(
QInputDialog,
"getItem",
staticmethod(lambda *_args, **_kwargs: ("通过", True)),
)
monkeypatch.setattr(
QInputDialog,
"getText",
staticmethod(lambda *_args, **_kwargs: ("离屏审核", True)),
)
page._handle_order_action("audit_prescription", row)
assert repository.get_patient_order(row["id"])["prescription_audit_status"] == 1
page.close()
application.processEvents()
def test_shell_uses_demo_session_menu_and_fits_minimum_window(
application: QApplication,
immediate_async: None,
) -> None:
repository = DemoDoctorRepository()
session = repository.login(repository.DEMO_ACCOUNT, repository.DEMO_PASSWORD)
reception_menu = next(
row for row in session.menu if row.get("perms") == "doctor.appointment/lists"
)
reception_menu["name"] = "接诊台"
reception_menu["sort"] = 99
session.menu = [reception_menu]
shell = ShellWindow(
repository,
{"session": session, "demo_mode": True},
permissions=session.permissions,
)
shell.resize(1024, 640)
shell.show()
application.processEvents()
assert list(shell.pages) == ["reception"]
assert shell.nav_buttons["reception"].text().endswith("接诊台")
assert shell.minimumWidth() == 1024
assert shell.minimumHeight() == 640
assert shell.size().width() == 1024
assert shell.size().height() == 640
assert {item.key for item in NAVIGATION} == {
"reception",
"appointments",
"prescription_library",
"prescriptions",
"patients",
"consultations",
}
shell.close()
application.processEvents()
def test_patient_list_reference_geometry_and_row_actions(
application: QApplication,
immediate_async: None,
) -> None:
repository = DemoDoctorRepository()
session = repository.login(repository.DEMO_ACCOUNT, repository.DEMO_PASSWORD)
page = PatientsPage(repository, permissions=session.permissions, current_user=session.user)
page.resize(1460, 820)
page.show()
application.processEvents()
page.patient_workspace.refresh()
application.processEvents()
workspace = page.patient_workspace
assert all(
button.minimumHeight() == 56 and button.maximumHeight() == 56
for button in workspace.summary_buttons.values()
)
assert workspace.table.objectName() == "PatientTable"
assert workspace.table.columnCount() == 10
assert workspace.table.horizontalHeaderItem(9).text() == "操作"
if workspace.table.rowCount():
assert workspace.table.rowHeight(0) == 40
assert workspace.table.cellWidget(0, 0) is not None
assert workspace.table.cellWidget(0, 9) is not None
page.close()
application.processEvents()