243 lines
8.9 KiB
Python
243 lines
8.9 KiB
Python
"""Scrolling, query isolation and refresh contracts for the two clinic queues."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from copy import deepcopy
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import Qt, Signal
|
|
from PySide6.QtTest import QTest
|
|
from PySide6.QtWidgets import QApplication, QComboBox, QSpinBox, QWidget
|
|
|
|
from doctor_workstation.ui.infinite_list import InfiniteList
|
|
from doctor_workstation.ui.pages import appointments, consultations
|
|
|
|
|
|
class _DiagnosisDialog(QWidget):
|
|
saved = Signal()
|
|
|
|
def __init__(self, _repository: Any, parent: QWidget | None = None) -> None:
|
|
super().__init__(parent)
|
|
|
|
|
|
class _Repository:
|
|
def __init__(self) -> None:
|
|
self.calls: list[dict[str, Any]] = []
|
|
self.fail_page: int | None = None
|
|
self.revision = 0
|
|
|
|
def _list(self, **query: Any) -> dict[str, Any]:
|
|
self.calls.append(dict(query))
|
|
page = query["page_no"]
|
|
if page == self.fail_page:
|
|
raise RuntimeError("暂时无法加载")
|
|
second = bool(query.get("keyword") or query.get("patient_name"))
|
|
offset = 1000 if second else 0
|
|
total = 4 if second else 34
|
|
start = (page - 1) * query["page_size"]
|
|
rows = [
|
|
{
|
|
"id": offset + index,
|
|
"diagnosis_id": offset + index,
|
|
"source_patient_id": index + 2000,
|
|
"patient_id": index + 2000,
|
|
"patient_name": f"患者{offset + index} · {self.revision}",
|
|
"status": 1,
|
|
"status_desc": "待接诊",
|
|
"appointment_id": offset + index,
|
|
"appointment_status": 1,
|
|
"appointment_date": date.today().isoformat(),
|
|
"appointment_time": "09:00-09:30",
|
|
"patient_phone": "13800001234",
|
|
"doctor_name": "测试医生",
|
|
"doctor_id": 30,
|
|
"diagnosis_confirmed": True,
|
|
"appointments": [],
|
|
}
|
|
for index in range(start + 1, min(total, start + query["page_size"]) + 1)
|
|
]
|
|
result = {"lists": rows, "count": total}
|
|
if page == 1:
|
|
result["extend"] = {
|
|
"status_count": {"1": total, "3": 7},
|
|
"date_counts": {"today": total, "tomorrow": 9},
|
|
}
|
|
return result
|
|
|
|
list_appointments = _list
|
|
list_consultations = _list
|
|
|
|
|
|
def _inline(function: Any, *, on_success=None, on_error=None, on_finished=None) -> None:
|
|
try:
|
|
result = function()
|
|
except Exception as error:
|
|
if on_error is not None:
|
|
on_error(error)
|
|
else:
|
|
if on_success is not None:
|
|
on_success(result)
|
|
finally:
|
|
if on_finished is not None:
|
|
on_finished()
|
|
|
|
|
|
def _settle(application: QApplication) -> None:
|
|
for _ in range(3):
|
|
application.processEvents()
|
|
QTest.qWait(35)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture(params=["appointments", "consultations"])
|
|
def queue_page(request: Any, application: QApplication, monkeypatch: pytest.MonkeyPatch):
|
|
module = appointments if request.param == "appointments" else consultations
|
|
monkeypatch.setattr(module, "run_async", _inline)
|
|
monkeypatch.setattr(consultations, "DiagnosisDialog", _DiagnosisDialog)
|
|
monkeypatch.setattr(consultations.ConsultationsPage, "_load_filter_options", lambda self: None)
|
|
monkeypatch.setattr(consultations.ConsultationsPage, "_refresh_counts", lambda self: None)
|
|
monkeypatch.setattr(appointments.AppointmentsPage, "_load_departments", lambda self: None)
|
|
repository = _Repository()
|
|
page_class = appointments.AppointmentsPage if module is appointments else consultations.ConsultationsPage
|
|
page = page_class(repository, permissions={"*"}, current_user={"role_id": 1})
|
|
page.resize(1280, 800)
|
|
page.show()
|
|
page.poll_timer.stop()
|
|
_settle(application)
|
|
yield page, repository, module
|
|
page.close()
|
|
page.deleteLater()
|
|
_settle(application)
|
|
|
|
|
|
def _scroll_bottom(page: Any, application: QApplication) -> None:
|
|
scrollbar = page.table.verticalScrollBar()
|
|
assert scrollbar.maximum() > 0
|
|
scrollbar.setValue(scrollbar.maximum())
|
|
_settle(application)
|
|
|
|
|
|
def test_scroll_appends_and_preserves_selection(queue_page: Any, application: QApplication) -> None:
|
|
page, repository, _module = queue_page
|
|
assert page.table.rowCount() == 15
|
|
page.table.selectRow(6)
|
|
if hasattr(page, "table_host"):
|
|
model = page.table_host.model
|
|
model.setData(model.index(6, 0), Qt.CheckState.Checked, Qt.ItemDataRole.CheckStateRole)
|
|
_scroll_bottom(page, application)
|
|
|
|
assert page.table.rowCount() == 30
|
|
assert [call["page_no"] for call in repository.calls] == [1, 2]
|
|
assert page.table.current_data()["id"] == 7
|
|
assert page.table.verticalScrollBar().value() > 0
|
|
if hasattr(page, "table_host"):
|
|
assert [row["id"] for row in page.table_host.selected_records()] == [7]
|
|
else:
|
|
assert page._status_counts[3] == 7
|
|
assert page.date_buttons["tomorrow"].text().endswith(" 9")
|
|
|
|
_scroll_bottom(page, application)
|
|
assert page.table.rowCount() == 34
|
|
assert len({row["id"] for row in page.pager.rows}) == 34
|
|
assert not page.pager.has_more
|
|
assert "已全部加载" in page.pager.summary_label.text()
|
|
_scroll_bottom(page, application)
|
|
assert [call["page_no"] for call in repository.calls] == [1, 2, 3]
|
|
|
|
|
|
def test_silent_refresh_keeps_the_loaded_prefix(queue_page: Any, application: QApplication) -> None:
|
|
page, repository, _module = queue_page
|
|
_scroll_bottom(page, application)
|
|
page.table.selectRow(19)
|
|
old_scroll = page.table.verticalScrollBar().value()
|
|
repository.calls.clear()
|
|
repository.revision = 2
|
|
page.refresh(silent=True)
|
|
_settle(application)
|
|
|
|
assert [call["page_no"] for call in repository.calls] == [1, 2]
|
|
assert page.table.rowCount() == 30
|
|
assert page.table.current_data()["id"] == 20
|
|
assert page.table.current_data()["patient_name"].endswith(" · 2")
|
|
assert page.table.verticalScrollBar().value() == old_scroll
|
|
if isinstance(page, appointments.AppointmentsPage):
|
|
assert page._status_counts[3] == 7
|
|
assert page.date_buttons["tomorrow"].text().endswith(" 9")
|
|
|
|
|
|
def test_filter_change_supersedes_pending_append(
|
|
queue_page: Any, application: QApplication, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
page, _repository, module = queue_page
|
|
jobs: list[tuple[Any, dict[str, Any]]] = []
|
|
|
|
def deferred(function: Any, **callbacks: Any) -> None:
|
|
jobs.append((function, callbacks))
|
|
|
|
monkeypatch.setattr(module, "run_async", deferred)
|
|
# Reconfigure the shared controller to use the deferred runner, then finish
|
|
# that refresh before simulating a slow next page.
|
|
page.refresh(silent=True)
|
|
function, callbacks = jobs.pop()
|
|
callbacks["on_success"](function())
|
|
page.pager.load_more()
|
|
append_function, append_callbacks = jobs.pop()
|
|
stale_result = deepcopy(append_function())
|
|
|
|
search = page.patient_input if module is appointments else page.keyword_edit
|
|
search.setText("第二组")
|
|
page.refresh(silent=True)
|
|
assert len(jobs) == 1
|
|
function, callbacks = jobs.pop()
|
|
assert function()["lists"][0]["id"] == 1001
|
|
callbacks["on_success"](function())
|
|
append_callbacks["on_success"](stale_result)
|
|
_settle(application)
|
|
|
|
assert page.table.rowCount() == 4
|
|
assert [row["id"] for row in page.pager.rows] == [1001, 1002, 1003, 1004]
|
|
assert page.pager.page == 1
|
|
assert not page.pager.loading
|
|
assert page.table.verticalScrollBar().value() == 0
|
|
|
|
|
|
def test_failed_append_retries_without_losing_rows(queue_page: Any, application: QApplication) -> None:
|
|
page, repository, _module = queue_page
|
|
repository.fail_page = 2
|
|
_scroll_bottom(page, application)
|
|
assert page.table.rowCount() == 15
|
|
assert page.pager.page == 1
|
|
assert page.pager.retry_button.isVisible()
|
|
calls = len(repository.calls)
|
|
_settle(application)
|
|
assert len(repository.calls) == calls
|
|
|
|
repository.fail_page = None
|
|
page.pager.retry_button.click()
|
|
_settle(application)
|
|
assert page.table.rowCount() == 30
|
|
assert page.pager.page == 2
|
|
assert page.pager.retry_button.isHidden()
|
|
|
|
|
|
def test_list_footer_is_compact_without_page_controls(queue_page: Any, application: QApplication) -> None:
|
|
page, _repository, _module = queue_page
|
|
for height in (768, 960):
|
|
page.resize(1280, height)
|
|
_settle(application)
|
|
assert isinstance(page.pager, InfiniteList)
|
|
assert page.pager.height() == 24
|
|
assert not page.pager.findChildren(QComboBox)
|
|
assert not page.pager.findChildren(QSpinBox)
|
|
content = page.page_scroll.widget() if hasattr(page, "page_scroll") else page
|
|
assert content.layout().contentsMargins().bottom() == 8
|