560 lines
20 KiB
Python
560 lines
20 KiB
Python
"""Behaviour tests for mutable demo data and tolerant model parsing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from doctor_workstation.core.errors import (
|
|
ApiBusinessError,
|
|
ApiProtocolError,
|
|
AuthenticationExpiredError,
|
|
RepositoryNotFoundError,
|
|
)
|
|
from doctor_workstation.core.models import Appointment, PageResult
|
|
from doctor_workstation.services.mock_repository import DemoDoctorRepository
|
|
from doctor_workstation.services.repository import RemoteDoctorRepository
|
|
from doctor_workstation.services.token_store import TokenStore
|
|
|
|
|
|
@pytest.fixture
|
|
def repository() -> DemoDoctorRepository:
|
|
"""Return a fresh deterministic repository for each test."""
|
|
|
|
return DemoDoctorRepository(today=date(2026, 8, 10))
|
|
|
|
|
|
def test_demo_login_has_all_doctor_permissions(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Documented credentials produce a typed, globally authorised session."""
|
|
|
|
with pytest.raises(ApiBusinessError):
|
|
repository.login("doctor", "wrong")
|
|
|
|
session = repository.login("doctor", "doctor123")
|
|
assert session.authenticated
|
|
assert session.user.name == "陈医生（演示）"
|
|
assert session.permissions.is_superuser
|
|
assert session.permissions.can("doctor.appointment", "complete")
|
|
assert session.permissions.can("tcm.prescriptionLibrary", "delete")
|
|
|
|
|
|
def test_complete_appointment_mutates_all_related_views(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Completing reception is observable in queue, patient and diagnosis lists."""
|
|
|
|
repository.complete_appointment(101)
|
|
|
|
completed = repository.list_appointments(status=3).items
|
|
assert [item.id for item in completed] == [101, 104]
|
|
patient = repository.list_patients(keyword="林晓岚").items[0]
|
|
assert patient.appointment_status == 3
|
|
assert patient.status_filter == "completed"
|
|
consultation = repository.list_consultations(patient_name="林晓岚").items[0]
|
|
assert consultation.status == 3
|
|
assert repository.get_reception(101)["appointment"]["status"] == 3
|
|
|
|
|
|
def test_add_note_persists_in_reception_detail(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Text and media added to a diagnosis remain available on later reads."""
|
|
|
|
before = len(repository.get_reception(101)["doctor_notes"])
|
|
created = repository.add_doctor_note(
|
|
501,
|
|
"午后睡意减轻",
|
|
tongue_images=["demo://tongue.png"],
|
|
report_files=["demo://report.pdf"],
|
|
)
|
|
|
|
after = repository.get_reception(101)["doctor_notes"]
|
|
assert len(after) == before + 1
|
|
assert after[-1] == created
|
|
assert after[-1]["tongue_images"] == ["demo://tongue.png"]
|
|
|
|
|
|
def test_demo_upload_material_returns_safe_uri_and_note_rejects_local_path(
|
|
repository: DemoDoctorRepository,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Demo mode exercises the same upload-before-note contract as production."""
|
|
|
|
source = tmp_path / "tongue.jpg"
|
|
source.write_bytes(b"demo-image")
|
|
uri = repository.upload_material(source, "image")
|
|
created = repository.add_doctor_note(501, tongue_images=[uri])
|
|
|
|
assert uri.startswith("/demo/uploads/image/")
|
|
assert str(tmp_path) not in uri
|
|
assert created["tongue_images"] == [uri]
|
|
with pytest.raises(ValueError, match="server uri/url"):
|
|
repository.add_doctor_note(501, tongue_images=[str(source)])
|
|
|
|
|
|
def test_prescription_template_crud_is_real_and_isolated(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Create, update and delete change subsequent list/detail reads."""
|
|
|
|
original_count = repository.list_prescription_templates().total
|
|
created = repository.create_prescription_template(
|
|
name="益气演示方",
|
|
formula_type=1,
|
|
herbs=[{"name": "黄芪", "dosage": "20g"}],
|
|
is_public=False,
|
|
)
|
|
assert repository.list_prescription_templates().total == original_count + 1
|
|
|
|
updated = repository.update_prescription_template(
|
|
created.id,
|
|
{"name": "益气健脾演示方", "is_public": True},
|
|
)
|
|
assert updated.name == "益气健脾演示方"
|
|
assert updated.is_public
|
|
assert repository.get_prescription_template(created.id).name == updated.name
|
|
|
|
repository.delete_prescription_template(created.id)
|
|
assert repository.list_prescription_templates().total == original_count
|
|
with pytest.raises(RepositoryNotFoundError):
|
|
repository.get_prescription_template(created.id)
|
|
|
|
|
|
def test_demo_pagination_and_returned_copies(repository: DemoDoctorRepository) -> None:
|
|
"""Pagination metadata is stable and callers cannot mutate repository state."""
|
|
|
|
page = repository.list_appointments(page_no=1, page_size=1)
|
|
assert page.total == 5
|
|
assert page.pages == 5
|
|
page.items[0].patient_name = "外部改写"
|
|
assert repository.list_appointments(page_no=1, page_size=1).items[0].patient_name != (
|
|
"外部改写"
|
|
)
|
|
|
|
|
|
def test_demo_prescription_lookup_is_appointment_authoritative(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Demo getByAppointment never substitutes another diagnosis-level record."""
|
|
|
|
first = repository.get_prescription_by_appointment(101)
|
|
second = repository.get_prescription_by_appointment(102)
|
|
missing = repository.get_prescription_by_appointment(103)
|
|
|
|
assert first is not None and first.id == 802 and first.appointment_id == 101
|
|
assert second is not None and second.id == 801 and second.appointment_id == 102
|
|
assert missing is None
|
|
assert first.case_record["appointment_id"] == 101
|
|
|
|
|
|
def test_demo_consultation_filters_and_dictionaries_cover_exposed_ui(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Every visible consultation filter has deterministic offline semantics."""
|
|
|
|
for dictionary_type in (
|
|
"diagnosis_type",
|
|
"consultation_type",
|
|
"syndrome_type",
|
|
"appointment_channel_source",
|
|
"channels",
|
|
):
|
|
assert repository.get_dictionary(dictionary_type)
|
|
|
|
assert {row.id for row in repository.list_consultations(diagnosis_confirmed="1").items} == {
|
|
501,
|
|
503,
|
|
}
|
|
assert {row.id for row in repository.list_consultations(diagnosis_type="integrated").items} == {
|
|
502,
|
|
504,
|
|
}
|
|
assert [row.id for row in repository.list_consultations(syndrome_type="phlegm_damp").items] == [
|
|
502,
|
|
504,
|
|
]
|
|
assert [
|
|
row.id
|
|
for row in repository.list_consultations(latest_appointment_channel_source="clinic").items
|
|
] == [502]
|
|
assert [row.id for row in repository.list_consultations(pending_booking="1").items] == [504]
|
|
assert [row.id for row in repository.list_consultations(pending_assign="1").items] == [504]
|
|
assert [row.id for row in repository.list_consultations(completed_appointment="1").items] == [
|
|
501
|
|
]
|
|
sorted_rows = repository.list_consultations(sort_unserved_days="desc").items
|
|
assert [row.unserved_days for row in sorted_rows] == [14, 8, 1, 0]
|
|
|
|
|
|
def test_demo_roster_and_slots_fail_closed_to_known_doctor(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Offline booking uses the same roster/slot boundary as production."""
|
|
|
|
rosters = repository.list_appointment_rosters(
|
|
doctor_id=1001,
|
|
start_date="2026-08-10",
|
|
end_date="2026-08-16",
|
|
)
|
|
slots = repository.get_available_appointment_slots(
|
|
doctor_id=1001,
|
|
appointment_date="2026-08-10",
|
|
)
|
|
|
|
assert rosters.total == 7
|
|
assert rosters.items[0]["date"] == "2026-08-10"
|
|
assert any(row["time"] == "09:00" and not row["available"] for row in slots["slots"])
|
|
assert (
|
|
repository.get_available_appointment_slots(
|
|
doctor_id=9999,
|
|
appointment_date="2026-08-10",
|
|
)["slots"]
|
|
== []
|
|
)
|
|
|
|
|
|
def test_demo_call_lifecycle_mutates_record(repository: DemoDoctorRepository) -> None:
|
|
"""Start, room binding and end operations share one mutable call record."""
|
|
|
|
ticket = repository.get_call_ticket(301, 501)
|
|
assert ticket.patient_user_id == "patient_301"
|
|
started = repository.start_call(501, 301)
|
|
assert started["status"] == "ringing"
|
|
bound = repository.bind_call_room(501, "room-501")
|
|
assert bound["room_id"] == "room-501"
|
|
ended = repository.end_call(501)
|
|
assert ended["status"] == "ended"
|
|
assert ended["room_id"] == "room-501"
|
|
|
|
|
|
def test_demo_transcript_upsert_and_finish_round_trip_in_call_records(
|
|
repository: DemoDoctorRepository,
|
|
) -> None:
|
|
"""Demo replay reads expose one finalized segment for a repeated segment ID."""
|
|
|
|
started = repository.start_call(501, 301)
|
|
call_record_id = started["id"]
|
|
repository.start_call_transcription(501, call_record_id, "session-1")
|
|
repository.upsert_call_transcript_segments(
|
|
501,
|
|
call_record_id,
|
|
"session-1",
|
|
[
|
|
{
|
|
"segment_id": "seg-1",
|
|
"speaker_user_id": "patient_301",
|
|
"speaker_role": "patient",
|
|
"timestamp": 1200,
|
|
"text": "draft words",
|
|
}
|
|
],
|
|
)
|
|
repository.upsert_call_transcript_segments(
|
|
501,
|
|
call_record_id,
|
|
"session-1",
|
|
[
|
|
{
|
|
"segment_id": "seg-1",
|
|
"speaker_user_id": "patient_301",
|
|
"speaker_role": "patient",
|
|
"timestamp": 1200,
|
|
"text": "final words",
|
|
}
|
|
],
|
|
)
|
|
repository.finish_call_transcription(
|
|
501,
|
|
call_record_id,
|
|
"session-1",
|
|
expected_segment_count=1,
|
|
status="completed",
|
|
)
|
|
repository.end_call(501)
|
|
|
|
record = next(
|
|
row for row in repository.list_call_records(501) if row["id"] == call_record_id
|
|
)
|
|
assert record["status"] == 2
|
|
assert record["transcription_status"] == "completed"
|
|
assert record["transcription_segment_count"] == 1
|
|
assert record["transcript_segments"] == [
|
|
{
|
|
"segment_id": "seg-1",
|
|
"speaker_role": "patient",
|
|
"speaker_user_id": "patient_301",
|
|
"timestamp": 1200,
|
|
"text": "final words",
|
|
}
|
|
]
|
|
assert "final words" in record["transcript_text"]
|
|
|
|
|
|
def test_tolerant_page_parsing_accepts_aliases_and_bad_rows() -> None:
|
|
"""List parsing handles nullable fields, aliases and non-object rows safely."""
|
|
|
|
page = PageResult.from_payload(
|
|
{
|
|
"rows": [
|
|
{
|
|
"appointment_id": "9",
|
|
"patient_name": "测试患者",
|
|
"status": "waiting",
|
|
},
|
|
None,
|
|
],
|
|
"total": "12",
|
|
"current_page": "2",
|
|
"per_page": "5",
|
|
"meta": {"scope": "demo"},
|
|
},
|
|
Appointment.from_dict,
|
|
)
|
|
|
|
assert len(page.items) == 1
|
|
assert page.items[0].id == 9
|
|
assert page.items[0].status == "waiting"
|
|
assert page.total == 12
|
|
assert page.page_no == 2
|
|
assert page.extend == {"scope": "demo"}
|
|
|
|
|
|
class _StubApiClient:
|
|
"""No-network API client double that records repository endpoint use."""
|
|
|
|
def __init__(self) -> None:
|
|
self.base_url = "https://example.test/adminapi/"
|
|
self.token = ""
|
|
self.get_calls: list[tuple[str, dict[str, Any]]] = []
|
|
self.post_calls: list[tuple[str, dict[str, Any]]] = []
|
|
|
|
def set_token(self, token: str) -> None:
|
|
"""Retain the synthetic login token."""
|
|
|
|
self.token = token
|
|
|
|
def clear_token(self) -> None:
|
|
"""Clear the synthetic login token."""
|
|
|
|
self.token = ""
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
"""Return a shape appropriate for the requested read endpoint."""
|
|
|
|
self.get_calls.append((endpoint, dict(params or {})))
|
|
if endpoint == "auth.admin/mySelf":
|
|
return {
|
|
"user": {"id": 1, "name": "远程医生", "role_ids": [1]},
|
|
"permissions": ["doctor.appointment/lists"],
|
|
"menu": [],
|
|
}
|
|
if endpoint.endswith("/detail"):
|
|
if endpoint.startswith("tcm.prescriptionLibrary"):
|
|
return {"id": 7, "prescription_name": "远程模板", "herbs": []}
|
|
if endpoint.startswith("tcm.prescription"):
|
|
return {"id": 8, "sn": "RX8", "patient_name": "远程患者"}
|
|
if endpoint == "doctor.appointment/reception":
|
|
return {"appointment": {"id": params["id"]}, "doctor_notes": []}
|
|
return {"lists": [], "count": 0, "extend": {}}
|
|
|
|
def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any:
|
|
"""Return synthetic mutation data and record the exact JSON payload."""
|
|
|
|
body = dict(payload or {})
|
|
self.post_calls.append((endpoint, body))
|
|
if endpoint == "login/account":
|
|
return {"token": "remote-token", "is_paw": 1}
|
|
if endpoint == "tcm.prescriptionLibrary/add":
|
|
return {"id": 9}
|
|
if endpoint == "tcm.diagnosis/getCallSignature":
|
|
return {
|
|
"sdkAppId": 123,
|
|
"userId": "doctor_1",
|
|
"userSig": "short-lived",
|
|
"patientUserId": "patient_2",
|
|
}
|
|
if endpoint == "tcm.diagnosis/startCall":
|
|
return {"call_record_id": 901}
|
|
return {"ok": True}
|
|
|
|
|
|
def test_remote_repository_uses_all_confirmed_endpoints_without_network() -> None:
|
|
"""Every required remote operation maps to its audited admin endpoint."""
|
|
|
|
client = _StubApiClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
session = repository.login(" doctor ", "secret")
|
|
assert session.token == "remote-token"
|
|
assert repository.get_current_user().name == "远程医生"
|
|
assert [path for path, _ in client.get_calls].count("auth.admin/mySelf") == 1
|
|
|
|
repository.list_appointments(keyword="林", page_no=2, page_size=15)
|
|
repository.get_reception(5)
|
|
repository.notify_assistant(5)
|
|
repository.add_doctor_note(6, "记录")
|
|
repository.complete_appointment(5)
|
|
repository.list_prescription_templates(keyword="方", formula_type="aux")
|
|
repository.get_prescription_template(7)
|
|
repository.create_prescription_template(
|
|
name="新方", formula_type="main", herbs=[{"name": "茯苓", "dosage": "10g"}]
|
|
)
|
|
repository.update_prescription_template(7, {"name": "改方"})
|
|
repository.delete_prescription_template(7)
|
|
repository.list_prescriptions(keyword="RX8", status=1)
|
|
repository.get_prescription(8)
|
|
repository.list_patients(status="completed")
|
|
repository.list_consultations(keyword="远程")
|
|
ticket = repository.get_call_ticket(2, 6)
|
|
repository.start_call(6, 2)
|
|
repository.end_call(6)
|
|
repository.bind_call_room(6, "room-6")
|
|
|
|
assert ticket.user_sig == "short-lived"
|
|
get_endpoints = {path for path, _ in client.get_calls}
|
|
assert {
|
|
"doctor.appointment/lists",
|
|
"doctor.appointment/reception",
|
|
"tcm.prescriptionLibrary/lists",
|
|
"tcm.prescriptionLibrary/detail",
|
|
"tcm.prescription/lists",
|
|
"tcm.prescription/detail",
|
|
"firstvisit.myPatient/lists",
|
|
"tcm.diagnosis/lists",
|
|
} <= get_endpoints
|
|
post_endpoints = {path for path, _ in client.post_calls}
|
|
assert {
|
|
"login/account",
|
|
"doctor.appointment/notifyAssistant",
|
|
"doctor.appointment/addDoctorNote",
|
|
"doctor.appointment/complete",
|
|
"tcm.prescriptionLibrary/add",
|
|
"tcm.prescriptionLibrary/edit",
|
|
"tcm.prescriptionLibrary/delete",
|
|
"tcm.diagnosis/getCallSignature",
|
|
"tcm.diagnosis/startCall",
|
|
"tcm.diagnosis/endCall",
|
|
"tcm.diagnosis/bindCallRoom",
|
|
} <= post_endpoints
|
|
template_list_call = next(
|
|
params
|
|
for endpoint, params in client.get_calls
|
|
if endpoint == "tcm.prescriptionLibrary/lists"
|
|
)
|
|
assert template_list_call["formula_type"] == "辅方"
|
|
prescription_list_call = next(
|
|
params for endpoint, params in client.get_calls if endpoint == "tcm.prescription/lists"
|
|
)
|
|
assert prescription_list_call == {
|
|
"sn": "RX8",
|
|
"audit_filter": "passed",
|
|
"page_no": 1,
|
|
"page_size": 20,
|
|
}
|
|
patient_call = next(
|
|
params for endpoint, params in client.get_calls if endpoint == "firstvisit.myPatient/lists"
|
|
)
|
|
assert patient_call["status_filter"] == "completed"
|
|
|
|
|
|
class _FailingProfileClient(_StubApiClient):
|
|
"""Client double whose post-login session validation always fails."""
|
|
|
|
def __init__(self, error: Exception) -> None:
|
|
super().__init__()
|
|
self.error = error
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
"""Raise the configured error only for the authoritative profile call."""
|
|
|
|
if endpoint == "auth.admin/mySelf":
|
|
self.get_calls.append((endpoint, dict(params or {})))
|
|
raise self.error
|
|
return super().get(endpoint, params)
|
|
|
|
|
|
def test_remote_login_persists_only_after_session_validation(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A failed ``mySelf`` call rolls back memory and never saves its token."""
|
|
|
|
client = _FailingProfileClient(ApiProtocolError("bad profile"))
|
|
store = TokenStore(tmp_path / "credentials.json", keyring_backend=None)
|
|
store.save_token("older-token", account="older", scope=client.base_url)
|
|
saved: list[tuple[str, dict[str, Any]]] = []
|
|
original_save = store.save_token
|
|
|
|
def record_save(token: str, **metadata: Any) -> None:
|
|
saved.append((token, metadata))
|
|
original_save(token, **metadata)
|
|
|
|
monkeypatch.setattr(store, "save_token", record_save)
|
|
repository = RemoteDoctorRepository(client, store) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ApiProtocolError, match="bad profile"):
|
|
repository.login(
|
|
"doctor",
|
|
"secret",
|
|
remember_account=True,
|
|
)
|
|
|
|
assert saved == []
|
|
assert client.token == ""
|
|
assert store.load_token() is None
|
|
|
|
|
|
def test_remote_login_applies_remember_account_to_token_store(tmp_path: Path) -> None:
|
|
"""The checkbox choice controls account metadata while retaining the token."""
|
|
|
|
client = _StubApiClient()
|
|
store = TokenStore(tmp_path / "credentials.json", keyring_backend=None)
|
|
repository = RemoteDoctorRepository(client, store) # type: ignore[arg-type]
|
|
|
|
repository.login("doctor", "secret", remember_account=True)
|
|
assert store.load_account() == "doctor"
|
|
assert store.load_token(scope=client.base_url) == "remote-token"
|
|
|
|
repository.logout()
|
|
repository.login("doctor", "secret", remember_account=False)
|
|
assert store.load_account() is None
|
|
assert store.load_token(scope=client.base_url) == "remote-token"
|
|
|
|
|
|
def test_expired_persisted_token_is_removed_during_restore(tmp_path: Path) -> None:
|
|
"""An invalid startup token cannot trigger the same failed restore next run."""
|
|
|
|
client = _FailingProfileClient(AuthenticationExpiredError("expired", code=-1))
|
|
store = TokenStore(tmp_path / "credentials.json", keyring_backend=None)
|
|
store.save_token("expired-token", account="doctor", scope=client.base_url)
|
|
repository = RemoteDoctorRepository(client, store) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(AuthenticationExpiredError):
|
|
repository.restore_session()
|
|
|
|
assert client.token == ""
|
|
assert store.load_token() is None
|
|
assert store.load_account() == "doctor"
|
|
|
|
|
|
def test_restore_never_sends_token_to_a_different_api_scope(tmp_path: Path) -> None:
|
|
"""Changing the configured server invalidates automatic token reuse."""
|
|
|
|
client = _StubApiClient()
|
|
store = TokenStore(tmp_path / "credentials.json", keyring_backend=None)
|
|
store.save_token(
|
|
"other-server-token",
|
|
account="doctor",
|
|
scope="https://other.test/adminapi/",
|
|
)
|
|
repository = RemoteDoctorRepository(client, store) # type: ignore[arg-type]
|
|
|
|
assert repository.restore_session() is None
|
|
assert client.token == ""
|
|
assert client.get_calls == []
|