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, Signal from PySide6.QtWidgets import ( QApplication, QDialog, QDialogButtonBox, QInputDialog, QLabel, QMessageBox, QToolButton, QWidget, ) from doctor_workstation.core import PermissionSet from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module from doctor_workstation.ui.pages import consultations as consultations_module from doctor_workstation.ui.pages.consultations import ( ConsultationsPage, _AppointmentLogDialog, _DiagnosisCreateDialog, _DiagnosisOrderDialog, _DiagnosisOrderQrDialog, _QrResultDialog, _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 def open_view_only(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( 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(consultations_module, "run_async", run_immediately) monkeypatch.setattr(diagnosis_module, "run_async", run_immediately) def _row(**changes: Any) -> dict[str, Any]: row = { "id": 501, "diagnosis_id": 501, "patient_id": 301, "patient_name": "林晓岚", "gender": 2, "age": 36, "status": 4, "has_appointment": 1, "appointment_id": 101, "appointment_status": 1, "appointments": [ { "id": 101, "status": 1, "doctor_name": "陈医生", "appointment_date": "2026-08-10", "time_text": "09:00", } ], "DiagnosisViewRecord": [{"is_confirmed": 1}], "has_prescription": 0, } row.update(changes) return row def test_consultation_business_dialogs_expose_shared_visual_contract( application: QApplication, ) -> None: record = _row(patient_name="鹿立核", patient_id=301) dialogs = ( _DiagnosisCreateDialog(), _DiagnosisOrderDialog(record), _AppointmentLogDialog(record, []), _QrResultDialog("视频二维码", record, "https://example.invalid/video.png"), _DiagnosisOrderQrDialog(record, "ORDER-301"), ) assert [dialog.objectName() for dialog in dialogs] == [ "DiagnosisCreateDialog", "DiagnosisOrderDialog", "AppointmentLogDialog", "DiagnosisQrResultDialog", "DiagnosisOrderQrDialog", ] 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.Save ).property("variant") == "primary" assert dialogs[1].findChild(QDialogButtonBox).button( QDialogButtonBox.StandardButton.Ok ).property("variant") == "primary" assert next( button for button in dialogs[3].findChild(QDialogButtonBox).buttons() if button.text() == "打开二维码" ).property("variant") == "primary" assert dialogs[4].open_button.property("variant") == "primary" for dialog in dialogs: dialog.close() application.processEvents() def test_video_condition_never_uses_diagnosis_status_or_missed_status() -> None: assert is_video_available(_row(status=4, appointment_status=1)) assert not is_video_available(_row(status=1, appointment_status=4)) assert not is_video_available(_row(status=1, appointment_status=1, has_appointment=0)) payload = _video_payload(_row(id=777, diagnosis_id=777, appointment_id=222)) assert payload["appointment_id"] == 222 assert payload["diagnosis_id"] == 777 assert payload["patient_id"] == 301 def test_nested_appointments_confirmation_and_prescription_labels() -> None: row = _row( appointment_id=None, appointment_status=None, DiagnosisViewRecord=[{"is_confirmed": 0}, {"is_confirmed": "1"}], ) assert appointment_rows(row)[0]["id"] == 101 assert is_diagnosis_confirmed(row) assert ( prescription_action_label({"prescription_audit_status": 1, "prescription_void_status": 0}) == "查看处方" ) assert ( prescription_action_label({"prescription_audit_status": 1, "prescription_void_status": 1}) == "编辑处方" ) assert ( prescription_action_label( { "has_prescription": 1, "prescription_audit_status": 0, "prescription_void_status": 0, } ) == "编辑处方" ) assert ( prescription_action_label( { "has_prescription": 1, "current_has_prescription": 0, "prescription_audit_status": 1, "prescription_void_status": 0, } ) == "开方" ) def test_view_prescription_intent_is_readonly_and_fails_closed_on_state_change( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: row = _row( has_prescription=1, current_has_prescription=1, current_prescription_id=701, prescription_audit_status=1, prescription_void_status=0, ) page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"])) page.table_host.set_rows([row]) page.table.selectRow(0) requested: list[tuple[Any, str]] = [] monkeypatch.setattr( page, "_begin_prescription_load", lambda record, *, mode: requested.append((record, mode)), ) page._open_prescription() assert requested == [(row, "view")] monkeypatch.setattr( page, "_open_existing_prescription_editor", lambda _existing: pytest.fail("view intent must never open the editor"), ) page._prescription_loaded( { "id": 701, "appointment_id": 101, "audit_status": 0, "void_status": 0, }, row, "view", page._prescription_generation, ) assert "状态已变化" in page.banner.label.text() page.close() application.processEvents() def test_default_query_matches_admin_today_and_page_size_contract( application: QApplication, immediate_async: None, ) -> None: calls: list[dict[str, Any]] = [] class Repository: def list_consultations(self, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs) return {"lists": [], "count": 0} page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) page.refresh(silent=True) assert len(calls) == 1 query = calls[0] assert query["page_no"] == 1 assert query["page_size"] == 15 assert query["appointment_date"] == QDate.currentDate().toString("yyyy-MM-dd") assert query["appointment_status"] == "1" assert "status" not in query assert query["has_appointment"] == "" assert query["diagnosis_confirmed"] == "" assert { "diagnosis_type", "syndrome_type", "assistant_id", "latest_appointment_start_date", "latest_appointment_end_date", "latest_appointment_channel_source", "latest_assign_start_date", "latest_assign_end_date", "sort_unserved_days", }.issubset(query) assert "consultation_type" not in query page.close() application.processEvents() def test_double_click_opens_readonly_and_never_emits_video( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: page = ConsultationsPage( SimpleNamespace(), permissions=PermissionSet(["tcm.diagnosis/readonlyDetail", "tcm.diagnosis/videoQr"]), ) page.table.set_rows([_row()]) page.table.selectRow(0) opened: list[int] = [] videos: list[dict[str, Any]] = [] monkeypatch.setattr( page._diagnosis_dialog, "open_view_only", lambda diagnosis_id, *, seed=None: opened.append(diagnosis_id), ) page.video_requested.connect(videos.append) page.table.itemDoubleClicked.emit(page.table.item(0, 0)) application.processEvents() assert opened == [501] assert videos == [] page.close() def test_action_visibility_requires_exact_canonical_permissions( application: QApplication, ) -> None: aliases = PermissionSet( [ "tcm.diagnosis.readonlyDetail", "tcm.diagnosis.edit", "tcm.diagnosis.add", "tcm.diagnosis.delete", ] ) page = ConsultationsPage(SimpleNamespace(), permissions=aliases) assert not page.view_button.isVisible() assert not page.edit_button.isVisible() assert not page.add_button.isVisible() assert not page.delete_button.isVisible() page.close() exact = PermissionSet( [ "tcm.diagnosis/readonlyDetail", "tcm.diagnosis/edit", "tcm.diagnosis/add", "tcm.diagnosis/delete", ] ) page = ConsultationsPage(SimpleNamespace(), permissions=exact) assert not page.view_button.isHidden() assert not page.edit_button.isHidden() assert not page.add_button.isHidden() assert not page.delete_button.isHidden() page.close() application.processEvents() def test_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(consultations_module, "run_async", queue_async) page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"])) page.refresh(silent=True) page.refresh(silent=True) callbacks[1]["on_success"]({"lists": [_row(id=902, diagnosis_id=902)], "count": 1}) callbacks[0]["on_success"]({"lists": [_row(id=901, diagnosis_id=901)], "count": 1}) application.processEvents() assert page.table.rowCount() == 1 assert page.table.item(0, 0).text().startswith("902") page.close() def test_identical_silent_refresh_has_zero_model_reset_and_fixed_widget_budget( application: QApplication, immediate_async: None, monkeypatch: pytest.MonkeyPatch, ) -> None: class Repository: calls = 0 def list_consultations(self, **_kwargs: Any) -> dict[str, Any]: self.calls += 1 return {"lists": [_row()], "count": 1} repository = Repository() page = ConsultationsPage(repository, permissions=PermissionSet(["*"])) page.refresh(silent=True) page.table.selectRow(0) model = page.table_host.model action_widget = page.table_host.fixed.indexWidget(model.index(0, 11)) video_widget = page.table_host.fixed.indexWidget(model.index(0, 10)) resets: list[str] = [] model.modelAboutToBeReset.connect(lambda: resets.append("begin")) model.modelReset.connect(lambda: resets.append("end")) install_calls: list[None] = [] original_install = page.table_host._install_fixed_widgets def count_install() -> None: install_calls.append(None) original_install() monkeypatch.setattr(page.table_host, "_install_fixed_widgets", count_install) page.refresh(silent=True) assert repository.calls == 2 assert resets == [] assert install_calls == [] assert page.table_host.fixed.indexWidget(model.index(0, 11)) is action_widget assert page.table_host.fixed.indexWidget(model.index(0, 10)) is video_widget assert page.table.currentIndex().row() == 0 page.close() application.processEvents() def test_timer_poll_has_one_request_budget_while_refresh_is_in_flight( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: jobs: list[dict[str, Any]] = [] def queue_async(_function: Any, **options: Any) -> object: jobs.append(options) return object() monkeypatch.setattr(consultations_module, "run_async", queue_async) page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"])) monkeypatch.setattr(page, "isVisible", lambda: True) monkeypatch.setattr(page, "_refresh_counts", lambda: None) page._poll_refresh() page._poll_refresh() page._poll_refresh() assert len(jobs) == 1 assert page._loading jobs[0]["on_success"]({"lists": [_row()], "count": 1}) jobs[0]["on_finished"]() assert not page._loading page.close() application.processEvents() def test_current_appointment_is_the_only_prescription_authority( application: QApplication, ) -> None: calls: list[tuple[str, int]] = [] class Repository: def get_prescription_by_appointment(self, appointment_id: int) -> None: calls.append(("appointment", appointment_id)) return None def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]: raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}") page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) assert page._load_context_prescription(_row(appointment_id=202)) is None assert calls == [("appointment", 202)] page.close() application.processEvents() def test_empty_appointment_wrapper_is_treated_as_a_new_prescription( application: QApplication, ) -> None: class Repository: def get_prescription_by_appointment(self, appointment_id: int) -> dict[str, Any]: assert appointment_id == 202 return {"data": {}} page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) assert page._load_context_prescription(_row(appointment_id=202)) is None page.close() application.processEvents() def test_prescription_query_error_is_fail_closed_without_diagnosis_fallback( application: QApplication, ) -> None: class Repository: def get_prescription_by_appointment(self, appointment_id: int) -> None: raise RuntimeError(f"appointment {appointment_id} unavailable") def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]: raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}") page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) with pytest.raises(RuntimeError, match="appointment 202 unavailable"): page._load_context_prescription(_row(appointment_id=202)) page.close() application.processEvents() def test_new_prescription_uses_authoritative_case_snapshot_and_exact_ids( application: QApplication, immediate_async: None, monkeypatch: pytest.MonkeyPatch, ) -> None: calls: list[tuple[str, int]] = [] created: list[dict[str, Any]] = [] dialog_seeds: list[dict[str, Any]] = [] case_record = { "diagnosis": { "id": 501, "patient_name": "林晓岚", "chief_complaint": "咳嗽", "tongue": "面象哨兵", "tongue_image": "舌象哨兵", "pulse": "脉象哨兵", "pulse_condition": "脉象详情哨兵", }, "patient": {"id": 301, "gender": 2, "age": 36}, } class Repository: def get_prescription_by_appointment(self, appointment_id: int) -> None: calls.append(("appointment", appointment_id)) return None def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]: raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}") def get_diagnosis_detail(self, diagnosis_id: int, **_kwargs: Any) -> dict[str, Any]: calls.append(("diagnosis_detail", diagnosis_id)) return case_record def create_prescription(self, prescription: Any) -> dict[str, Any]: created.append(dict(prescription)) return {"id": 901} def list_consultations(self, **_kwargs: Any) -> dict[str, Any]: return {"lists": [], "count": 0} class AcceptedEditor: def __init__( self, _repository: Any, seed: dict[str, Any], **_kwargs: Any, ) -> None: dialog_seeds.append(seed) def exec(self) -> QDialog.DialogCode: return QDialog.DialogCode.Accepted def payload(self) -> dict[str, Any]: return {"formula_name": "止咳方", "medicines": []} monkeypatch.setattr(consultations_module, "PrescriptionEditorDialog", AcceptedEditor) page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) page._begin_prescription_load(_row(appointment_id=202), mode="open") assert calls[:2] == [("appointment", 202), ("diagnosis_detail", 501)] assert len(created) == 1 assert created[0]["diagnosis_id"] == 501 assert created[0]["appointment_id"] == 202 assert created[0]["case_record"] == case_record assert created[0]["case_record"] is not case_record assert dialog_seeds[0]["case_record"] == case_record assert dialog_seeds[0]["case_record"] is not case_record assert dialog_seeds[0]["tongue"] == "面象哨兵" assert dialog_seeds[0]["tongue_image"] == "舌象哨兵" assert dialog_seeds[0]["pulse"] == "脉象哨兵" assert dialog_seeds[0]["pulse_condition"] == "脉象详情哨兵" page.close() application.processEvents() def test_existing_pending_prescription_is_edited_in_place_not_duplicated( application: QApplication, immediate_async: None, monkeypatch: pytest.MonkeyPatch, ) -> None: updated: list[tuple[int, dict[str, Any]]] = [] created: list[dict[str, Any]] = [] existing = { "id": 901, "diagnosis_id": 501, "patient_name": "林晓岚", "audit_status": 0, "void_status": 0, "clinical_diagnosis": "气虚", "herbs": [{"medicine_id": 31, "name": "黄芪", "dosage": 15}], } class Repository: def get_prescription_by_appointment(self, appointment_id: int) -> dict[str, Any]: assert appointment_id == 202 return existing def update_prescription( self, prescription: int, changes: dict[str, Any] ) -> dict[str, Any]: updated.append((prescription, dict(changes))) return {"id": prescription} def create_prescription(self, prescription: dict[str, Any]) -> dict[str, Any]: created.append(dict(prescription)) return {"id": 999} def list_consultations(self, **_kwargs: Any) -> dict[str, Any]: return {"lists": [], "count": 0} class AcceptedEditor: def __init__(self, _repository: Any, source: Any, **kwargs: Any) -> None: assert source is existing assert kwargs["mode"] == "edit" def exec(self) -> QDialog.DialogCode: return QDialog.DialogCode.Accepted def payload(self) -> dict[str, Any]: return {"id": 901, "clinical_diagnosis": "气血两虚"} monkeypatch.setattr(consultations_module, "PrescriptionEditorDialog", AcceptedEditor) page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"])) page._begin_prescription_load(_row(appointment_id=202), mode="open") assert updated == [(901, {"id": 901, "clinical_diagnosis": "气血两虚"})] assert created == [] page.close() application.processEvents() def test_switching_rows_invalidates_prescription_worker_and_clears_busy( application: QApplication, immediate_async: None, monkeypatch: pytest.MonkeyPatch, ) -> None: page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"])) callbacks: list[dict[str, Any]] = [] def queue_async(_function: Any, **options: Any) -> object: callbacks.append(options) return object() monkeypatch.setattr(consultations_module, "run_async", queue_async) page.table.set_rows( [ _row(id=501, diagnosis_id=501, appointment_id=101), _row(id=502, diagnosis_id=502, appointment_id=202), ] ) page.table.selectRow(0) page._begin_prescription_load(page.table.current_data(), mode="open") assert page._prescription_busy assert not page.prescription_button.isEnabled() page.table.selectRow(1) application.processEvents() assert not page._prescription_busy assert page.prescription_button.isEnabled() callbacks[0]["on_success"]({"id": 88, "appointment_id": 101}) callbacks[0]["on_finished"]() assert page.table.current_data()["appointment_id"] == 202 assert not page._prescription_busy page.close() application.processEvents() def test_native_call_does_not_reuse_video_qr_permission( application: QApplication, ) -> None: 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 live_row = _row( video_call_hint={ "state": "live", "label": "视频通话进行中", "start_time": 1787102100, } ) page = ConsultationsPage(VideoRepository(), permissions=PermissionSet([])) emitted: list[dict[str, Any]] = [] page.video_requested.connect(emitted.append) page.table.set_rows([live_row]) page.table.selectRow(0) application.processEvents() 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 emitted == [] assert "摄像头和麦克风" in video_action.toolTip() page._request_video() assert emitted == [_video_payload(live_row)] assert emitted[0]["mode"] == "im" page.close() application.processEvents() @pytest.mark.parametrize( ("hint", "status_text"), [ ({"state": "none", "label": ""}, "暂无通话"), ({"state": "pending_room", "label": "通话发起中，待同步房间"}, "等待接通"), ], ) def test_video_join_action_is_hidden_until_doctor_session_is_live( application: QApplication, hint: dict[str, Any], status_text: str, ) -> None: 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(video_call_hint=hint)]) page.table.selectRow(0) application.processEvents() video_cell = page.table_host.fixed.indexWidget(page.table_host.model.index(0, 10)) assert video_cell.findChildren(QToolButton) == [] assert status_text in " ".join(label.text() for label in video_cell.findChildren(QLabel)) page._request_video() assert emitted == [] 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, *, diagnosis_id: int, ) -> dict[str, str]: calls.append( ( "video_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/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 = { "diagnosis_id": 501, "patient_id": 301, "doctor_id": 77, "share_user_id": 66, } assert calls == [ ("video_qr", qr_payload), ("confirm_qr", qr_payload), ("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, *, diagnosis_id: int, ) -> dict[str, str]: del diagnosis_id, 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()