"""Interaction contracts at risk when the patient list moves to the blue layout.""" from __future__ import annotations import os import socket from copy import deepcopy from typing import Any os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") import pytest from PySide6.QtCore import QDate, Qt, QTimer from PySide6.QtTest import QTest from PySide6.QtWidgets import QApplication, QCheckBox, QLabel from doctor_workstation.core import PermissionSet from doctor_workstation.ui.pages import patients as patients_module from doctor_workstation.ui.pages.patients import PatientListWorkspace from doctor_workstation.ui.theme import apply_theme class _Repository: def __init__(self) -> None: self.queries: list[dict[str, Any]] = [] self.rows = [ { "id": 601 + index, "diagnosis_id": 601 + index, "source_patient_id": 301 + index, "appointment_id": 401 + index if index != 2 else 0, "patient_name": name, "gender": 2, "age": 36, "assistant_name": "测试医助", "assistant_id": 81 if index != 2 else 0, "appointment_doctor_name": "测试医生" if index != 2 else "", "appointment_status": (1, 4, 0)[index], "appointment_status_text": "未预约" if index == 2 else "待接诊", "appointment_time_text": "2026-09-05 09:00" if index != 2 else "—", "has_id_card": index != 1, "revisit_count": index, "confirmation_text": "待确认", "diagnosis_date_text": "初诊", "phone_masked": f"138****120{index}", } for index, name in enumerate(("阿青", "林青", "赵青")) ] def list_patients(self, **query: Any) -> dict[str, Any]: self.queries.append(query) return { "lists": deepcopy(self.rows), "count": len(self.rows), "extend": { "scope": {"label": "测试部门可见患者"}, "summary": {"today": 2, "tomorrow": 1, "day_after": 0}, "dates": {"today": QDate.currentDate().toString("yyyy-MM-dd")}, }, } def _settle(application: QApplication) -> None: for _ in range(3): application.processEvents() @pytest.fixture(scope="module") def application() -> QApplication: application = QApplication.instance() or QApplication([]) apply_theme(application) return application @pytest.fixture def workspace_factory(application: QApplication, monkeypatch: pytest.MonkeyPatch): def immediate(function: Any, *, on_success=None, on_error=None, on_finished=None): try: result = function() except Exception as error: if on_error: on_error(error) raise else: if on_success: on_success(result) finally: if on_finished: on_finished() def reject_network(*_args: Any, **_kwargs: Any): pytest.fail("Patient visual tests must use only local fixture data") monkeypatch.setattr(socket.socket, "connect", reject_network) monkeypatch.setattr(socket.socket, "connect_ex", reject_network) monkeypatch.setattr(socket, "create_connection", reject_network) monkeypatch.setattr(patients_module, "run_async", immediate) opened = [] def create(*, permissions=("*",), width=1270, height=700): repository = _Repository() workspace = PatientListWorkspace(repository, PermissionSet(list(permissions))) opened.append(workspace) workspace.resize(width, height) workspace.show() workspace.refresh() _settle(application) return workspace, repository yield create for workspace in opened: for timer in workspace.findChildren(QTimer): timer.stop() workspace.close() workspace.deleteLater() _settle(application) def _click(widget, application: QApplication) -> None: QTest.mouseClick(widget, Qt.MouseButton.LeftButton) _settle(application) def _date_selection(workspace: PatientListWorkspace) -> list[str]: return [key for key, button in workspace.quick_buttons.items() if button.isChecked()] + ( ["custom"] if workspace.custom_date_button.isChecked() else [] ) def test_filters_preserve_independent_dimensions_and_reset_to_unlimited_dates( application: QApplication, workspace_factory ) -> None: workspace, repository = workspace_factory() assert repository.queries == [{ "keyword": "", "status": "", "start_date": "", "end_date": "", "page_no": 1, "page_size": 15, }] assert _date_selection(workspace) == ["all"] assert not workspace.start_date.isEnabled() assert not workspace.end_date.isEnabled() assert workspace.start_date.date() == workspace.end_date.date() == QDate.currentDate() assert workspace.scope_label.text() == "测试部门可见患者" assert workspace.summary_buttons["today"].text() == f"今日预约 · {QDate.currentDate().toString('yyyy-MM-dd')}\n2 人" assert workspace.summary_buttons["tomorrow"].text() == "明日预约\n1 人" assert workspace.summary_buttons["day_after"].text() == "后天预约\n0 人" workspace.keyword_edit.setText(" 林医生 ") workspace.pager.load_more() _click(workspace.status_buttons["pending_interview"], application) assert repository.queries[-1]["page_no"] == 1 for mode, offset in (("today", 0), ("tomorrow", 1), ("day_after", 2)): _click(workspace.summary_buttons[mode], application) expected = QDate.currentDate().addDays(offset).toString("yyyy-MM-dd") assert repository.queries[-1] == { "keyword": "林医生", "status": "pending_interview", "start_date": expected, "end_date": expected, "page_no": 1, "page_size": 15, } assert _date_selection(workspace) == [mode] assert workspace.status_buttons["pending_interview"].isChecked() assert workspace.summary_buttons["today"].text().endswith("2 人") _click(workspace.custom_date_button, application) assert _date_selection(workspace) == ["custom"] assert workspace.start_date.isEnabled() and workspace.end_date.isEnabled() # Clicking the selected custom mode keeps a visible selection. _click(workspace.custom_date_button, application) assert _date_selection(workspace) == ["custom"] workspace.start_date.setDate(QDate.currentDate().addDays(-4)) workspace.end_date.setDate(QDate.currentDate().addDays(-2)) workspace.end_date.editingFinished.emit() assert repository.queries[-1]["start_date"] == QDate.currentDate().addDays(-4).toString("yyyy-MM-dd") assert repository.queries[-1]["end_date"] == QDate.currentDate().addDays(-2).toString("yyyy-MM-dd") assert _date_selection(workspace) == ["custom"] workspace.start_date.setDate(QDate.currentDate().addDays(3)) before = len(repository.queries) workspace.start_date.editingFinished.emit() _click(workspace.search_button, application) assert len(repository.queries) == before assert workspace.banner.isVisible() assert "开始日期不能晚于结束日期" in " ".join(label.text() for label in workspace.banner.findChildren(QLabel)) _click(workspace.reset_button, application) assert repository.queries[-1] == { "keyword": "", "status": "", "start_date": "", "end_date": "", "page_no": 1, "page_size": 15, } assert workspace.keyword_edit.text() == "" assert [key for key, button in workspace.status_buttons.items() if button.isChecked()] == [""] assert _date_selection(workspace) == ["all"] assert not workspace.start_date.isEnabled() and not workspace.end_date.isEnabled() assert not workspace.banner.isVisible() @pytest.mark.parametrize(("permissions", "expected", "editable"), [ (("*",), ["诊单", "AI 分析"], True), (("tcm.diagnosis/readonlyDetail",), ["查看"], False), ((), [], None), ]) def test_row_actions_obey_edit_readonly_and_no_permission( application: QApplication, workspace_factory, permissions, expected, editable ) -> None: workspace, _repository = workspace_factory(permissions=permissions) opened = [] workspace.diagnosis_requested.connect(lambda row, can_edit: opened.append((row["diagnosis_id"], can_edit))) for index in range(workspace.table.rowCount()): row = workspace.table.item(index, 0).data(Qt.ItemDataRole.UserRole) actions = workspace.table.cellWidget(index, 9) assert actions.visible_labels == expected if permissions == ("*",): overflow = ["预约", "重新指派" if row["assistant_id"] else "指派医助"] if not row["has_id_card"]: overflow.append("补全身份证") overflow.append("关联订单") if row["appointment_id"] and row["appointment_status"] in (1, 4): overflow.append("取消") assert actions.overflow_labels == overflow assert all(entry.property("danger") is True for entry in actions.menu.actions() if entry.text() == "取消") else: assert actions.more_button is None assert actions.overflow_labels == [] if editable is not None: _click(actions.buttons[0], application) assert opened[-1] == (row["diagnosis_id"], editable) if editable is None: workspace.table.selectRow(0) workspace._open_selected_diagnosis() assert opened == [] assert workspace.bottom_actions.isHidden() def _selector(workspace: PatientListWorkspace, row: int) -> QCheckBox: return workspace.table.cellWidget(row, 0).findChild(QCheckBox) def _assert_single_selection(workspace: PatientListWorkspace) -> None: checked = [row for row in range(workspace.table.rowCount()) if _selector(workspace, row).isChecked()] assert len(checked) <= 1 if checked: assert checked == [workspace.table.currentRow()] def test_sorted_selectors_and_row_actions_keep_distinct_resource_ids( application: QApplication, workspace_factory ) -> None: workspace, _repository = workspace_factory() table = workspace.table first_id = table.item(0, 0).data(Qt.ItemDataRole.UserRole)["diagnosis_id"] table.sortItems(1, Qt.SortOrder.DescendingOrder) _settle(application) assert table.item(0, 0).data(Qt.ItemDataRole.UserRole)["diagnosis_id"] != first_id for index in (0, 1): expected = table.item(index, 0).data(Qt.ItemDataRole.UserRole) _click(_selector(workspace, index), application) assert table.current_data()["diagnosis_id"] == expected["diagnosis_id"] assert _selector(workspace, index).isChecked() _assert_single_selection(workspace) # Leave a different row current: every callback must use its own sorted row. table.selectRow(2) _settle(application) _assert_single_selection(workspace) table.horizontalScrollBar().setValue(table.horizontalScrollBar().maximum()) emitted = [] workspace.diagnosis_requested.connect(lambda row, editable: emitted.append(("diagnosis", row, editable))) workspace.ai_consult_requested.connect(lambda row: emitted.append(("ai", row, None))) workspace.appointment_requested.connect(lambda row: emitted.append(("appointment", row, None))) for index in range(table.rowCount()): expected = table.item(index, 0).data(Qt.ItemDataRole.UserRole) actions = table.cellWidget(index, 9) actions.buttons[0].click() actions.buttons[1].click() next(action for action in actions.menu.actions() if action.text() == "预约").trigger() for kind, row, editable in emitted[-3:]: assert row["diagnosis_id"] == expected["diagnosis_id"] assert row["source_patient_id"] == expected["source_patient_id"] assert row["appointment_id"] == expected["appointment_id"] if kind == "diagnosis": assert editable is True @pytest.mark.parametrize("silent", [False, True]) def test_loading_resets_changed_query_and_ignores_stale_success_and_error( application: QApplication, workspace_factory, monkeypatch: pytest.MonkeyPatch, silent: bool ) -> None: workspace, repository = workspace_factory() table = workspace.table queued = [] monkeypatch.setattr(patients_module, "run_async", lambda function, **callbacks: queued.append((function, callbacks))) workspace.keyword_edit.setText("旧请求") workspace.refresh(silent=silent) workspace.keyword_edit.setText("最新请求") workspace.refresh(silent=silent) _settle(application) assert table.rowCount() == 0 assert workspace.content_stack.currentIndex() == 1 assert workspace.pager.isVisible() assert not workspace.banner.isVisible() # Workers must keep the UI-thread snapshot even after text changes again. workspace.keyword_edit.setText("尚未查询") newer = queued[1][0]() assert repository.queries[-1]["keyword"] == "最新请求" newer["lists"][0]["patient_name"] = "最新结果" newer["extend"]["scope"]["label"] = "最新范围" queued[1][1]["on_success"](newer) current_item = table.item(0, 1) assert current_item.text().startswith("最新结果") stale = queued[0][0]() assert repository.queries[-1]["keyword"] == "旧请求" queued[0][1]["on_success"](stale) queued[0][1]["on_error"](RuntimeError("过期失败")) _settle(application) assert table.item(0, 1) is current_item assert workspace.scope_label.text() == "最新范围" assert not workspace.banner.isVisible() workspace.keyword_edit.setText("最新请求") workspace.refresh(silent=silent) queued[-1][1]["on_error"](RuntimeError("患者查询失败")) _settle(application) assert table.item(0, 1) is current_item assert workspace.banner.isVisible() assert "患者查询失败" in " ".join(label.text() for label in workspace.banner.findChildren(QLabel)) workspace.refresh(silent=silent) queued[-1][1]["on_success"](queued[-1][0]()) _settle(application) assert not workspace.banner.isVisible()