340 lines
11 KiB
Python
340 lines
11 KiB
Python
"""Parity contracts for the admin appointment list port."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtWidgets import QApplication, QLabel
|
|
|
|
from doctor_workstation.core.errors import ApiProtocolError
|
|
from doctor_workstation.core.models import Appointment, PageResult
|
|
from doctor_workstation.core.permissions import PermissionSet
|
|
from doctor_workstation.services.mock_repository import DemoDoctorRepository
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.ui.pages import appointments as appointments_module
|
|
from doctor_workstation.ui.pages.appointments import (
|
|
AppointmentsPage,
|
|
_diagnosis_id,
|
|
_video_patient_id,
|
|
prescription_action_label,
|
|
)
|
|
from doctor_workstation.ui.shell import _match_navigation
|
|
|
|
|
|
class RecordingClient:
|
|
def __init__(self) -> None:
|
|
self.get_calls: list[tuple[str, dict[str, Any]]] = []
|
|
self.post_calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
self.get_calls.append((endpoint, dict(params or {})))
|
|
if endpoint == "doctor.appointment/detail":
|
|
return {"id": int((params or {}).get("id", 0)), "patient_name": "测试"}
|
|
if endpoint == "dept.dept/all":
|
|
return [{"id": 10, "name": "中医门诊", "children": []}]
|
|
return {"lists": [], "count": 0, "extend": {"status_count": {"1": 2}}}
|
|
|
|
def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any:
|
|
self.post_calls.append((endpoint, dict(payload or {})))
|
|
if endpoint == "tcm.diagnosis/generateMiniProgramQrcode":
|
|
return {"qrcode_url": "https://example.test/uploads/video-qr.png"}
|
|
return {"ok": True}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def application() -> QApplication:
|
|
return QApplication.instance() or QApplication([])
|
|
|
|
|
|
def test_remote_appointment_list_and_detail_hit_admin_endpoints() -> None:
|
|
client = RecordingClient()
|
|
repo = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
repo.list_appointments(
|
|
page_no=1,
|
|
page_size=15,
|
|
status=1,
|
|
start_date="2026-08-11",
|
|
end_date="2026-08-11",
|
|
include_status_counts=1,
|
|
diagnosis_confirmed="1",
|
|
assistant_dept_id=10,
|
|
patient_name="林",
|
|
)
|
|
detail = repo.get_appointment_detail(101)
|
|
departments = repo.list_departments()
|
|
|
|
assert client.get_calls[0][0] == "doctor.appointment/lists"
|
|
assert client.get_calls[0][1]["include_status_counts"] == 1
|
|
assert client.get_calls[0][1]["assistant_dept_id"] == 10
|
|
assert "diag_scope_relax" not in client.get_calls[0][1]
|
|
assert client.get_calls[1] == ("doctor.appointment/detail", {"id": 101})
|
|
assert client.get_calls[2] == ("dept.dept/all", {})
|
|
assert detail["id"] == 101
|
|
assert departments[0]["id"] == 10
|
|
|
|
|
|
def test_unlimited_date_relaxes_today_scope_for_pending_status() -> None:
|
|
client = RecordingClient()
|
|
repo = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
repo.list_appointments(status=1, diag_scope_relax=1, page_no=1, page_size=15)
|
|
params = client.get_calls[0][1]
|
|
assert "start_date" not in params
|
|
assert "end_date" not in params
|
|
assert "diag_scope_relax" not in params
|
|
|
|
|
|
def test_video_qr_uses_doctor_id_without_fake_diagnosis_id() -> None:
|
|
client = RecordingClient()
|
|
repo = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
result = repo.generate_video_qrcode(
|
|
diagnosis_id=501,
|
|
doctor_id=88,
|
|
patient_id=301,
|
|
share_user_id=9,
|
|
)
|
|
|
|
endpoint, payload = client.post_calls[-1]
|
|
assert endpoint == "tcm.diagnosis/generateMiniProgramQrcode"
|
|
assert payload == {
|
|
"diagnosis_id": 501,
|
|
"doctor_id": 88,
|
|
"patient_id": 301,
|
|
"share_user_id": 9,
|
|
"mini_program_path": "pages/login/login",
|
|
}
|
|
assert result["qrcode_url"].endswith("video-qr.png")
|
|
|
|
|
|
def test_legacy_empty_string_prescription_response_is_scoped_to_not_found() -> None:
|
|
class LegacyClient(RecordingClient):
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
if endpoint == "tcm.prescription/getByAppointment":
|
|
raise ApiProtocolError("envelope must be an object", data="")
|
|
return super().get(endpoint, params)
|
|
|
|
repo = RemoteDoctorRepository(LegacyClient()) # type: ignore[arg-type]
|
|
assert repo.get_prescription_by_appointment(101) is None
|
|
|
|
class BrokenClient(LegacyClient):
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
raise ApiProtocolError("invalid JSON")
|
|
|
|
broken = RemoteDoctorRepository(BrokenClient()) # type: ignore[arg-type]
|
|
with pytest.raises(ApiProtocolError, match="invalid JSON"):
|
|
broken.get_prescription_by_appointment(101)
|
|
|
|
|
|
def test_shell_matches_appointment_list_by_route_not_shared_permission() -> None:
|
|
row = {
|
|
"name": "挂号列表",
|
|
"paths": "/appointments",
|
|
"component": "tcm/appointment/list",
|
|
"perms": "doctor.appointment/lists",
|
|
}
|
|
item = _match_navigation(row)
|
|
assert item is not None
|
|
assert item.key == "appointments"
|
|
|
|
reception = {
|
|
"name": "接诊台",
|
|
"paths": "/reception",
|
|
"component": "patient/reception/index",
|
|
"perms": "doctor.appointment/lists",
|
|
}
|
|
assert _match_navigation(reception).key == "reception" # type: ignore[union-attr]
|
|
|
|
|
|
def test_diagnosis_and_patient_ids_stay_distinct_for_video() -> None:
|
|
row = Appointment.from_dict(
|
|
{
|
|
"id": 101,
|
|
"patient_id": 301,
|
|
"diagnosis_id": 501,
|
|
"source_patient_id": 301,
|
|
"status": 1,
|
|
"has_prescription": 0,
|
|
}
|
|
)
|
|
assert _diagnosis_id(row) == 501
|
|
assert _video_patient_id(row) == 301
|
|
assert prescription_action_label(row) == "开方"
|
|
|
|
approved = Appointment.from_dict(
|
|
{
|
|
"id": 102,
|
|
"prescription_audit_status": 1,
|
|
"prescription_void_status": 0,
|
|
"has_prescription": 1,
|
|
}
|
|
)
|
|
assert prescription_action_label(approved) == "查看"
|
|
|
|
|
|
def test_appointments_page_default_query_is_today_pending(
|
|
application: QApplication,
|
|
) -> None:
|
|
repo = DemoDoctorRepository()
|
|
page = AppointmentsPage(
|
|
repo,
|
|
permissions=PermissionSet(
|
|
[
|
|
"doctor.appointment/lists",
|
|
"doctor.appointment/complete",
|
|
"doctor.appointment/cancel",
|
|
"doctor.appointment/prescription",
|
|
"doctor.appointment/addDoctorNote",
|
|
"tcm.diagnosis/edit",
|
|
"tcm.diagnosis/kaifang",
|
|
"tcm.diagnosis/videoQr",
|
|
]
|
|
),
|
|
current_user={"id": 1001, "name": "陈医生", "role_id": 1},
|
|
)
|
|
page.show()
|
|
application.processEvents()
|
|
page.refresh()
|
|
application.processEvents()
|
|
|
|
filters = page._query_filters()
|
|
assert filters["status"] == 1
|
|
assert filters["include_status_counts"] == 1
|
|
assert filters["start_date"] == filters["end_date"]
|
|
assert "diag_scope_relax" not in filters
|
|
assert page.table.rowCount() >= 1
|
|
|
|
|
|
def test_demo_appointment_status_counts_respect_date_scope() -> None:
|
|
repo = DemoDoctorRepository()
|
|
today = repo._today.isoformat()
|
|
result = repo.list_appointments(
|
|
page_no=1,
|
|
page_size=20,
|
|
start_date=today,
|
|
end_date=today,
|
|
include_status_counts=1,
|
|
)
|
|
assert isinstance(result, PageResult)
|
|
counts = result.extend.get("status_count", {})
|
|
assert int(counts.get("1", counts.get(1, 0))) >= 1
|
|
assert int(counts.get("3", counts.get(3, 0))) >= 1
|
|
|
|
|
|
def test_appointment_multiline_cells_receive_enough_row_height(
|
|
application: QApplication,
|
|
) -> None:
|
|
page = AppointmentsPage(
|
|
DemoDoctorRepository(),
|
|
permissions=PermissionSet(["doctor.appointment/lists"]),
|
|
current_user={"id": 1001, "role_id": 1},
|
|
)
|
|
page._loaded(
|
|
{
|
|
"lists": [
|
|
{
|
|
"id": 15534,
|
|
"patient_name": "张玉英",
|
|
"patient_phone": "13800138000",
|
|
"gender": 0,
|
|
"age": 42,
|
|
"height": 162,
|
|
"weight": 55,
|
|
"doctor_name": "徐国军",
|
|
"appointment_date": "2026-08-11",
|
|
"appointment_time": "14:30",
|
|
"assistant_name": "蒋露露",
|
|
"diagnosis_confirmed": 0,
|
|
"has_prescription": 0,
|
|
"status": 1,
|
|
"status_desc": "已预约",
|
|
"remark": "—",
|
|
}
|
|
],
|
|
"count": 1,
|
|
},
|
|
page._generation,
|
|
False,
|
|
)
|
|
|
|
patient_text = page.table.item(0, 1).text()
|
|
appointment_text = page.table.item(0, 3).text()
|
|
assert patient_text.count("\n") == 2
|
|
assert appointment_text == "2026-08-11\n14:30"
|
|
required = 3 * max(16, page.table.fontMetrics().lineSpacing()) + 10
|
|
assert page.table.rowHeight(0) >= required
|
|
assert page.table.item(0, 1).toolTip() == patient_text
|
|
assert page.table.item(0, 3).toolTip() == appointment_text
|
|
page.close()
|
|
|
|
|
|
def test_video_qr_dialog_renders_downloaded_image_inside_app(
|
|
application: QApplication,
|
|
) -> None:
|
|
repo = DemoDoctorRepository()
|
|
page = AppointmentsPage(
|
|
repo,
|
|
permissions=PermissionSet(["tcm.diagnosis/videoQr"]),
|
|
current_user={"id": 1001, "role_id": 1},
|
|
)
|
|
url = "https://demo.invalid/qrcode/video/88.png"
|
|
dialog = page._build_qr_dialog(
|
|
{"patient_name": "鹿立核"},
|
|
url,
|
|
{"qrcode_url": url, "_image_bytes": repo.download_public_image(url)},
|
|
)
|
|
|
|
image = dialog.findChild(QLabel, "VideoQrImage")
|
|
assert image is not None
|
|
assert image.pixmap() is not None and not image.pixmap().isNull()
|
|
assert dialog.findChild(QLabel, "VideoQrImage").text() == ""
|
|
dialog.close()
|
|
page.close()
|
|
application.processEvents()
|
|
|
|
|
|
def test_prescription_case_snapshot_uses_keyword_diagnosis_id(
|
|
application: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
requested: list[int] = []
|
|
opened: list[tuple[Any, Any]] = []
|
|
|
|
class Repository:
|
|
def get_diagnosis_detail(self, *, diagnosis_id: int) -> dict[str, Any]:
|
|
requested.append(diagnosis_id)
|
|
return {"id": diagnosis_id, "patient_name": "测试患者"}
|
|
|
|
def run_immediately(function: Any, **callbacks: Any) -> object:
|
|
try:
|
|
callbacks["on_success"](function())
|
|
except Exception as error: # pragma: no cover - assertion output is clearer
|
|
callbacks["on_error"](error)
|
|
finally:
|
|
callbacks["on_finished"]()
|
|
return object()
|
|
|
|
monkeypatch.setattr(appointments_module, "run_async", run_immediately)
|
|
page = AppointmentsPage(
|
|
Repository(),
|
|
permissions=PermissionSet(["tcm.diagnosis/kaifang"]),
|
|
current_user={"id": 9, "role_id": 1},
|
|
)
|
|
monkeypatch.setattr(
|
|
page,
|
|
"_open_prescription_editor",
|
|
lambda record, detail: opened.append((record, detail)),
|
|
)
|
|
|
|
page._begin_case_record_load(
|
|
{"id": 101, "appointment_id": 101, "diagnosis_id": 501, "patient_id": 501}
|
|
)
|
|
|
|
assert requested == [501]
|
|
assert opened and opened[0][1]["id"] == 501
|
|
page.close()
|
|
application.processEvents()
|