from __future__ import annotations import json import os from datetime import date from pathlib import Path from typing import Any os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") import httpx import pytest from PySide6.QtWidgets import QApplication, QPushButton from doctor_workstation.core import PermissionSet from doctor_workstation.services.api_client import ApiClient from doctor_workstation.services.mock_repository import DemoDoctorRepository from doctor_workstation.services.repository import RemoteDoctorRepository from doctor_workstation.ui.pages import reception as reception_module from doctor_workstation.ui.pages.reception import ( NOTE_LIMIT, QueueRow, ReceptionPage, _is_image_attachment, ) from doctor_workstation.ui.widgets import StatusBadge @pytest.fixture(scope="module") def application() -> QApplication: return QApplication.instance() or QApplication([]) @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(reception_module, "run_async", run_immediately) @pytest.mark.parametrize( ("path", "expected"), [ ("https://cdn.test/tongue.JPG?token=1", True), ("https://cdn.test/report.webp", True), ("https://cdn.test/report.pdf", False), ], ) def test_note_attachment_preview_type_is_extension_aware(path: str, expected: bool) -> None: assert _is_image_attachment(path) is expected def test_note_attachments_offer_image_preview_and_file_open( application: QApplication, ) -> None: page = ReceptionPage(DemoDoctorRepository(), PermissionSet([])) page._render_notes( [ { "id": 1, "note_date": "2026-08-12", "tongue_images": ["https://cdn.test/tongue.jpg"], "report_files": [ "https://cdn.test/check.png", "https://cdn.test/check.pdf", ], } ] ) labels = [button.text() for button in page.notes_container.findChildren(QPushButton)] assert labels.count("预览") == 2 assert labels.count("打开") == 1 page.close() application.processEvents() def test_queue_status_badge_is_not_clipped_in_narrow_panel( application: QApplication, ) -> None: row = QueueRow( { "patient_name": "张蒙", "status": 1, "status_desc": "待接诊", "appointment_time": "12:45:00", "gender": 1, "age": 36, "assistant_name": "苏亚梅", } ) row.setFixedWidth(280) row.show() application.processEvents() badge = row.findChild(StatusBadge) assert badge is not None assert badge.height() >= 24 assert badge.width() >= 54 assert badge.geometry().right() < row.width() row.close() application.processEvents() def _detail( appointment_id: int, *, name: str, status: int = 1, phone: str = "13800138000", ) -> dict[str, Any]: patient_id = appointment_id + 100 diagnosis_id = appointment_id + 200 return { "appointment": { "id": appointment_id, "patient_id": patient_id, "patient_name": name, "status": status, "appointment_date": date.today().isoformat(), "appointment_time": "09:30", "doctor_name": "张医生", "assistant_name": "李医助", "appointment_type_text": "复诊", "channel_text": "线上", "remark": "准时到诊", }, "patient": { "id": patient_id, "phone": phone, "gender": 2, "age": 42, "height": 165, "weight": 55, "region_text": "浙江省杭州市", }, "diagnosis": { "id": diagnosis_id, "patient_id": patient_id, "patient_name": name, "phone": phone, "chief_complaint": "反复口渴", "present_illness": "持续两周", "clinical_diagnosis": "消渴", }, } def test_queue_uses_admin_same_day_contract( application: QApplication, immediate_async: None, ) -> None: calls: list[dict[str, Any]] = [] class Repository: def list_appointments(self, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs) return {"lists": [], "count": 0} page = ReceptionPage(Repository(), PermissionSet([])) page.search_edit.setText(" 王小明 ") page.refresh() assert calls == [ { "status": 1, "start_date": date.today().isoformat(), "end_date": date.today().isoformat(), "page_no": 1, "page_size": 15, "patient_name": "王小明", } ] page.queue_tabs.setCurrentIndex(1) assert calls[-1]["status"] == 4 assert calls[-1]["start_date"] == calls[-1]["end_date"] page.close() application.processEvents() def test_fast_patient_switch_rejects_late_detail( 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(reception_module, "run_async", queue_async) page = ReceptionPage(object(), PermissionSet(["*"])) first = {"id": 11, "patient_id": 111, "diagnosis_id": 211, "patient_name": "甲患者"} second = {"id": 22, "patient_id": 122, "diagnosis_id": 222, "patient_name": "乙患者"} page._select_record(first) page.note_edit.setPlainText("甲患者的未保存草稿") page._pending_report_files = [r"C:\records\first.pdf"] page._select_record(second) assert len(callbacks) == 2 assert page.patient_name_label.text() == "乙患者" assert page.note_edit.toPlainText() == "" assert page._pending_report_files == [] callbacks[0]["on_success"]({"detail": _detail(11, name="甲患者")}) callbacks[0]["on_finished"]() assert page._selected_appointment_id == 22 assert page.patient_name_label.text() == "乙患者" assert page._selected_detail is None callbacks[1]["on_success"]({"detail": _detail(22, name="乙患者")}) callbacks[1]["on_finished"]() assert page._selected_detail == _detail(22, name="乙患者") assert page.patient_name_label.text() == "乙患者" assert "反复口渴" in page.diagnosis_text.text() page.close() application.processEvents() def test_queue_load_more_accumulates_to_total_boundary( application: QApplication, immediate_async: None, ) -> None: calls: list[dict[str, Any]] = [] all_rows = [ { "id": index, "patient_id": 1000 + index, "diagnosis_id": 2000 + index, "patient_name": f"患者{index:02d}", "status": 1, } for index in range(1, 23) ] class Repository: def list_appointments(self, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs) start = (kwargs["page_no"] - 1) * kwargs["page_size"] return { "lists": all_rows[start : start + kwargs["page_size"]], "count": len(all_rows), } def get_reception(self, appointment_id: int) -> dict[str, Any]: row = all_rows[appointment_id - 1] return { "appointment": row, "diagnosis": {"id": row["diagnosis_id"], "patient_id": row["patient_id"]}, } page = ReceptionPage(Repository(), PermissionSet([])) page.refresh() assert page.queue_list.count() == 15 assert page.load_more_button.isVisibleTo(page) page._load_more() assert [call["page_no"] for call in calls] == [1, 2] assert all(call["page_size"] == 15 for call in calls) assert page.queue_list.count() == 22 assert page.queue_summary.text() == "已加载 22 / 共 22 位患者" assert page.load_more_button.isHidden() page.close() application.processEvents() def test_search_and_tab_changes_reset_accumulated_pages( application: QApplication, immediate_async: None, ) -> None: calls: list[dict[str, Any]] = [] class Repository: def list_appointments(self, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs) if kwargs["status"] == 4: rows = [{"id": 401, "patient_name": "过号患者", "status": 4}] elif kwargs["patient_name"]: rows = [{"id": 201, "patient_name": "搜索患者", "status": 1}] else: rows = [ {"id": index, "patient_name": f"患者{index}", "status": 1} for index in range(1, 19) ] start = (kwargs["page_no"] - 1) * kwargs["page_size"] return {"lists": rows[start : start + kwargs["page_size"]], "count": len(rows)} def get_reception(self, appointment_id: int) -> dict[str, Any]: return { "appointment": { "id": appointment_id, "patient_id": appointment_id + 1000, "status": 1, }, "diagnosis": { "id": appointment_id + 2000, "patient_id": appointment_id + 1000, }, } page = ReceptionPage(Repository(), PermissionSet([])) page.refresh() page._load_more() assert page.queue_list.count() == 18 page.search_edit.setText("搜索") page.refresh() assert page.queue_list.count() == 1 assert page._queue_page == 1 assert calls[-1]["patient_name"] == "搜索" page.queue_tabs.setCurrentIndex(1) assert page.queue_list.count() == 1 assert page._queue_page == 1 assert calls[-1]["status"] == 4 page.close() application.processEvents() def test_queue_worker_uses_frozen_widget_snapshot( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: jobs: list[Any] = [] calls: list[dict[str, Any]] = [] def queue_async(function: Any, **_options: Any) -> object: jobs.append(function) return object() class Repository: def list_appointments(self, **kwargs: Any) -> dict[str, Any]: calls.append(kwargs) return {"lists": [], "count": 0} monkeypatch.setattr(reception_module, "run_async", queue_async) page = ReceptionPage(Repository(), PermissionSet([])) page.search_edit.setText("甲患者") page.refresh() page.search_edit.blockSignals(True) page.search_edit.setText("乙患者") page.search_edit.blockSignals(False) page.queue_tabs.blockSignals(True) page.queue_tabs.setCurrentIndex(1) page.queue_tabs.blockSignals(False) jobs[0]() assert calls == [ { "status": 1, "start_date": date.today().isoformat(), "end_date": date.today().isoformat(), "page_no": 1, "page_size": 15, "patient_name": "甲患者", } ] page.close() application.processEvents() def test_phone_permission_and_ungated_notify_video_actions( application: QApplication, immediate_async: None, ) -> None: detail = _detail(31, name="脱敏患者") class Repository: def get_reception(self, appointment_id: int) -> dict[str, Any]: assert appointment_id == 31 return detail masked_page = ReceptionPage(Repository(), PermissionSet([])) masked_page._select_record(detail["appointment"]) assert masked_page.patient_labels["phone"].text() == "138****8000" assert not masked_page.notify_button.isHidden() assert not masked_page.video_button.isHidden() plain_page = ReceptionPage(Repository(), PermissionSet(["tcm.diagnosis/phonePlain"])) plain_page._select_record(detail["appointment"]) assert plain_page.patient_labels["phone"].text() == "13800138000" assert plain_page.appointment_labels["doctor"].text() == "张医生" assert plain_page.appointment_labels["assistant"].text() == "李医助" assert plain_page.patient_labels["region"].text() == "浙江省杭州市" masked_page.close() plain_page.close() application.processEvents() def test_video_payload_keeps_three_identifiers_distinct( application: QApplication, immediate_async: None, ) -> None: detail = _detail(41, name="视频患者") class Repository: def get_reception(self, appointment_id: int) -> dict[str, Any]: assert appointment_id == 41 return detail page = ReceptionPage(Repository(), PermissionSet([])) page._select_record(detail["appointment"]) emitted: list[dict[str, Any]] = [] page.video_requested.connect(emitted.append) page._request_video() assert emitted == [ { "source": "reception", "appointment_id": 41, "patient_id": 141, "diagnosis_id": 241, "patient_name": "视频患者", "mode": "im", "record": detail["appointment"], } ] page.close() application.processEvents() def test_completion_revalidates_server_status_before_write( application: QApplication, ) -> None: completed: list[int] = [] class Repository: status = 3 def get_reception(self, appointment_id: int) -> dict[str, Any]: return { "appointment": { "id": appointment_id, "patient_id": 151, "status": self.status, } } def complete_appointment(self, appointment_id: int) -> dict[str, bool]: completed.append(appointment_id) return {"ok": True} repository = Repository() page = ReceptionPage(repository, PermissionSet(["doctor.appointment/complete"])) with pytest.raises(ValueError, match="状态已变化"): page._complete_after_revalidation(51) assert completed == [] repository.status = 4 assert page._complete_after_revalidation(51) == {"ok": True} assert completed == [51] page.close() application.processEvents() def test_note_limit_and_attachment_payload_contract( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: jobs: list[tuple[Any, dict[str, Any]]] = [] received: list[dict[str, Any]] = [] uploads: list[dict[str, Any]] = [] def queue_async(function: Any, **options: Any) -> object: jobs.append((function, options)) return object() class Repository: def upload_material(self, **kwargs: Any) -> str: uploads.append(kwargs) suffix = "tongue.jpg" if kwargs["material_type"] == "image" else "report.pdf" return f"/uploads/{kwargs['material_type']}/{suffix}" def add_doctor_note(self, diagnosis_id: int, content: str, **kwargs: Any) -> None: received.append({"diagnosis_id": diagnosis_id, "content": content, **kwargs}) monkeypatch.setattr(reception_module, "run_async", queue_async) page = ReceptionPage(Repository(), PermissionSet(["doctor.appointment/addDoctorNote"])) detail = _detail(61, name="备注患者") page._selected_record = detail["appointment"] page._selected_appointment_id = 61 page._selected_detail = detail page._detail_generation = 7 page.note_edit.setPlainText("字" * (NOTE_LIMIT + 20)) page._pending_tongue_images = [r"C:\records\tongue.jpg"] page._pending_report_files = [r"C:\records\report.pdf"] assert len(page.note_edit.toPlainText()) == NOTE_LIMIT assert page.note_counter.text() == f"{NOTE_LIMIT} / {NOTE_LIMIT}" page._save_note() assert len(jobs) == 1 jobs[0][0]() assert uploads == [ {"path": r"C:\records\tongue.jpg", "material_type": "image", "cid": 0}, {"path": r"C:\records\report.pdf", "material_type": "file", "cid": 0}, ] assert received == [ { "diagnosis_id": 261, "content": "字" * NOTE_LIMIT, "tongue_images": ["/uploads/image/tongue.jpg"], "report_files": ["/uploads/file/report.pdf"], } ] jobs[0][1]["on_finished"]() assert not page._note_busy page.close() application.processEvents() def test_remote_note_uses_multipart_then_server_urls_only( application: QApplication, tmp_path: Path, ) -> None: requests: list[httpx.Request] = [] tongue = tmp_path / "tongue.jpg" report = tmp_path / "report.pdf" tongue.write_bytes(b"tongue-image") report.write_bytes(b"report-file") def handler(request: httpx.Request) -> httpx.Response: requests.append(request) if request.url.path.endswith("/upload/image"): return httpx.Response( 200, json={"code": 1, "data": {"uri": "/materials/tongue.jpg"}}, ) if request.url.path.endswith("/upload/file"): return httpx.Response( 200, json={"code": 1, "data": {"url": "https://cdn.test/report.pdf"}}, ) return httpx.Response(200, json={"code": 1, "data": {"id": 9}}) with ApiClient( "https://example.test", transport=httpx.MockTransport(handler), ) as client: repository = RemoteDoctorRepository(client) page = ReceptionPage(repository, PermissionSet([])) result = page._upload_and_add_note( 501, "两阶段备注", [str(tongue)], [str(report)], ) assert result == {"id": 9} assert [request.url.path.rsplit("/", 2)[-2:] for request in requests] == [ ["upload", "image"], ["upload", "file"], ["doctor.appointment", "addDoctorNote"], ] for upload_request in requests[:2]: assert upload_request.headers["content-type"].startswith("multipart/form-data; boundary=") assert b'name="cid"' in upload_request.content assert b"\r\n0\r\n" in upload_request.content note_payload = json.loads(requests[-1].content) assert note_payload == { "diagnosis_id": 501, "content": "两阶段备注", "tongue_images": ["/materials/tongue.jpg"], "report_files": ["https://cdn.test/report.pdf"], } assert str(tmp_path) not in requests[-1].content.decode("utf-8") page.close() application.processEvents() def test_partial_upload_failure_never_submits_note( application: QApplication, ) -> None: submitted: list[dict[str, Any]] = [] class Repository: def upload_material(self, path: str, material_type: str, cid: int = 0) -> str: del material_type, cid if path.endswith("bad.pdf"): raise OSError("磁盘读取失败") return "/materials/good.jpg" def add_doctor_note(self, **kwargs: Any) -> None: submitted.append(kwargs) page = ReceptionPage(Repository(), PermissionSet([])) with pytest.raises(RuntimeError, match="bad.pdf.*上传失败"): page._upload_and_add_note( 501, "不会提交", [r"C:\records\good.jpg"], [r"C:\records\bad.pdf"], ) assert submitted == [] page.close() application.processEvents()