This commit is contained in:
Your Name
2026-09-01 10:42:16 +08:00
parent 456dd667df
commit 5bd5eae62d
358 changed files with 2276 additions and 495 deletions
+306
View File
@@ -0,0 +1,306 @@
from __future__ import annotations
import os
from typing import Any
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtWidgets import QApplication
from doctor_workstation.core import PermissionSet
from doctor_workstation.ui.dialogs.appointment_complete import (
COMPLETION_NOTE_LIMIT,
AppointmentCompleteDialog,
)
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 CompletionRepository:
def __init__(self) -> None:
self.calls: list[tuple[Any, ...]] = []
self.fail_complete = False
self.fail_note = False
self.detail: dict[str, Any] = {
"appointment": {"id": 51, "patient_id": 251, "status": 1},
"diagnosis": {"id": 251, "patient_id": 151},
"patient": {"id": 151},
}
def get_reception(self, appointment_id: int) -> dict[str, Any]:
self.calls.append(("revalidate", appointment_id))
return self.detail
def complete_appointment(self, appointment_id: int) -> dict[str, bool]:
self.calls.append(("complete", appointment_id))
if self.fail_complete:
raise RuntimeError("完成接口失败")
return {"ok": True}
def add_doctor_note(self, diagnosis_id: int, content: str) -> dict[str, bool]:
self.calls.append(("note", diagnosis_id, content))
if self.fail_note:
raise RuntimeError("备注接口失败")
return {"ok": True}
@pytest.fixture
def harness(application: QApplication, monkeypatch: pytest.MonkeyPatch):
jobs: list[dict[str, Any]] = []
toasts: list[tuple[str, str]] = []
refreshes: list[bool] = []
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,
"show_toast",
lambda _parent, text, kind, *_args: toasts.append((text, kind)),
)
repository = CompletionRepository()
page = ReceptionPage(
repository,
PermissionSet(["doctor.appointment/complete", "doctor.appointment/addDoctorNote"]),
)
page._selected_appointment_id = 51
page._selected_record = dict(repository.detail["appointment"])
page._selected_detail = repository.detail
page._update_action_state(repository.detail["appointment"], repository.detail["diagnosis"])
monkeypatch.setattr(page, "refresh", lambda *, silent=False: refreshes.append(silent))
yield page, repository, jobs, toasts, refreshes
if page._completion_dialog is not None:
page._completion_dialog.set_busy(False)
page._completion_dialog.reject()
page.close()
application.processEvents()
def finish_job(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"]()
@pytest.mark.parametrize("can_note", [True, False])
def test_completion_dialog_optional_note_limit_and_busy_state(
application: QApplication, can_note: bool
) -> None:
dialog = AppointmentCompleteDialog(can_note=can_note)
submitted: list[str] = []
dialog.submitted.connect(submitted.append)
dialog.show()
application.processEvents()
try:
assert dialog.windowTitle() == "完成问诊"
assert dialog.note_edit.isVisible() is can_note
assert dialog.note_counter.isVisible() is can_note
assert dialog.note_counter.text() == "0 / 500"
dialog.note_edit.setPlainText("" * 501)
assert dialog.note_edit.toPlainText() == "" * COMPLETION_NOTE_LIMIT
assert dialog.note_counter.text() == "500 / 500"
dialog.note_edit.insertPlainText("额外内容")
assert len(dialog.note_edit.toPlainText()) == COMPLETION_NOTE_LIMIT
dialog.note_edit.setPlainText(" 测试备注\n第二行 ")
dialog.confirm_button.click()
assert submitted == ["测试备注\n第二行" if can_note else ""]
dialog.set_busy(True)
assert dialog.note_edit.isReadOnly()
assert not dialog.cancel_button.isEnabled()
dialog.confirm_button.click()
dialog.reject()
dialog.close()
assert len(submitted) == 1
assert dialog.isVisible()
dialog.show_error("提交失败,请重试")
assert dialog.note_edit.toPlainText() == " 测试备注\n第二行 "
assert dialog.confirm_button.isEnabled()
dialog.cancel_button.click()
assert not dialog.isVisible()
finally:
dialog.set_busy(False)
dialog.close()
def test_cancel_completion_does_not_make_requests(harness: Any) -> None:
page, repository, jobs, _toasts, _refreshes = harness
page.complete_button.click()
dialog = page._completion_dialog
assert isinstance(dialog, AppointmentCompleteDialog)
page._complete_appointment()
assert page._completion_dialog is dialog
assert jobs == []
dialog.note_edit.setPlainText("取消后不应保存")
dialog.cancel_button.click()
assert page._completion_dialog is None
assert repository.calls == []
assert jobs == []
@pytest.mark.parametrize("note", ["", " \n ", " 测试备注\n补充内容 ", "" * 500])
def test_complete_then_append_note_and_refresh_without_duplicate_submission(
harness: Any, note: str
) -> None:
page, repository, jobs, toasts, refreshes = harness
page.complete_button.click()
dialog = page._completion_dialog
dialog.note_edit.setPlainText(note)
dialog.confirm_button.click()
assert page._completion_pending
assert not page.complete_button.isEnabled()
assert len(jobs) == 1
# Polling and direct handler calls cannot enable/dispatch a second request.
page._update_action_state(repository.detail["appointment"], repository.detail["diagnosis"])
assert not page.complete_button.isEnabled()
page._complete_appointment()
dialog.confirm_button.click()
assert len(jobs) == 1
finish_job(jobs.pop())
expected = [("revalidate", 51), ("complete", 51)]
if note.strip():
expected.append(("note", 251, note.strip()))
assert repository.calls == expected
assert page._completion_dialog is None
assert not page._completion_pending
assert page._selected_appointment_id is None
assert toasts[-1] == ("接诊已完成。", "success")
assert refreshes == [True]
def test_completion_without_note_permission_does_not_submit_hidden_note(harness: Any) -> None:
page, repository, jobs, _toasts, _refreshes = harness
page._can_note = False
page._complete_appointment()
dialog = page._completion_dialog
assert dialog.note_edit.isHidden()
dialog.note_edit.setPlainText("不可提交")
dialog.confirm_button.click()
finish_job(jobs.pop())
assert repository.calls == [("revalidate", 51), ("complete", 51)]
def test_completion_failure_preserves_note_and_allows_explicit_retry(harness: Any) -> None:
page, repository, jobs, toasts, refreshes = harness
repository.fail_complete = True
page._complete_appointment()
dialog = page._completion_dialog
dialog.note_edit.setPlainText("需要保留的备注")
dialog.confirm_button.click()
finish_job(jobs.pop())
assert repository.calls == [("revalidate", 51), ("complete", 51)]
assert page._completion_dialog is dialog
assert dialog.isVisible()
assert dialog.note_edit.toPlainText() == "需要保留的备注"
assert dialog.confirm_button.isEnabled()
assert "完成接口失败" in dialog.banner.label.text()
assert not page._completion_pending
assert page.complete_button.isEnabled()
assert page._selected_appointment_id == 51
assert toasts == []
assert refreshes == []
repository.fail_complete = False
dialog.confirm_button.click()
finish_job(jobs.pop())
assert repository.calls[-1] == ("note", 251, "需要保留的备注")
assert page._completion_dialog is None
@pytest.mark.parametrize("missing_diagnosis", [True, False])
def test_note_failure_is_partial_success_and_keeps_note_for_copying(
harness: Any, missing_diagnosis: bool
) -> None:
page, repository, jobs, toasts, refreshes = harness
if missing_diagnosis:
repository.detail["diagnosis"] = {}
else:
repository.fail_note = True
page._complete_appointment()
dialog = page._completion_dialog
dialog.note_edit.setPlainText("备注不能丢失")
dialog.confirm_button.click()
finish_job(jobs.pop())
assert [call[0] for call in repository.calls] == (
["revalidate", "complete"] if missing_diagnosis else ["revalidate", "complete", "note"]
)
assert page._selected_appointment_id is None
assert refreshes == [True]
assert toasts[-1][1] == "warning"
assert "问诊已完成" in dialog.banner.label.text()
assert "备注未保存" in dialog.banner.label.text()
assert dialog.note_edit.toPlainText() == "备注不能丢失"
assert dialog.note_edit.isReadOnly()
assert dialog.confirm_button.isHidden()
assert not dialog.confirm_button.isEnabled()
assert dialog.cancel_button.text() == "关闭"
dialog.confirm_button.click()
assert jobs == []
@pytest.mark.parametrize("changed_id", [True, False])
def test_completion_revalidation_rejects_changed_record_before_any_write(
harness: Any, changed_id: bool
) -> None:
page, repository, _jobs, _toasts, _refreshes = harness
if changed_id:
repository.detail["appointment"]["id"] = 52
else:
repository.detail["appointment"]["status"] = 3
with pytest.raises(ValueError, match="不一致|状态已变化"):
page._complete_after_revalidation(51, "测试备注")
assert repository.calls == [("revalidate", 51)]
def test_completion_checks_permissions_and_note_length_before_requests(harness: Any) -> None:
page, repository, jobs, _toasts, _refreshes = harness
with pytest.raises(ValueError, match="500"):
page._complete_after_revalidation(51, "" * 501)
page._can_note = False
with pytest.raises(ValueError, match="备注权限"):
page._complete_after_revalidation(51, "测试备注")
page._can_complete = False
page._complete_appointment()
assert page._completion_dialog is None
with pytest.raises(ValueError, match="完成接诊权限"):
page._complete_after_revalidation(51)
assert repository.calls == []
assert jobs == []
@pytest.mark.parametrize("switch_before_confirm", [True, False])
def test_completion_does_not_mutate_or_clear_a_new_selection(
harness: Any, switch_before_confirm: bool
) -> None:
page, repository, jobs, _toasts, _refreshes = harness
page._complete_appointment()
dialog = page._completion_dialog
dialog.note_edit.setPlainText("原患者的备注")
if not switch_before_confirm:
dialog.confirm_button.click()
page._selected_appointment_id = 52
page._selected_record = {"id": 52, "status": 1}
page._selected_detail = {"appointment": page._selected_record, "diagnosis": {"id": 252}}
page._detail_generation += 1
page._update_action_state(page._selected_record, {"id": 252})
if switch_before_confirm:
dialog.confirm_button.click()
assert jobs == []
assert repository.calls == []
assert "已切换" in dialog.banner.label.text()
else:
finish_job(jobs.pop())
assert repository.calls[-1] == ("note", 251, "原患者的备注")
assert page.complete_button.isEnabled()
assert page._selected_appointment_id == 52
@@ -0,0 +1,122 @@
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
import pytest
from PySide6.QtCore import QPoint, Qt
from PySide6.QtTest import QTest
from PySide6.QtWidgets import QApplication, QHBoxLayout, QPushButton, QWidget
from doctor_workstation.ui.pages.reception import RECEPTION_QSS, _ReceptionCompleteButton
from doctor_workstation.ui.theme import apply_theme
@pytest.fixture
def controls():
app = QApplication.instance() or QApplication([])
old_stylesheet, old_font, old_palette = app.styleSheet(), app.font(), app.palette()
old_style = app.style().objectName()
apply_theme(app)
host = QWidget()
host.setObjectName("ReceptionPage")
host.setStyleSheet(RECEPTION_QSS)
layout = QHBoxLayout(host)
layout.setContentsMargins(20, 20, 20, 20)
layout.setSpacing(12)
button = _ReceptionCompleteButton(host)
neighbor = QPushButton("通知医助", host)
neighbor.setObjectName("ReceptionNotifyButton")
layout.addWidget(button)
layout.addWidget(neighbor)
host.show()
host.activateWindow()
button.clearFocus()
neighbor.clearFocus()
QTest.mouseMove(host, QPoint(1, 1))
app.processEvents()
yield app, host, button, neighbor
host.close()
host.deleteLater()
app.processEvents()
app.setStyle(old_style)
app.setFont(old_font)
app.setPalette(old_palette)
app.setStyleSheet(old_stylesheet)
def surface_color(button: QPushButton, *, border: bool = False) -> str:
image = button.grab().toImage()
scale = image.devicePixelRatio()
return image.pixelColor(
round((0 if border else 7) * scale), round(button.height() / 2 * scale)
).name()
def test_completion_hover_press_and_leave_have_feedback_without_layout_shift(controls):
app, host, button, neighbor = controls
geometry, neighbor_geometry = button.geometry(), neighbor.geometry()
clicked: list[bool] = []
button.clicked.connect(lambda: clicked.append(True))
assert surface_color(button) == "#fff7f8"
assert button.cursor().shape() == Qt.CursorShape.PointingHandCursor
QTest.mouseMove(button, button.rect().center())
app.processEvents()
assert button.underMouse()
assert surface_color(button) == "#fff0f2"
assert button.geometry() == geometry
assert neighbor.geometry() == neighbor_geometry
QTest.mousePress(button, Qt.MouseButton.LeftButton, pos=button.rect().center())
app.processEvents()
assert surface_color(button) == "#ffe4e8"
assert button.geometry() == geometry
assert neighbor.geometry() == neighbor_geometry
# Dragging outside and releasing must not activate the completion action.
QTest.mouseMove(button, QPoint(-5, -5))
QTest.mouseRelease(button, Qt.MouseButton.LeftButton, pos=QPoint(-5, -5))
QTest.mouseMove(host, QPoint(1, 1))
button.clearFocus()
app.processEvents()
assert clicked == []
assert surface_color(button) == "#fff7f8"
assert button.geometry() == geometry
assert neighbor.geometry() == neighbor_geometry
def test_completion_disabled_hover_does_not_look_or_act_enabled(controls):
app, _host, button, _neighbor = controls
clicked: list[bool] = []
button.clicked.connect(lambda: clicked.append(True))
button.setEnabled(False)
QTest.mouseMove(button, button.rect().center())
app.processEvents()
assert surface_color(button) == "#f7f8fb"
assert button.cursor().shape() == Qt.CursorShape.ArrowCursor
QTest.mouseClick(button, Qt.MouseButton.LeftButton)
assert clicked == []
button.setEnabled(True)
app.processEvents()
assert button.cursor().shape() == Qt.CursorShape.PointingHandCursor
assert surface_color(button) != "#f7f8fb"
def test_completion_keyboard_focus_is_visible_without_resizing(controls):
app, _host, button, neighbor = controls
geometry, neighbor_geometry = button.geometry(), neighbor.geometry()
neighbor.setFocus(Qt.FocusReason.TabFocusReason)
app.processEvents()
border_before = surface_color(button, border=True)
button.setFocus(Qt.FocusReason.TabFocusReason)
app.processEvents()
assert button.hasFocus()
assert surface_color(button, border=True) == "#cf4656"
assert surface_color(button, border=True) != border_before
assert button.geometry() == geometry
assert neighbor.geometry() == neighbor_geometry
QTest.mouseMove(button, button.rect().center())
app.processEvents()
assert surface_color(button) == "#fff0f2"
assert surface_color(button, border=True) == "#cf4656"
+1 -1
View File
@@ -2579,7 +2579,7 @@ def test_completion_revalidates_server_status_before_write(
assert completed == []
repository.status = 4
assert page._complete_after_revalidation(51) == {"ok": True}
assert page._complete_after_revalidation(51) == ""
assert completed == [51]
page.close()
application.processEvents()
+413
View File
@@ -0,0 +1,413 @@
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 page.refresh_button.text() == "刷新"
assert page.refresh_button.width() >= 50
assert page.refresh_button.geometry().right() < page.queue_date_button.geometry().left()
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)