416 lines
16 KiB
Python
416 lines
16 KiB
Python
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)
|