from __future__ import annotations import os from copy import deepcopy from threading import Event from typing import Any os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") import pytest from PySide6.QtCore import QCoreApplication, QEvent, Qt, QTimer from PySide6.QtTest import QTest from PySide6.QtWidgets import QApplication from doctor_workstation.core import PermissionSet from doctor_workstation.ui.pages import reception as reception_module from doctor_workstation.ui.pages.reception import ReceptionPage @pytest.fixture(scope="module") def application() -> QApplication: return QApplication.instance() or QApplication([]) class RefreshRepository: def __init__(self) -> None: self.calls: list[tuple[str, Any]] = [] self.detail = { "appointment": {"id": 51, "patient_name": "刷新测试患者", "status": 1}, "diagnosis": {"id": 251, "patient_id": 151, "symptoms": "原病历"}, "patient": {"id": 151}, } self.rows = [deepcopy(self.detail["appointment"])] self.reports: list[dict[str, Any]] = [] def list_appointments(self, **query: Any) -> dict[str, Any]: self.calls.append(("queue", query)) return {"lists": deepcopy(self.rows), "count": len(self.rows)} def get_reception(self, appointment_id: int) -> dict[str, Any]: self.calls.append(("detail", appointment_id)) return deepcopy(self.detail) def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: self.calls.append(("reports", patient_id)) return {"patient_id": patient_id, "reports": deepcopy(self.reports)} def generate_patient_ai_report(self, patient_id: int, **_options: Any) -> None: self.calls.append(("POST", patient_id)) raise AssertionError("刷新不得生成 AI 报告") def get_diagnosis_ai_analysis(self, diagnosis_id: int, **_options: Any) -> None: self.calls.append(("legacy_POST", diagnosis_id)) raise AssertionError("刷新不得调用名称为 get 的旧版生成接口") @pytest.fixture def harness(application: QApplication, monkeypatch: pytest.MonkeyPatch): jobs: list[dict[str, Any]] = [] toasts: list[str] = [] def queue(function: Any, **options: Any) -> object: jobs.append({"function": function, **options}) return object() monkeypatch.setattr(reception_module, "run_async", queue) monkeypatch.setattr(reception_module, "_AI_AUTOMATIC_GENERATION_SETTLE_SECONDS", 0) monkeypatch.setattr( reception_module, "show_toast", lambda _parent, text, *_args: toasts.append(text) ) repository = RefreshRepository() page = ReceptionPage( repository, PermissionSet( [ "doctor.appointment/addDoctorNote", "doctor.appointment/complete", "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", "tcm.diagnosis/aiAnalysis", ] ), ) page._selected_appointment_id = 51 page._selected_record = deepcopy(repository.detail["appointment"]) page._selected_detail = deepcopy(repository.detail) page._update_action_state(repository.detail["appointment"], repository.detail["diagnosis"]) yield page, repository, jobs, toasts page.close() page.deleteLater() QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete) def finish(job: dict[str, Any]) -> None: try: result = job["function"]() except Exception as error: job["on_error"](error) else: job["on_success"](result) finally: job["on_finished"]() def expire_cooldown(page: ReceptionPage) -> None: page._refresh_cooldown.stop() page._refresh_cooldown.timeout.emit() def snapshot(version: int) -> dict[str, Any]: return { "id": version, "patient_id": 151, "model_key": "qwen", "version": version, "generated_at": f"2026-08-31 10:{version:02}:00", "report": {"diagnosis": f"测试报告第 {version} 版", "treatment_advice": "测试建议"}, } def test_manual_refresh_supersedes_hung_queue_and_detail_independently(harness) -> None: page, repository, jobs, _toasts = harness page.refresh(workspace_refresh=True) old_queue = jobs.pop() page._load_detail(page._selected_record, clear=False) old_detail = jobs.pop() old_generation = page._detail_generation page.refresh_button.click() assert len(jobs) == 2 assert page._detail_generation > old_generation assert page._queue_loading and page._detail_loading # A slow queue cannot hold the new detail back. finish(jobs[0]) assert repository.calls == [("detail", 51)] assert not page._detail_loading assert page._queue_loading assert len(jobs) == 3 # exactly one fresh saved-report read finish(jobs[1]) assert len(jobs) == 3 # queue must not duplicate detail work finish(jobs[2]) old_queue["on_success"]({"lists": [{"id": 999, "patient_name": "迟到队列"}]}) old_queue["on_finished"]() old_detail["on_success"]({"detail": {"appointment": {"id": 51, "patient_name": "旧详情"}}}) old_detail["on_finished"]() assert page._selected_appointment_id == 51 assert page.patient_name_label.text() == "刷新测试患者" assert not page._queue_loading and not page._detail_loading assert [call[0] for call in repository.calls] == ["detail", "queue", "reports"] def test_refresh_preserves_query_loaded_pages_selection_and_drafts(harness) -> None: page, repository, jobs, _toasts = harness page._queue_date = "2026-08-03" page._queue_filter_status = None page.search_edit.blockSignals(True) page.search_edit.setText(" 测试 ") page.search_edit.blockSignals(False) page._queue_page = 3 page._queue_records = [dict(id=index) for index in range(1, 46)] page.note_edit.setPlainText("尚未保存的备注") page._pending_tongue_images.append("draft-image.png") page._pending_report_files.append("draft-report.pdf") page.detail_tabs.setCurrentIndex(4) # The current patient is no longer in this queue: don't silently select another. repository.rows = [{"id": 52, "patient_name": "其他测试患者", "status": 1}] page._refresh_workspace() finish(jobs[1]) query = repository.calls[-1][1] assert query == { "status": None, "start_date": "2026-08-03", "end_date": "2026-08-03", "patient_name": "测试", "page_no": 1, "page_size": 45, "include_status_counts": 1, } finish(jobs[0]) assert page._selected_appointment_id == 51 assert page.queue_list.currentRow() == -1 assert page.note_edit.toPlainText() == "尚未保存的备注" assert page._pending_tongue_images == ["draft-image.png"] assert page._pending_report_files == ["draft-report.pdf"] assert page.detail_tabs.currentIndex() == 4 def test_refresh_can_retry_while_old_requests_never_finish(harness) -> None: page, _repository, jobs, _toasts = harness page._refresh_workspace() page._refresh_workspace() assert len(jobs) == 2 assert not page.refresh_button.isEnabled() expire_cooldown(page) page.refresh_button.click() assert len(jobs) == 4 # Superseded queued work never goes to the server. assert jobs[0]["function"]() == {"cancelled": True} assert jobs[1]["function"]() is reception_module._ASYNC_REQUEST_CANCELLED jobs[0]["on_finished"]() jobs[1]["on_finished"]() assert page._queue_loading and page._detail_loading @pytest.mark.parametrize("empty", [True, False]) def test_automatic_polls_after_refresh_keep_missing_patient_and_draft(harness, empty) -> None: page, repository, jobs, _toasts = harness page.note_edit.setPlainText("不能因轮询丢失的备注") page._pending_tongue_images.append("draft-image.png") repository.rows = [] if empty else [{"id": 52, "patient_name": "其他患者", "status": 1}] page._refresh_workspace() finish(jobs[0]) finish(jobs[1]) finish(jobs[2]) for _ in range(3): before = len(jobs) page._poll_queue() assert len(jobs) == before + 1 finish(jobs[-1]) assert page._selected_appointment_id == 51 assert page.note_edit.toPlainText() == "不能因轮询丢失的备注" assert page._pending_tongue_images == ["draft-image.png"] def test_destroyed_page_ignores_every_late_refresh_callback(harness) -> None: _page, repository, jobs, _toasts = harness closed_page = ReceptionPage(repository, PermissionSet([])) closed_page._refresh_workspace() job = jobs[-1] closed_page.deleteLater() QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete) job["on_success"]({"lists": []}) job["on_error"](RuntimeError("迟到异常")) job["on_finished"]() assert job["function"]() is reception_module._ASYNC_REQUEST_CANCELLED assert not repository.calls @pytest.mark.parametrize("busy_flag", ["_note_busy", "_completion_pending"]) def test_refresh_does_not_disturb_business_submission(harness, busy_flag: str) -> None: page, _repository, jobs, toasts = harness setattr(page, busy_flag, True) generation = page._detail_generation page._refresh_workspace() assert not jobs assert page._detail_generation == generation assert getattr(page, busy_flag) assert "正在提交" in toasts[-1] def test_refresh_invalidates_in_flight_ai_read_and_rejects_late_result(harness) -> None: page, repository, jobs, _toasts = harness page._load_patient_ai_reports(151, appointment_id=51, diagnosis_id=251, force=True) old_read = jobs.pop() old_key = next(iter(page._patient_ai_list_requests)) page._patient_ai_list_started.add(old_key) cancel = page._patient_ai_list_cancel_events[old_key] repository.reports = [snapshot(2)] page._refresh_workspace() assert cancel.is_set() assert old_key not in page._patient_ai_list_requests finish(jobs[0]) finish(jobs[2]) assert page._ai_analysis_payloads["qwen"]["version"] == 2 old_read["on_success"]({"patient_id": 151, "reports": [snapshot(1)]}) old_read["on_finished"]() assert page._ai_analysis_payloads["qwen"]["version"] == 2 assert not page._ai_analysis_loading def test_manual_ai_read_does_not_need_automatic_slots_or_generate_missing_reports(harness) -> None: page, repository, jobs, _toasts = harness slots = reception_module._AI_AUTOMATIC_REQUEST_SLOTS assert slots.acquire(blocking=False) assert slots.acquire(blocking=False) try: page._refresh_workspace() finish(jobs[0]) finish(jobs[2]) assert ("reports", 151) in repository.calls assert page._ai_analysis_state == "missing" assert not page._ai_analysis_loading assert not slots.acquire(blocking=False) # GUI must not release others' slots finally: slots.release() slots.release() assert not any("POST" in call[0] for call in repository.calls) assert page.ai_analysis_regenerate_button.isEnabled() def test_refresh_keeps_started_generation_single_flight(harness) -> None: page, repository, jobs, _toasts = harness key = (1, 51, 151, "qwen") page._patient_ai_generation_requests.add(key) page._patient_ai_generation_started.add(key) cancel = Event() page._patient_ai_generation_cancel_events[key] = cancel page._refresh_workspace() finish(jobs[0]) assert len(jobs) == 2 assert key in page._patient_ai_generation_requests assert key in page._patient_ai_generation_started assert not cancel.is_set() assert "不会重复生成" in page._ai_analysis_operation_error assert not any(call[0] in {"POST", "reports"} for call in repository.calls) def test_legacy_ai_refresh_never_invokes_post_endpoint(harness) -> None: page, repository, jobs, _toasts = harness page._can_patient_ai_read = False page._refresh_workspace() finish(jobs[0]) assert len(jobs) == 2 assert repository.calls == [("detail", 51)] assert "无法只读刷新" in page.ai_analysis_state_label.text() def test_first_selection_from_manual_refresh_is_read_only(harness) -> None: page, repository, jobs, _toasts = harness page._clear_selection() page._refresh_workspace() assert len(jobs) == 1 finish(jobs[0]) finish(jobs[1]) finish(jobs[2]) assert [call[0] for call in repository.calls] == ["queue", "detail", "reports"] assert page._selected_appointment_id == 51 assert page._ai_analysis_state == "missing" def test_late_refresh_cannot_switch_back_to_previous_patient(harness) -> None: page, _repository, jobs, _toasts = harness page._refresh_workspace() second = {"id": 52, "patient_name": "新选择的测试患者", "status": 1} page._select_record(second) assert page._selected_appointment_id == 52 jobs[0]["on_success"]({"detail": {"appointment": {"id": 51, "patient_name": "旧患者"}}}) jobs[1]["on_success"]({"lists": [{"id": 51, "patient_name": "旧患者"}]}) assert page._selected_appointment_id == 52 assert page.patient_name_label.text() == "新选择的测试患者" assert page._detail_loading # stale finished must not clear patient B's loading def test_ai_read_timeout_is_retryable_and_cannot_install_late_snapshot( harness, monkeypatch ) -> None: page, _repository, jobs, _toasts = harness page._refresh_workspace() monkeypatch.setattr(reception_module, "_REFRESH_TIMEOUT_MS", 10) finish(jobs[0]) finish(jobs[1]) assert len(jobs) == 3 QTest.qWait(60) assert not page._ai_analysis_loading assert page._ai_analysis_state == "error" assert page.ai_analysis_retry_button.isEnabled() jobs[2]["on_success"]({"patient_id": 151, "reports": [snapshot(1)]}) assert not page._ai_analysis_payloads assert not page._patient_ai_list_requests def test_refresh_timeout_stops_loading_and_ignores_late_success(harness, monkeypatch) -> None: page, _repository, jobs, _toasts = harness monkeypatch.setattr(reception_module, "_REFRESH_TIMEOUT_MS", 10) monkeypatch.setattr(reception_module, "_REFRESH_COOLDOWN_MS", 10) page._refresh_workspace() QTest.qWait(60) assert not page._queue_loading and not page._detail_loading assert page.refresh_button.isEnabled() assert "刷新超时" in page.queue_banner.label.text() assert "刷新超时" in page.detail_banner.label.text() jobs[0]["on_success"]({"detail": {"appointment": {"id": 51, "patient_name": "迟到详情"}}}) jobs[1]["on_success"]({"lists": []}) assert page._selected_appointment_id == 51 assert page.patient_name_label.text() != "迟到详情" assert "刷新超时" in page.detail_banner.label.text() assert jobs[0]["function"]() is reception_module._ASYNC_REQUEST_CANCELLED page._refresh_workspace() assert len(jobs) == 4 def test_refresh_read_error_and_missing_callback_are_terminal(harness) -> None: page, _repository, jobs, _toasts = harness page._refresh_workspace() jobs[0]["on_error"](RuntimeError("测试断网")) jobs[0]["on_finished"]() jobs[1]["on_finished"]() assert not page._queue_loading and not page._detail_loading assert "测试断网" in page.detail_banner.label.text() assert "未返回有效结果" in page.queue_banner.label.text() def test_refresh_button_visible_and_f5_uses_same_debounce(harness, application) -> None: page, _repository, jobs, _toasts = harness page._queue_records = [page._selected_record] # prevent showEvent's initial fetch page.resize(1440, 1000) page.show() page.activateWindow() page.poll_timer.stop() application.processEvents() assert page.refresh_button.isVisible() assert not page.refresh_button.icon().isNull() assert page.refresh_button.accessibleName() == "刷新接诊台" assert page.refresh_button.width() >= 34 assert page.refresh_button.geometry().right() < page.filter_disclosure.button.geometry().left() assert not page.queue_date_button.isVisible() page.note_edit.setFocus() application.processEvents() QTest.keyClick(page.note_edit, Qt.Key.Key_F5) assert len(jobs) == 2 QTest.keyClick(page.note_edit, Qt.Key.Key_F5) assert len(jobs) == 2 assert page._refresh_cooldown.isActive() assert page._refresh_cooldown in page.findChildren(QTimer)