406 lines
13 KiB
Python
406 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtCore import QDate
|
|
from PySide6.QtWidgets import QApplication, QDialog
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.ui.dialogs import diagnosis as diagnosis_module
|
|
from doctor_workstation.ui.pages import consultations as consultations_module
|
|
from doctor_workstation.ui.pages.consultations import (
|
|
ConsultationsPage,
|
|
_video_payload,
|
|
appointment_rows,
|
|
is_diagnosis_confirmed,
|
|
is_video_available,
|
|
prescription_action_label,
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
@pytest.fixture
|
|
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def run_immediately(
|
|
function: Any,
|
|
*args: Any,
|
|
on_success: Any = None,
|
|
on_error: Any = None,
|
|
on_finished: Any = None,
|
|
**kwargs: Any,
|
|
) -> object:
|
|
try:
|
|
result = function(*args, **kwargs)
|
|
except Exception as error:
|
|
if on_error:
|
|
on_error(error)
|
|
else:
|
|
if on_success:
|
|
on_success(result)
|
|
finally:
|
|
if on_finished:
|
|
on_finished()
|
|
return object()
|
|
|
|
monkeypatch.setattr(consultations_module, "run_async", run_immediately)
|
|
monkeypatch.setattr(diagnosis_module, "run_async", run_immediately)
|
|
|
|
|
|
def _row(**changes: Any) -> dict[str, Any]:
|
|
row = {
|
|
"id": 501,
|
|
"diagnosis_id": 501,
|
|
"patient_id": 301,
|
|
"patient_name": "林晓岚",
|
|
"gender": 2,
|
|
"age": 36,
|
|
"status": 4,
|
|
"has_appointment": 1,
|
|
"appointment_id": 101,
|
|
"appointment_status": 1,
|
|
"appointments": [
|
|
{
|
|
"id": 101,
|
|
"status": 1,
|
|
"doctor_name": "陈医生",
|
|
"appointment_date": "2026-08-10",
|
|
"time_text": "09:00",
|
|
}
|
|
],
|
|
"DiagnosisViewRecord": [{"is_confirmed": 1}],
|
|
"has_prescription": 0,
|
|
}
|
|
row.update(changes)
|
|
return row
|
|
|
|
|
|
def test_video_condition_never_uses_diagnosis_status_or_missed_status() -> None:
|
|
assert is_video_available(_row(status=4, appointment_status=1))
|
|
assert not is_video_available(_row(status=1, appointment_status=4))
|
|
assert not is_video_available(_row(status=1, appointment_status=1, has_appointment=0))
|
|
|
|
payload = _video_payload(_row(id=777, diagnosis_id=777, appointment_id=222))
|
|
assert payload["appointment_id"] == 222
|
|
assert payload["diagnosis_id"] == 777
|
|
assert payload["patient_id"] == 301
|
|
|
|
|
|
def test_nested_appointments_confirmation_and_prescription_labels() -> None:
|
|
row = _row(
|
|
appointment_id=None,
|
|
appointment_status=None,
|
|
DiagnosisViewRecord=[{"is_confirmed": 0}, {"is_confirmed": "1"}],
|
|
)
|
|
assert appointment_rows(row)[0]["id"] == 101
|
|
assert is_diagnosis_confirmed(row)
|
|
assert (
|
|
prescription_action_label({"prescription_audit_status": 1, "prescription_void_status": 0})
|
|
== "查看处方"
|
|
)
|
|
assert (
|
|
prescription_action_label({"prescription_audit_status": 1, "prescription_void_status": 1})
|
|
== "开方"
|
|
)
|
|
|
|
|
|
def test_default_query_matches_admin_today_and_page_size_contract(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
calls: list[dict[str, Any]] = []
|
|
|
|
class Repository:
|
|
def list_consultations(self, **kwargs: Any) -> dict[str, Any]:
|
|
calls.append(kwargs)
|
|
return {"lists": [], "count": 0}
|
|
|
|
page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"]))
|
|
page.refresh(silent=True)
|
|
|
|
assert len(calls) == 1
|
|
query = calls[0]
|
|
assert query["page_no"] == 1
|
|
assert query["page_size"] == 15
|
|
assert query["appointment_date"] == QDate.currentDate().toString("yyyy-MM-dd")
|
|
assert "status" not in query
|
|
assert query["has_appointment"] == ""
|
|
assert query["diagnosis_confirmed"] == ""
|
|
assert {
|
|
"diagnosis_type",
|
|
"syndrome_type",
|
|
"assistant_id",
|
|
"latest_appointment_start_date",
|
|
"latest_appointment_end_date",
|
|
"latest_appointment_channel_source",
|
|
"latest_assign_start_date",
|
|
"latest_assign_end_date",
|
|
"sort_unserved_days",
|
|
}.issubset(query)
|
|
assert "consultation_type" not in query
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_double_click_opens_readonly_and_never_emits_video(
|
|
application: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
page = ConsultationsPage(
|
|
SimpleNamespace(),
|
|
permissions=PermissionSet(["tcm.diagnosis/readonlyDetail", "tcm.diagnosis/videoQr"]),
|
|
)
|
|
page.table.set_rows([_row()])
|
|
page.table.selectRow(0)
|
|
opened: list[tuple[int, bool]] = []
|
|
videos: list[dict[str, Any]] = []
|
|
monkeypatch.setattr(
|
|
page._diagnosis_dialog,
|
|
"open_for",
|
|
lambda diagnosis_id, *, editable=False, seed=None: opened.append((diagnosis_id, editable)),
|
|
)
|
|
page.video_requested.connect(videos.append)
|
|
|
|
page.table.itemDoubleClicked.emit(page.table.item(0, 0))
|
|
application.processEvents()
|
|
|
|
assert opened == [(501, False)]
|
|
assert videos == []
|
|
page.close()
|
|
|
|
|
|
def test_action_visibility_requires_exact_canonical_permissions(
|
|
application: QApplication,
|
|
) -> None:
|
|
aliases = PermissionSet(
|
|
[
|
|
"tcm.diagnosis.readonlyDetail",
|
|
"tcm.diagnosis.edit",
|
|
"tcm.diagnosis.add",
|
|
"tcm.diagnosis.delete",
|
|
]
|
|
)
|
|
page = ConsultationsPage(SimpleNamespace(), permissions=aliases)
|
|
assert not page.view_button.isVisible()
|
|
assert not page.edit_button.isVisible()
|
|
assert not page.add_button.isVisible()
|
|
assert not page.delete_button.isVisible()
|
|
page.close()
|
|
|
|
exact = PermissionSet(
|
|
[
|
|
"tcm.diagnosis/readonlyDetail",
|
|
"tcm.diagnosis/edit",
|
|
"tcm.diagnosis/add",
|
|
"tcm.diagnosis/delete",
|
|
]
|
|
)
|
|
page = ConsultationsPage(SimpleNamespace(), permissions=exact)
|
|
assert not page.view_button.isHidden()
|
|
assert not page.edit_button.isHidden()
|
|
assert not page.add_button.isHidden()
|
|
assert not page.delete_button.isHidden()
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_refresh_generation_ignores_late_results(
|
|
application: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
callbacks: list[dict[str, Any]] = []
|
|
|
|
def queue_async(_function: Any, **options: Any) -> object:
|
|
callbacks.append(options)
|
|
return object()
|
|
|
|
monkeypatch.setattr(consultations_module, "run_async", queue_async)
|
|
page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"]))
|
|
page.refresh(silent=True)
|
|
page.refresh(silent=True)
|
|
|
|
callbacks[1]["on_success"]({"lists": [_row(id=902, diagnosis_id=902)], "count": 1})
|
|
callbacks[0]["on_success"]({"lists": [_row(id=901, diagnosis_id=901)], "count": 1})
|
|
application.processEvents()
|
|
|
|
assert page.table.rowCount() == 1
|
|
assert page.table.item(0, 0).text().startswith("902")
|
|
page.close()
|
|
|
|
|
|
def test_current_appointment_is_the_only_prescription_authority(
|
|
application: QApplication,
|
|
) -> None:
|
|
calls: list[tuple[str, int]] = []
|
|
|
|
class Repository:
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> None:
|
|
calls.append(("appointment", appointment_id))
|
|
return None
|
|
|
|
def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}")
|
|
|
|
page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"]))
|
|
assert page._load_context_prescription(_row(appointment_id=202)) is None
|
|
assert calls == [("appointment", 202)]
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_empty_appointment_wrapper_is_treated_as_a_new_prescription(
|
|
application: QApplication,
|
|
) -> None:
|
|
class Repository:
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> dict[str, Any]:
|
|
assert appointment_id == 202
|
|
return {"data": {}}
|
|
|
|
page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"]))
|
|
assert page._load_context_prescription(_row(appointment_id=202)) is None
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_prescription_query_error_is_fail_closed_without_diagnosis_fallback(
|
|
application: QApplication,
|
|
) -> None:
|
|
class Repository:
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> None:
|
|
raise RuntimeError(f"appointment {appointment_id} unavailable")
|
|
|
|
def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}")
|
|
|
|
page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"]))
|
|
with pytest.raises(RuntimeError, match="appointment 202 unavailable"):
|
|
page._load_context_prescription(_row(appointment_id=202))
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_new_prescription_uses_authoritative_case_snapshot_and_exact_ids(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
calls: list[tuple[str, int]] = []
|
|
created: list[dict[str, Any]] = []
|
|
dialog_seeds: list[dict[str, Any]] = []
|
|
case_record = {
|
|
"diagnosis": {"id": 501, "patient_name": "林晓岚", "chief_complaint": "咳嗽"},
|
|
"patient": {"id": 301, "gender": 2, "age": 36},
|
|
}
|
|
|
|
class Repository:
|
|
def get_prescription_by_appointment(self, appointment_id: int) -> None:
|
|
calls.append(("appointment", appointment_id))
|
|
return None
|
|
|
|
def list_prescriptions_by_diagnosis(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
raise AssertionError(f"diagnosis fallback is forbidden: {diagnosis_id}")
|
|
|
|
def get_diagnosis_detail(self, diagnosis_id: int, **_kwargs: Any) -> dict[str, Any]:
|
|
calls.append(("diagnosis_detail", diagnosis_id))
|
|
return case_record
|
|
|
|
def create_prescription(self, prescription: Any) -> dict[str, Any]:
|
|
created.append(dict(prescription))
|
|
return {"id": 901}
|
|
|
|
def list_consultations(self, **_kwargs: Any) -> dict[str, Any]:
|
|
return {"lists": [], "count": 0}
|
|
|
|
class AcceptedEditor:
|
|
def __init__(
|
|
self,
|
|
_repository: Any,
|
|
seed: dict[str, Any],
|
|
**_kwargs: Any,
|
|
) -> None:
|
|
dialog_seeds.append(seed)
|
|
|
|
def exec(self) -> QDialog.DialogCode:
|
|
return QDialog.DialogCode.Accepted
|
|
|
|
def payload(self) -> dict[str, Any]:
|
|
return {"formula_name": "止咳方", "medicines": []}
|
|
|
|
monkeypatch.setattr(consultations_module, "PrescriptionEditorDialog", AcceptedEditor)
|
|
page = ConsultationsPage(Repository(), permissions=PermissionSet(["*"]))
|
|
page._begin_prescription_load(_row(appointment_id=202), mode="open")
|
|
|
|
assert calls[:2] == [("appointment", 202), ("diagnosis_detail", 501)]
|
|
assert len(created) == 1
|
|
assert created[0]["diagnosis_id"] == 501
|
|
assert created[0]["appointment_id"] == 202
|
|
assert created[0]["case_record"] == case_record
|
|
assert created[0]["case_record"] is not case_record
|
|
assert dialog_seeds[0]["case_record"] == case_record
|
|
assert dialog_seeds[0]["case_record"] is not case_record
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_switching_rows_invalidates_prescription_worker_and_clears_busy(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet(["*"]))
|
|
callbacks: list[dict[str, Any]] = []
|
|
|
|
def queue_async(_function: Any, **options: Any) -> object:
|
|
callbacks.append(options)
|
|
return object()
|
|
|
|
monkeypatch.setattr(consultations_module, "run_async", queue_async)
|
|
page.table.set_rows(
|
|
[
|
|
_row(id=501, diagnosis_id=501, appointment_id=101),
|
|
_row(id=502, diagnosis_id=502, appointment_id=202),
|
|
]
|
|
)
|
|
page.table.selectRow(0)
|
|
page._begin_prescription_load(page.table.current_data(), mode="open")
|
|
assert page._prescription_busy
|
|
assert not page.prescription_button.isEnabled()
|
|
|
|
page.table.selectRow(1)
|
|
application.processEvents()
|
|
assert not page._prescription_busy
|
|
assert page.prescription_button.isEnabled()
|
|
|
|
callbacks[0]["on_success"]({"id": 88, "appointment_id": 101})
|
|
callbacks[0]["on_finished"]()
|
|
assert page.table.current_data()["appointment_id"] == 202
|
|
assert not page._prescription_busy
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_native_call_does_not_reuse_video_qr_permission(
|
|
application: QApplication,
|
|
) -> None:
|
|
page = ConsultationsPage(SimpleNamespace(), permissions=PermissionSet([]))
|
|
emitted: list[dict[str, Any]] = []
|
|
page.video_requested.connect(emitted.append)
|
|
page.table.set_rows([_row()])
|
|
page.table.selectRow(0)
|
|
application.processEvents()
|
|
|
|
assert not page.video_button.isHidden()
|
|
assert page.video_button.isEnabled()
|
|
page._request_video()
|
|
assert emitted == [_video_payload(_row())]
|
|
page.close()
|
|
application.processEvents()
|