zyt/app/tests/test_reception_refresh_ui.py

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)