307 lines
12 KiB
Python
307 lines
12 KiB
Python
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
|