新增
This commit is contained in:
@@ -0,0 +1,339 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user