129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
"""Repository contracts for privacy-safe AI patient diagnosis options."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from inspect import signature
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from doctor_workstation.services.mock_repository import DemoDoctorRepository
|
|
from doctor_workstation.services.repository import DoctorRepository, RemoteDoctorRepository
|
|
|
|
|
|
class AiPatientOptionsClient:
|
|
"""Record the exact AI option request and return deliberately unsafe extras."""
|
|
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
self.calls.append((endpoint, dict(params or {})))
|
|
return {
|
|
"lists": [
|
|
{
|
|
"diagnosis_id": 9001,
|
|
"source_patient_id": 42,
|
|
"patient_name": "测试患者",
|
|
"phone_masked": "13800138000",
|
|
"phone": "13800138000",
|
|
"id_card": "110101199001011234",
|
|
"gender": 2,
|
|
"age": 36,
|
|
"diagnosis_date": "2026-08-19",
|
|
"diagnosis_summary": "随访诊单",
|
|
"last_visit_at": "2026-08-19 09:30:00",
|
|
"next_appointment_at": "2026-08-26 09:30:00",
|
|
}
|
|
],
|
|
"count": 1,
|
|
}
|
|
|
|
|
|
def test_protocol_exposes_ai_patient_option_page_defaults() -> None:
|
|
method = signature(DoctorRepository.list_ai_patient_options)
|
|
|
|
assert method.parameters["page_no"].default == 1
|
|
assert method.parameters["page_size"].default == 20
|
|
assert method.parameters["keyword"].default == ""
|
|
|
|
|
|
def test_remote_ai_patient_options_use_exact_endpoint_params_and_safe_dto() -> None:
|
|
client = AiPatientOptionsClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
page = repository.list_ai_patient_options(
|
|
page_no=3,
|
|
page_size=7,
|
|
keyword=" 测试 ",
|
|
)
|
|
|
|
assert client.calls == [
|
|
(
|
|
"tcm.diagnosis/aiPatientOptions",
|
|
{"keyword": "测试", "page_no": 3, "page_size": 7},
|
|
)
|
|
]
|
|
assert page.total == 1
|
|
assert page.page_no == 3
|
|
assert page.page_size == 7
|
|
assert page.items == [
|
|
{
|
|
"diagnosis_id": 9001,
|
|
"source_patient_id": 42,
|
|
"patient_name": "测试患者",
|
|
"phone_masked": "138****8000",
|
|
"gender": 2,
|
|
"age": 36,
|
|
"diagnosis_date": "2026-08-19",
|
|
"diagnosis_summary": "随访诊单",
|
|
"last_visit_at": "2026-08-19 09:30:00",
|
|
"next_appointment_at": "2026-08-26 09:30:00",
|
|
}
|
|
]
|
|
assert page.items[0]["diagnosis_id"] != page.items[0]["source_patient_id"]
|
|
assert "phone" not in page.items[0]
|
|
assert "id_card" not in page.items[0]
|
|
|
|
|
|
@pytest.mark.parametrize(("page_no", "page_size"), [(0, 20), (1, 0)])
|
|
def test_remote_ai_patient_options_reject_invalid_pagination_before_get(
|
|
page_no: int,
|
|
page_size: int,
|
|
) -> None:
|
|
client = AiPatientOptionsClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ValueError, match="must be positive"):
|
|
repository.list_ai_patient_options(page_no=page_no, page_size=page_size)
|
|
|
|
assert client.calls == []
|
|
|
|
|
|
def test_demo_ai_patient_options_are_stable_searchable_paginated_and_private() -> None:
|
|
repository = DemoDoctorRepository(today=date(2026, 8, 20))
|
|
|
|
first = repository.list_ai_patient_options(page_no=1, page_size=2)
|
|
second = repository.list_ai_patient_options(page_no=2, page_size=2)
|
|
repeated = repository.list_ai_patient_options(page_no=1, page_size=2)
|
|
|
|
assert first.total == 4
|
|
assert first.pages == 2
|
|
assert [row["diagnosis_id"] for row in first.items] == [504, 503]
|
|
assert [row["diagnosis_id"] for row in second.items] == [502, 501]
|
|
assert repeated.items == first.items
|
|
assert all(row["diagnosis_id"] != row["source_patient_id"] for row in first.items)
|
|
assert all("****" in row["phone_masked"] for row in first.items)
|
|
assert all("phone" not in row and "id_card" not in row for row in first.items)
|
|
assert "13700006618" not in repr(first.items)
|
|
|
|
by_diagnosis = repository.list_ai_patient_options(keyword=" 503 ")
|
|
by_plain_phone = repository.list_ai_patient_options(keyword="15900007732")
|
|
|
|
assert [row["diagnosis_id"] for row in by_diagnosis.items] == [503]
|
|
assert [row["diagnosis_id"] for row in by_plain_phone.items] == [503]
|
|
assert by_plain_phone.items[0]["phone_masked"] == "159****7732"
|
|
assert "15900007732" not in repr(by_plain_phone.items)
|
|
|