"""Parity contracts for the admin appointment list port.""" 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.QtWidgets import QApplication, QDialog, QDialogButtonBox, QLabel from doctor_workstation.core.errors import ApiProtocolError from doctor_workstation.core.models import Appointment, PageResult from doctor_workstation.core.permissions import PermissionSet from doctor_workstation.services.mock_repository import DemoDoctorRepository from doctor_workstation.services.repository import RemoteDoctorRepository from doctor_workstation.ui.pages import appointments as appointments_module from doctor_workstation.ui.pages.appointments import ( AppointmentsPage, _diagnosis_id, _video_patient_id, prescription_action_label, ) from doctor_workstation.ui.shell import _match_navigation class RecordingClient: def __init__(self) -> None: self.get_calls: list[tuple[str, dict[str, Any]]] = [] self.post_calls: list[tuple[str, dict[str, Any]]] = [] def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any: self.get_calls.append((endpoint, dict(params or {}))) if endpoint == "doctor.appointment/detail": return {"id": int((params or {}).get("id", 0)), "patient_name": "测试"} if endpoint == "dept.dept/all": return [{"id": 10, "name": "中医门诊", "children": []}] return {"lists": [], "count": 0, "extend": {"status_count": {"1": 2}}} def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any: self.post_calls.append((endpoint, dict(payload or {}))) if endpoint == "tcm.diagnosis/generateMiniProgramQrcode": return {"qrcode_url": "https://example.test/uploads/video-qr.png"} return {"ok": True} @pytest.fixture(scope="module") def application() -> QApplication: return QApplication.instance() or QApplication([]) def test_remote_appointment_list_and_detail_hit_admin_endpoints() -> None: client = RecordingClient() repo = RemoteDoctorRepository(client) # type: ignore[arg-type] repo.list_appointments( page_no=1, page_size=15, status=1, start_date="2026-08-11", end_date="2026-08-11", include_status_counts=1, diagnosis_confirmed="1", assistant_dept_id=10, patient_name="林", ) detail = repo.get_appointment_detail(101) departments = repo.list_departments() assert client.get_calls[0][0] == "doctor.appointment/lists" assert client.get_calls[0][1]["include_status_counts"] == 1 assert client.get_calls[0][1]["assistant_dept_id"] == 10 assert "diag_scope_relax" not in client.get_calls[0][1] assert client.get_calls[1] == ("doctor.appointment/detail", {"id": 101}) assert client.get_calls[2] == ("dept.dept/all", {}) assert detail["id"] == 101 assert departments[0]["id"] == 10 def test_unlimited_date_relaxes_today_scope_for_pending_status() -> None: client = RecordingClient() repo = RemoteDoctorRepository(client) # type: ignore[arg-type] repo.list_appointments(status=1, diag_scope_relax=1, page_no=1, page_size=15) params = client.get_calls[0][1] assert "start_date" not in params assert "end_date" not in params assert "diag_scope_relax" not in params def test_video_qr_uses_doctor_id_without_fake_diagnosis_id() -> None: client = RecordingClient() repo = RemoteDoctorRepository(client) # type: ignore[arg-type] result = repo.generate_video_qrcode( diagnosis_id=501, doctor_id=88, patient_id=301, share_user_id=9, ) endpoint, payload = client.post_calls[-1] assert endpoint == "tcm.diagnosis/generateMiniProgramQrcode" assert payload == { "diagnosis_id": 501, "doctor_id": 88, "patient_id": 301, "share_user_id": 9, "mini_program_path": "pages/login/login", } assert result["qrcode_url"].endswith("video-qr.png") def test_legacy_empty_string_prescription_response_is_scoped_to_not_found() -> None: class LegacyClient(RecordingClient): def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any: if endpoint == "tcm.prescription/getByAppointment": raise ApiProtocolError("envelope must be an object", data="") return super().get(endpoint, params) repo = RemoteDoctorRepository(LegacyClient()) # type: ignore[arg-type] assert repo.get_prescription_by_appointment(101) is None class BrokenClient(LegacyClient): def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any: raise ApiProtocolError("invalid JSON") broken = RemoteDoctorRepository(BrokenClient()) # type: ignore[arg-type] with pytest.raises(ApiProtocolError, match="invalid JSON"): broken.get_prescription_by_appointment(101) def test_shell_matches_appointment_list_by_route_not_shared_permission() -> None: row = { "name": "挂号列表", "paths": "/appointments", "component": "tcm/appointment/list", "perms": "doctor.appointment/lists", } item = _match_navigation(row) assert item is not None assert item.key == "appointments" reception = { "name": "接诊台", "paths": "/reception", "component": "patient/reception/index", "perms": "doctor.appointment/lists", } assert _match_navigation(reception).key == "reception" # type: ignore[union-attr] def test_diagnosis_and_patient_ids_stay_distinct_for_video() -> None: row = Appointment.from_dict( { "id": 101, "patient_id": 301, "diagnosis_id": 501, "source_patient_id": 301, "status": 1, "has_prescription": 0, } ) assert _diagnosis_id(row) == 501 assert _video_patient_id(row) == 301 assert prescription_action_label(row) == "开方" approved = Appointment.from_dict( { "id": 102, "prescription_audit_status": 1, "prescription_void_status": 0, "has_prescription": 1, } ) assert prescription_action_label(approved) == "查看" pending = Appointment.from_dict( { "id": 103, "prescription_audit_status": 0, "prescription_void_status": 0, "has_prescription": 1, } ) assert prescription_action_label(pending) == "编辑处方" def test_appointment_pending_prescription_uses_full_edit_contract( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: updated: list[tuple[int, dict[str, Any]]] = [] existing = { "id": 81, "diagnosis_id": 501, "audit_status": 0, "void_status": 0, "patient_name": "林晓岚", } class Repository: def update_prescription( self, prescription: int, changes: dict[str, Any] ) -> dict[str, Any]: updated.append((prescription, dict(changes))) return {"id": prescription} 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": 81, "clinical_diagnosis": "脾气虚"} def run_immediately(function: Any, **callbacks: Any) -> object: try: result = function() except Exception as error: callbacks["on_error"](error) else: callbacks["on_success"](result) finally: callbacks["on_finished"]() return object() monkeypatch.setattr(appointments_module, "PrescriptionEditorDialog", AcceptedEditor) monkeypatch.setattr(appointments_module, "run_async", run_immediately) page = AppointmentsPage(Repository(), permissions=PermissionSet(["*"])) monkeypatch.setattr(page, "refresh", lambda **_kwargs: None) page._prescription_loaded(existing, {"id": 101}, page._prescription_generation) assert updated == [(81, {"id": 81, "clinical_diagnosis": "脾气虚"})] page.close() application.processEvents() def test_appointment_prescription_seed_keeps_admin_observation_fields( application: QApplication, ) -> None: page = AppointmentsPage(SimpleNamespace(), permissions=PermissionSet(["*"])) seed = page._prescription_seed( {"id": 101, "appointment_id": 101, "diagnosis_id": 501}, { "id": 501, "patient_name": "林晓岚", "tongue": "面象哨兵", "tongue_image": "舌象哨兵", "pulse": "脉象哨兵", "pulse_condition": "脉象详情哨兵", }, ) assert seed["tongue"] == "面象哨兵" assert seed["tongue_image"] == "舌象哨兵" assert seed["pulse"] == "脉象哨兵" assert seed["pulse_condition"] == "脉象详情哨兵" page.close() application.processEvents() def test_appointments_page_default_query_is_today_pending( application: QApplication, ) -> None: repo = DemoDoctorRepository() page = AppointmentsPage( repo, permissions=PermissionSet( [ "doctor.appointment/lists", "doctor.appointment/complete", "doctor.appointment/cancel", "doctor.appointment/prescription", "doctor.appointment/addDoctorNote", "tcm.diagnosis/edit", "tcm.diagnosis/kaifang", "tcm.diagnosis/videoQr", ] ), current_user={"id": 1001, "name": "陈医生", "role_id": 1}, ) page.show() application.processEvents() page.refresh() application.processEvents() filters = page._query_filters() assert filters["status"] == 1 assert filters["include_status_counts"] == 1 assert filters["start_date"] == filters["end_date"] assert "diag_scope_relax" not in filters assert page.table.rowCount() >= 1 def test_demo_appointment_status_counts_respect_date_scope() -> None: repo = DemoDoctorRepository() today = repo._today.isoformat() result = repo.list_appointments( page_no=1, page_size=20, start_date=today, end_date=today, include_status_counts=1, ) assert isinstance(result, PageResult) counts = result.extend.get("status_count", {}) assert int(counts.get("1", counts.get(1, 0))) >= 1 assert int(counts.get("3", counts.get(3, 0))) >= 1 def test_appointment_multiline_cells_receive_enough_row_height( application: QApplication, ) -> None: page = AppointmentsPage( DemoDoctorRepository(), permissions=PermissionSet(["doctor.appointment/lists"]), current_user={"id": 1001, "role_id": 1}, ) page._loaded( { "lists": [ { "id": 15534, "patient_name": "张玉英", "patient_phone": "13800138000", "gender": 0, "age": 42, "height": 162, "weight": 55, "doctor_name": "徐国军", "appointment_date": "2026-08-11", "appointment_time": "14:30", "assistant_name": "蒋露露", "diagnosis_confirmed": 0, "has_prescription": 0, "status": 1, "status_desc": "已预约", "remark": "—", } ], "count": 1, }, page._generation, False, ) patient_text = page.table.item(0, 2).text() appointment_text = page.table.item(0, 4).text() assert patient_text == "张玉英" assert appointment_text.count("\n") == 2 assert "2026-08-11 14:30" in appointment_text required = 3 * max(16, page.table.fontMetrics().lineSpacing()) + 10 assert page.table.rowHeight(0) >= required assert page.table.item(0, 4).toolTip() == appointment_text page.close() def test_video_qr_dialog_renders_downloaded_image_inside_app( application: QApplication, ) -> None: repo = DemoDoctorRepository() page = AppointmentsPage( repo, permissions=PermissionSet(["tcm.diagnosis/videoQr"]), current_user={"id": 1001, "role_id": 1}, ) url = "https://demo.invalid/qrcode/video/88.png" dialog = page._build_qr_dialog( {"patient_name": "鹿立核"}, url, {"qrcode_url": url, "_image_bytes": repo.download_public_image(url)}, ) image = dialog.findChild(QLabel, "VideoQrImage") assert image is not None assert image.pixmap() is not None and not image.pixmap().isNull() assert dialog.findChild(QLabel, "VideoQrImage").text() == "" assert dialog.objectName() == "VideoQrDialog" assert dialog.property("businessDialog") is True buttons = dialog.findChild(QDialogButtonBox) assert next(button for button in buttons.buttons() if button.text() == "浏览器打开").property( "variant" ) == "primary" dialog.close() page.close() application.processEvents() def test_page_owned_appointment_dialogs_have_stable_visual_contract( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: page = AppointmentsPage( DemoDoctorRepository(), permissions=PermissionSet( ["doctor.appointment/complete", "doctor.appointment/addDoctorNote"] ), current_user={"id": 1001, "role_id": 1}, ) captured: list[QDialog] = [] def reject_dialog(dialog: QDialog) -> QDialog.DialogCode: captured.append(dialog) return QDialog.DialogCode.Rejected monkeypatch.setattr(QDialog, "exec", reject_dialog) page._open_custom_date() page._show_detail({"id": 101, "patient_name": "鹿立核"}, page._action_generation) page.table.set_rows( [{"id": 101, "diagnosis_id": 501, "patient_id": 301, "status": 1}] ) page.table.selectRow(0) page._complete_selected() assert [dialog.objectName() for dialog in captured] == [ "AppointmentCustomDateDialog", "AppointmentDetailDialog", "AppointmentCompleteDialog", ] assert all(dialog.property("businessDialog") is True for dialog in captured) assert all( any(label.property("dialogRole") == "title" for label in dialog.findChildren(QLabel)) for dialog in captured ) custom_buttons = captured[0].findChild(QDialogButtonBox) complete_buttons = captured[2].findChild(QDialogButtonBox) assert custom_buttons.button(QDialogButtonBox.StandardButton.Ok).property( "variant" ) == "primary" assert complete_buttons.button(QDialogButtonBox.StandardButton.Ok).property( "variant" ) == "primary" for dialog in captured: dialog.close() page.close() application.processEvents() def test_prescription_case_snapshot_uses_keyword_diagnosis_id( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: requested: list[int] = [] opened: list[tuple[Any, Any]] = [] class Repository: def get_diagnosis_detail(self, *, diagnosis_id: int) -> dict[str, Any]: requested.append(diagnosis_id) return {"id": diagnosis_id, "patient_name": "测试患者"} def run_immediately(function: Any, **callbacks: Any) -> object: try: callbacks["on_success"](function()) except Exception as error: # pragma: no cover - assertion output is clearer callbacks["on_error"](error) finally: callbacks["on_finished"]() return object() monkeypatch.setattr(appointments_module, "run_async", run_immediately) page = AppointmentsPage( Repository(), permissions=PermissionSet(["tcm.diagnosis/kaifang"]), current_user={"id": 9, "role_id": 1}, ) monkeypatch.setattr( page, "_open_prescription_editor", lambda record, detail: opened.append((record, detail)), ) page._begin_case_record_load( {"id": 101, "appointment_id": 101, "diagnosis_id": 501, "patient_id": 501} ) assert requested == [501] assert opened and opened[0][1]["id"] == 501 page.close() application.processEvents() def test_appointment_ai_report_button_visible_with_reception_permission( application: QApplication, ) -> None: hidden = AppointmentsPage( SimpleNamespace(), permissions=PermissionSet(["tcm.diagnosis/kaifang"]), ) assert hidden.ai_button.isHidden() hidden.close() page = AppointmentsPage( SimpleNamespace(), permissions=PermissionSet(["doctor.appointment/reception"]), ) assert not page.ai_button.isHidden() assert not page.ai_button.isEnabled() page.close() application.processEvents() def test_appointments_reference_split_layout_and_video_list( application: QApplication, ) -> None: page = AppointmentsPage( DemoDoctorRepository(), permissions=PermissionSet(["doctor.appointment/lists"]), current_user={"id": 1001, "role_id": 1}, ) page.resize(1460, 820) page._loaded( { "lists": [ { "id": 101, "diagnosis_id": 501, "patient_name": "赵俊霞", "gender": 0, "age": 53, "assistant_name": "自媒体4", "appointment_date": "2026-08-13", "appointment_time": "09:50", "status": 1, "status_desc": "已挂号", } ], "count": 1, }, page._generation, False, ) application.processEvents() assert page.video_list.count() == 1 assert "赵俊霞" in page.video_list.item(0).text() assert page.video_list.parentWidget().width() == 420 assert page.table.objectName() == "AppointmentTable" assert page.date_buttons["today"].isChecked() page.close() application.processEvents()