230 lines
8.1 KiB
Python
230 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import Qt
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QPushButton
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.ui.dialogs import ai_consult_picker, diagnosis, prescription
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def run(function: Any, *, on_success: Any, on_error: Any, on_finished: Any = None) -> None:
|
|
try:
|
|
result = function()
|
|
except Exception as error:
|
|
on_error(error)
|
|
else:
|
|
on_success(result)
|
|
if on_finished:
|
|
on_finished()
|
|
|
|
for module in (ai_consult_picker, diagnosis, prescription):
|
|
monkeypatch.setattr(module, "run_async", run)
|
|
|
|
|
|
class ListRepository:
|
|
def __init__(self) -> None:
|
|
self.calls: list[dict[str, Any]] = []
|
|
self.fail_page = 0
|
|
|
|
def _list(self, **filters: Any) -> dict[str, Any]:
|
|
self.calls.append(filters)
|
|
page, size = filters["page_no"], filters["page_size"]
|
|
if page == self.fail_page:
|
|
raise RuntimeError("暂时无法加载")
|
|
start = (page - 1) * size + 1
|
|
return {
|
|
"lists": [
|
|
{
|
|
"id": number,
|
|
"diagnosis_id": number,
|
|
"patient_id": 1000 + number,
|
|
"patient_name": f"患者{number}",
|
|
"prescription_id": 77,
|
|
"prescription_name": f"处方{number}",
|
|
"herbs": [{"name": "白术", "dosage": 10}],
|
|
"order_no": f"ORDER{number}",
|
|
}
|
|
for number in range(start, min(start + size, 46))
|
|
],
|
|
"count": 45,
|
|
}
|
|
|
|
list_prescription_templates = _list
|
|
list_prescription_orders = _list
|
|
list_ai_patient_options = _list
|
|
|
|
|
|
def scroll_to_bottom(table: Any) -> None:
|
|
scrollbar = table.verticalScrollBar()
|
|
scrollbar.setValue(scrollbar.maximum())
|
|
QTest.qWait(70)
|
|
|
|
|
|
@pytest.mark.parametrize("kind", ["template", "order", "ai"])
|
|
def test_scroll_appends_and_retains_selected_business_identity(
|
|
application: QApplication, immediate_async: None, kind: str
|
|
) -> None:
|
|
repository = ListRepository()
|
|
if kind == "template":
|
|
dialog = prescription.TemplateImportDialog(repository, 42)
|
|
controller = dialog.infinite_list
|
|
elif kind == "order":
|
|
dialog = prescription.PrescriptionOrderListDialog(repository, prescription_id=77)
|
|
controller = dialog.infinite_list
|
|
else:
|
|
dialog = ai_consult_picker.AiConsultTargetDialog(repository, PermissionSet(["*"]))
|
|
controller = dialog.pager
|
|
dialog.resize(980, 520)
|
|
dialog.show()
|
|
application.processEvents()
|
|
try:
|
|
page_size = controller.page_size
|
|
assert dialog.table.rowCount() == page_size
|
|
assert not any(
|
|
button.text() in {"上一页", "下一页"} for button in dialog.findChildren(QPushButton)
|
|
)
|
|
dialog.table.selectRow(2)
|
|
scroll_to_bottom(dialog.table)
|
|
assert [call["page_no"] for call in repository.calls] == [1, 2]
|
|
assert dialog.table.rowCount() == page_size * 2
|
|
assert dialog.table.currentRow() == 2
|
|
assert all(
|
|
dialog.table.item(row, column).data(Qt.ItemDataRole.CheckStateRole) is None
|
|
for row in range(dialog.table.rowCount())
|
|
for column in range(dialog.table.columnCount())
|
|
)
|
|
repository.fail_page = 3
|
|
scroll_to_bottom(dialog.table)
|
|
assert controller.retry_button.isVisible()
|
|
assert dialog.table.rowCount() == page_size * 2
|
|
repository.fail_page = 0
|
|
controller.retry_button.click()
|
|
assert dialog.table.rowCount() == 45
|
|
assert not controller.has_more
|
|
assert dialog.table.currentRow() == 2
|
|
if kind == "template":
|
|
dialog.accept()
|
|
assert dialog.selected_template()["id"] == 3
|
|
elif kind == "ai":
|
|
dialog.accept()
|
|
assert dialog.selected_target().diagnosis_id == 3
|
|
assert dialog.selected_target().patient_id == 1003
|
|
else:
|
|
assert dialog.table.item(2, 0).data(Qt.ItemDataRole.UserRole)["id"] == 3
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_template_query_supersedes_pending_load_and_keeps_creator_scope(
|
|
application: QApplication, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
pending: list[tuple[Any, dict[str, Any]]] = []
|
|
monkeypatch.setattr(
|
|
prescription, "run_async", lambda function, **callbacks: pending.append((function, callbacks))
|
|
)
|
|
repository = ListRepository()
|
|
dialog = prescription.TemplateImportDialog(repository, 42)
|
|
dialog.show()
|
|
application.processEvents()
|
|
try:
|
|
dialog.name_edit.setText("白术")
|
|
dialog.formula_combo.setCurrentIndex(1)
|
|
dialog.search()
|
|
assert len(pending) == 2
|
|
latest_function, latest_callbacks = pending[1]
|
|
latest_callbacks["on_success"](latest_function())
|
|
old_function, old_callbacks = pending[0]
|
|
old_callbacks["on_success"](old_function())
|
|
old_callbacks["on_error"](RuntimeError("过期错误"))
|
|
assert dialog.table.rowCount() == 15
|
|
assert not dialog.banner.isVisible()
|
|
assert repository.calls[0] == {
|
|
"page_no": 1,
|
|
"page_size": 15,
|
|
"prescription_name": "白术",
|
|
"formula_type": "主方",
|
|
"prescribing_creator_id": 42,
|
|
}
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_ai_picker_continues_when_first_page_has_no_valid_diagnosis_targets(
|
|
application: QApplication, immediate_async: None
|
|
) -> None:
|
|
class Repository(ListRepository):
|
|
def list_ai_patient_options(self, **filters: Any) -> dict[str, Any]:
|
|
result = self._list(**filters)
|
|
if filters["page_no"] == 1:
|
|
for row in result["lists"]:
|
|
row["id"] = -row["id"]
|
|
row["diagnosis_id"] = row["id"]
|
|
return result
|
|
|
|
repository = Repository()
|
|
dialog = ai_consult_picker.AiConsultTargetDialog(repository, PermissionSet(["*"]))
|
|
dialog.show()
|
|
application.processEvents()
|
|
try:
|
|
QTest.qWait(100)
|
|
assert [call["page_no"] for call in repository.calls] == [1, 2]
|
|
assert dialog.table.rowCount() == 20
|
|
assert dialog.table.item(0, 3).text() == "21"
|
|
assert dialog.table.isVisible()
|
|
assert not dialog.empty_state.isVisible()
|
|
finally:
|
|
dialog.close()
|
|
|
|
|
|
def test_diagnosis_readonly_orders_scroll_and_switching_patient_resets_scope(
|
|
application: QApplication, immediate_async: None
|
|
) -> None:
|
|
repository = ListRepository()
|
|
dialog = diagnosis.DiagnosisDialog(
|
|
repository, permissions=PermissionSet(["tcm.diagnosis/patientOrders"])
|
|
)
|
|
dialog.open_for(
|
|
501, authoritative_detail={"id": 501, "patient_id": 301, "patient_name": "患者一"}
|
|
)
|
|
application.processEvents()
|
|
try:
|
|
table = dialog._table_registry["orders"][0]
|
|
dialog.readonly_scroll.ensureWidgetVisible(table)
|
|
assert table.rowCount() == 10
|
|
assert dialog.orders_list.parentWidget() is dialog._readonly_sections["orders"]
|
|
table.selectRow(2)
|
|
scroll_to_bottom(table)
|
|
assert table.rowCount() == 20
|
|
assert table.currentRow() == 2
|
|
assert table.item(2, 0).data(Qt.ItemDataRole.CheckStateRole) is None
|
|
assert repository.calls[-1] == {
|
|
"page_no": 2,
|
|
"page_size": 10,
|
|
"patient_id": 301,
|
|
"context_diagnosis_id": 501,
|
|
"scene": "diagnosis_edit",
|
|
}
|
|
dialog.open_for(
|
|
502, authoritative_detail={"id": 502, "patient_id": 302, "patient_name": "患者二"}
|
|
)
|
|
assert table.rowCount() == 10
|
|
assert repository.calls[-1]["page_no"] == 1
|
|
assert repository.calls[-1]["patient_id"] == 302
|
|
assert repository.calls[-1]["context_diagnosis_id"] == 502
|
|
finally:
|
|
dialog.close()
|