761 lines
28 KiB
Python
761 lines
28 KiB
Python
"""No-network contracts for the audited five doctor workspaces."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from doctor_workstation.core.errors import ApiProtocolError
|
|
from doctor_workstation.core.models import Appointment, Consultation, PageResult, Prescription
|
|
from doctor_workstation.services.mock_repository import DemoDoctorRepository
|
|
from doctor_workstation.services.repository import (
|
|
DIAGNOSIS_AI_PERMISSIONS,
|
|
PRESCRIPTION_LIBRARY_PERMISSIONS,
|
|
PRESCRIPTION_PERMISSIONS,
|
|
RemoteDoctorRepository,
|
|
)
|
|
|
|
|
|
class RecordingClient:
|
|
"""Small API client double retaining exact endpoint and DTO calls."""
|
|
|
|
def __init__(self) -> None:
|
|
self.base_url = "https://example.test/adminapi/"
|
|
self.token = "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:
|
|
"""Set the current synthetic token."""
|
|
|
|
self.token = token
|
|
|
|
def clear_token(self) -> None:
|
|
"""Clear the current synthetic token."""
|
|
|
|
self.token = ""
|
|
|
|
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
|
|
"""Record a GET and return a shape appropriate for its contract."""
|
|
|
|
self.get_calls.append((endpoint, dict(params or {})))
|
|
if endpoint == "auth.admin/mySelf":
|
|
return {
|
|
"user": {"id": 1, "name": "Doctor", "role_ids": [1]},
|
|
"permissions": ["tcm.diagnosis/lists"],
|
|
"menu": [
|
|
{
|
|
"name": "问诊列表",
|
|
"component": "tcm/diagnosis/index",
|
|
"future": {"badge": 3},
|
|
"children": [{"name": "只读", "perms": "tcm.diagnosis/readonlyDetail"}],
|
|
"unsafe": lambda: None,
|
|
7: "non-string key",
|
|
}
|
|
],
|
|
}
|
|
if endpoint in {
|
|
"tcm.diagnosis/detail",
|
|
"tcm.diagnosis/readonlyDetail",
|
|
"firstvisit.myPatient/orderDetail",
|
|
"tcm.prescriptionOrder/detail",
|
|
}:
|
|
return {"id": int((params or {}).get("id", 0)), "patient_name": "测试患者"}
|
|
if endpoint == "tcm.prescription/getByAppointment":
|
|
return {}
|
|
if endpoint == "tcm.prescriptionLibrary/aiReports":
|
|
return {
|
|
"prescription_id": int((params or {}).get("id", 0)),
|
|
"prescription_name": "疏肝健脾基础方",
|
|
"formula_type": "主方",
|
|
"reports": [],
|
|
"missing_model_keys": ["qwen", "openai"],
|
|
"can_view": True,
|
|
"can_refresh": True,
|
|
"can_edit": True,
|
|
"capabilities": {"can_view": True, "can_refresh": True, "can_edit": True},
|
|
}
|
|
if endpoint == "tcm.diagnosis/aiReports":
|
|
return {
|
|
"diagnosis_id": int((params or {}).get("id", 0)),
|
|
"patient_name": "林晓岚",
|
|
"case_summary": "临床诊断：肝郁脾虚证",
|
|
"reports": [],
|
|
"missing_model_keys": ["qwen", "openai"],
|
|
"can_view": True,
|
|
"can_refresh": True,
|
|
"can_edit": True,
|
|
"capabilities": {"can_view": True, "can_refresh": True, "can_edit": True},
|
|
}
|
|
if endpoint == "doctor.appointment/availableSlots":
|
|
return {"slots": [{"time": "09:00", "available": True}]}
|
|
if endpoint == "tcm.prescriptionOrder/paidPayOrders":
|
|
return {"lists": [{"id": 9}], "deposit_min_amount": 50}
|
|
return {"lists": [], "count": 0, "extend": {"scope": {"label": "server"}}}
|
|
|
|
def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any:
|
|
"""Record a POST and return a stable synthetic mutation result."""
|
|
|
|
body = dict(payload or {})
|
|
self.post_calls.append((endpoint, body))
|
|
if endpoint in {"tcm.prescription/add", "tcm.prescriptionOrder/create"}:
|
|
return {"id": 88}
|
|
if endpoint == "tcm.diagnosis/startCall":
|
|
return {"call_record_id": 901}
|
|
if endpoint == "tcm.prescriptionLibrary/generateAiReports":
|
|
return {
|
|
"prescription_id": int(body.get("id", 0)),
|
|
"reports": [],
|
|
"can_refresh": True,
|
|
"can_edit": True,
|
|
"status": "success",
|
|
}
|
|
if endpoint == "tcm.prescriptionLibrary/editAiReport":
|
|
return {
|
|
"prescription_id": int(body.get("id", 0)),
|
|
"report": {
|
|
"report_id": int(body.get("report_id", 0)),
|
|
"model_key": "qwen",
|
|
"content": body.get("content", ""),
|
|
},
|
|
"can_edit": True,
|
|
"can_refresh": True,
|
|
}
|
|
if endpoint == "tcm.diagnosis/generateAiReports":
|
|
return {
|
|
"diagnosis_id": int(body.get("id", 0)),
|
|
"reports": [],
|
|
"can_refresh": True,
|
|
"can_edit": True,
|
|
"status": "success",
|
|
}
|
|
if endpoint == "tcm.diagnosis/editAiReport":
|
|
return {
|
|
"diagnosis_id": int(body.get("id", 0)),
|
|
"report": {
|
|
"report_id": int(body.get("report_id", 0)),
|
|
"model_key": "qwen",
|
|
"content": body.get("content", ""),
|
|
},
|
|
"can_edit": True,
|
|
"can_refresh": True,
|
|
}
|
|
if endpoint == "tcm.diagnosis/aiAssistant":
|
|
return {
|
|
"diagnosis_id": int(body.get("id", 0)),
|
|
"answer": "服务端分析结果",
|
|
"model_key": "qwen",
|
|
"task": body.get("task"),
|
|
}
|
|
if endpoint == "tcm.diagnosis/aiAnalysis":
|
|
model = str(body.get("model") or "")
|
|
if model == "openai":
|
|
return {
|
|
"diagnosis_advice": "2 型糖尿病，需结合客观检查复核",
|
|
"risk_assessment": [
|
|
{"label": "用药安全风险", "level": "medium"},
|
|
{"label": "肾功能风险", "level": "low"},
|
|
],
|
|
"treatment_advice": "复核近期检查趋势并评估联合用药安全性。",
|
|
"model_key": "openai",
|
|
"model_label": "OpenAI",
|
|
"model_name": "gpt-5.2",
|
|
"generated_at": "2026-08-14 10:31:00",
|
|
}
|
|
return {
|
|
"diagnosis_advice": "2 型糖尿病，血糖控制不佳",
|
|
"risk_assessment": [
|
|
{"label": "高血糖风险", "level": "high"},
|
|
{"label": "心血管风险", "level": "medium"},
|
|
],
|
|
"treatment_advice": "复核用药依从性并安排糖化血红蛋白检查。",
|
|
"model_key": "qwen",
|
|
"model_label": "千问",
|
|
"model_name": "qwen3.6-35b",
|
|
"generated_at": "2026-08-14 10:30:00",
|
|
}
|
|
return {"ok": True}
|
|
|
|
|
|
def test_page_result_preserves_outer_and_nested_extend() -> None:
|
|
"""Server scope metadata must survive nested data pagination envelopes."""
|
|
|
|
page = PageResult.from_payload(
|
|
{
|
|
"extend": {"scope": {"label": "doctor"}},
|
|
"data": {
|
|
"lists": [{"id": 1}],
|
|
"count": 4,
|
|
"extend": {"schedule_mode": "roster"},
|
|
},
|
|
},
|
|
Appointment.from_dict,
|
|
)
|
|
|
|
assert page.total == 4
|
|
assert page.extend == {
|
|
"scope": {"label": "doctor"},
|
|
"schedule_mode": "roster",
|
|
}
|
|
|
|
|
|
def test_consultation_keeps_diagnosis_and_appointment_status_separate() -> None:
|
|
"""Video eligibility fields cannot be overwritten by diagnosis enablement."""
|
|
|
|
row = Consultation.from_dict(
|
|
{
|
|
"id": 3,
|
|
"status": 1,
|
|
"status_desc": "启用",
|
|
"has_appointment": 1,
|
|
"appointment_status": 4,
|
|
"appointment_status_text": "已过号",
|
|
"appointments": [{"id": 8, "status": 4}],
|
|
"DiagnosisViewRecord": [{"is_confirmed": 1}],
|
|
}
|
|
)
|
|
|
|
assert row.status == 1
|
|
assert row.appointment_status == 4
|
|
assert row.has_appointment
|
|
assert row.confirmed
|
|
assert row.appointments == [{"id": 8, "status": 4}]
|
|
|
|
|
|
def test_prescription_round_trips_appointment_and_case_record() -> None:
|
|
"""Appointment authority and immutable case snapshot survive model DTOs."""
|
|
|
|
prescription = Prescription.from_dict(
|
|
{
|
|
"id": 8,
|
|
"diagnosis_id": 5,
|
|
"appointment_id": 17,
|
|
"case_record": {
|
|
"diagnosis_id": 5,
|
|
"appointment_id": 17,
|
|
"clinical_diagnosis": "气阴两虚证",
|
|
},
|
|
}
|
|
)
|
|
|
|
assert prescription.appointment_id == 17
|
|
assert prescription.case_record["clinical_diagnosis"] == "气阴两虚证"
|
|
payload = prescription.to_api_dict()
|
|
assert payload["appointment_id"] == 17
|
|
assert payload["case_record"] == {
|
|
"diagnosis_id": 5,
|
|
"appointment_id": 17,
|
|
"clinical_diagnosis": "气阴两虚证",
|
|
}
|
|
|
|
|
|
def test_remote_reception_is_forcibly_scoped_to_today() -> None:
|
|
"""Status 1/4 queue reads always carry the audited same-day date range."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
repository.list_appointments(status=1, keyword="林", page_no=2, page_size=15)
|
|
|
|
endpoint, params = client.get_calls[-1]
|
|
assert endpoint == "doctor.appointment/lists"
|
|
assert params == {
|
|
"status": 1,
|
|
"patient_name": "林",
|
|
"start_date": date.today().isoformat(),
|
|
"end_date": date.today().isoformat(),
|
|
"page_no": 2,
|
|
"page_size": 15,
|
|
}
|
|
|
|
|
|
def test_remote_new_contracts_use_exact_admin_endpoints_and_dtos() -> None:
|
|
"""Prescription, patient and diagnosis methods remain thin endpoint adapters."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
repository.create_prescription(
|
|
patient_name="测试患者",
|
|
diagnosis_id=5,
|
|
herbs=[{"medicine_id": 1, "name": "黄芪", "dosage": 10}],
|
|
)
|
|
repository.patch_prescription_patient(8, patient_name="修正患者", phone="13800000000", gender=2)
|
|
repository.audit_prescription(8, action="reject", remark="剂量不符")
|
|
repository.create_prescription_order(
|
|
prescription_id=8,
|
|
diagnosis_id=5,
|
|
recipient_name="修正患者",
|
|
recipient_phone="13800000000",
|
|
)
|
|
repository.list_paid_prescription_orders(5, prescription_order_id=3)
|
|
repository.list_medicines(name="黄芪")
|
|
repository.patient_orders(page_no=1, page_size=15, fulfillment_status=2)
|
|
repository.patient_progress(page_no=1, page_size=15, status=1)
|
|
repository.patient_detail(5)
|
|
repository.appointment_history(5)
|
|
repository.assign_history(5)
|
|
repository.assign_patient(5, 20, is_inherit=1)
|
|
repository.fill_patient_id_card(5, "410000199001010000")
|
|
repository.book_patient_appointment({"diagnosis_id": 5, "appointment_date": "2026-08-10"})
|
|
repository.cancel_patient_appointment(7)
|
|
repository.update_diagnosis(5, {"clinical_diagnosis": "气虚证"})
|
|
repository.list_appointment_rosters(
|
|
doctor_id=1,
|
|
start_date="2026-08-10",
|
|
end_date="2026-08-16",
|
|
)
|
|
slots = repository.get_available_appointment_slots(
|
|
doctor_id=1,
|
|
appointment_date="2026-08-10",
|
|
)
|
|
|
|
assert (
|
|
"tcm.prescription/patchPatient",
|
|
{
|
|
"id": 8,
|
|
"patient_name": "修正患者",
|
|
"phone": "13800000000",
|
|
"gender": 2,
|
|
},
|
|
) in client.post_calls
|
|
assert (
|
|
"tcm.prescription/audit",
|
|
{
|
|
"id": 8,
|
|
"action": "reject",
|
|
"remark": "剂量不符",
|
|
},
|
|
) in client.post_calls
|
|
assert (
|
|
"firstvisit.myPatient/assign",
|
|
{
|
|
"id": 5,
|
|
"assistant_id": 20,
|
|
"is_inherit": 1,
|
|
},
|
|
) in client.post_calls
|
|
assert ("tcm.diagnosis/edit", {"id": 5, "clinical_diagnosis": "气虚证"}) in client.post_calls
|
|
assert slots == {"slots": [{"time": "09:00", "available": True}]}
|
|
get_endpoints = {endpoint for endpoint, _ in client.get_calls}
|
|
assert {
|
|
"tcm.prescriptionOrder/paidPayOrders",
|
|
"doctor.medicine/lists",
|
|
"firstvisit.myPatient/orders",
|
|
"firstvisit.myPatient/progress",
|
|
"tcm.diagnosis/readonlyDetail",
|
|
"doctor.appointment/lists",
|
|
"tcm.diagnosis/assignLogList",
|
|
"doctor.roster/lists",
|
|
"doctor.appointment/availableSlots",
|
|
} <= get_endpoints
|
|
|
|
|
|
def test_remote_transcription_endpoints_use_exact_normalized_dtos() -> None:
|
|
"""Realtime transcript persistence stays within the three audited POST DTOs."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
repository.start_call_transcription(501, 901, " session-1 ", language=" zh-CN ")
|
|
repository.upsert_call_transcript_segments(
|
|
501,
|
|
901,
|
|
"session-1",
|
|
[
|
|
{
|
|
"segment_id": "seg-1",
|
|
"speaker_user_id": "patient_301",
|
|
"speaker_role": "patient",
|
|
"timestamp": "1200",
|
|
"text": " patient words ",
|
|
}
|
|
],
|
|
)
|
|
repository.finish_call_transcription(
|
|
501,
|
|
901,
|
|
"session-1",
|
|
expected_segment_count=1,
|
|
status="completed",
|
|
)
|
|
|
|
assert client.post_calls == [
|
|
(
|
|
"tcm.diagnosis/startCallTranscription",
|
|
{
|
|
"diagnosis_id": 501,
|
|
"call_record_id": 901,
|
|
"transcription_session_id": "session-1",
|
|
"language": "zh-CN",
|
|
},
|
|
),
|
|
(
|
|
"tcm.diagnosis/upsertCallTranscriptSegments",
|
|
{
|
|
"diagnosis_id": 501,
|
|
"call_record_id": 901,
|
|
"transcription_session_id": "session-1",
|
|
"segments": [
|
|
{
|
|
"segment_id": "seg-1",
|
|
"speaker_user_id": "patient_301",
|
|
"speaker_role": "patient",
|
|
"timestamp": 1200,
|
|
"text": "patient words",
|
|
}
|
|
],
|
|
},
|
|
),
|
|
(
|
|
"tcm.diagnosis/finishCallTranscription",
|
|
{
|
|
"diagnosis_id": 501,
|
|
"call_record_id": 901,
|
|
"transcription_session_id": "session-1",
|
|
"expected_segment_count": 1,
|
|
"status": "completed",
|
|
},
|
|
),
|
|
]
|
|
|
|
|
|
def test_remote_start_call_requires_and_normalizes_current_record_id() -> None:
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
assert repository.start_call(501, 301) == {"call_record_id": 901}
|
|
assert client.post_calls == [
|
|
(
|
|
"tcm.diagnosis/startCall",
|
|
{"diagnosis_id": 501, "patient_id": 301, "call_type": 2},
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response",
|
|
[None, {}, {"ok": True}, {"call_record_id": 0}, {"callRecordId": -1}, {"id": True}],
|
|
)
|
|
def test_remote_start_call_rejects_missing_or_invalid_record_id(response: Any) -> None:
|
|
class StartCallClient(RecordingClient):
|
|
def post(self, endpoint: str, payload: dict[str, Any] | None = None) -> Any:
|
|
if endpoint == "tcm.diagnosis/startCall":
|
|
self.post_calls.append((endpoint, dict(payload or {})))
|
|
return response
|
|
return super().post(endpoint, payload)
|
|
|
|
repository = RemoteDoctorRepository(StartCallClient()) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ApiProtocolError, match="call_record"):
|
|
repository.start_call(501, 301)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"unsafe_reference",
|
|
[r"C:\records\tongue.jpg", r"\\server\share\report.pdf", "file:///tmp/a.jpg"],
|
|
)
|
|
def test_remote_note_rejects_local_material_references(unsafe_reference: str) -> None:
|
|
"""No drive, UNC or file URI can reach addDoctorNote JSON."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
|
|
|
|
with pytest.raises(ValueError, match="server uri/url"):
|
|
repository.add_doctor_note(5, tongue_images=[unsafe_reference])
|
|
assert not any(
|
|
endpoint == "doctor.appointment/addDoctorNote" for endpoint, _payload in client.post_calls
|
|
)
|
|
|
|
|
|
def test_remote_dynamic_menu_preserves_json_metadata_but_drops_runtime_objects() -> None:
|
|
"""Future menu fields pass through safely without evaluating arbitrary values."""
|
|
|
|
repository = RemoteDoctorRepository(RecordingClient()) # type: ignore[arg-type]
|
|
|
|
menu = repository.get_session().menu
|
|
|
|
assert menu[0]["component"] == "tcm/diagnosis/index"
|
|
assert menu[0]["future"] == {"badge": 3}
|
|
assert menu[0]["children"][0]["perms"] == "tcm.diagnosis/readonlyDetail"
|
|
assert "unsafe" not in menu[0]
|
|
assert 7 not in menu[0]
|
|
|
|
|
|
def test_canonical_prescription_permissions_match_routed_views() -> None:
|
|
"""Service exports one canonical spelling for each routed action."""
|
|
|
|
assert PRESCRIPTION_LIBRARY_PERMISSIONS == {
|
|
"create": "wcf.prescription/add",
|
|
"read": "wcf.prescription/read",
|
|
"update": "wcf.prescription/edit",
|
|
"delete": "wcf.prescription/delete",
|
|
"ai_reports": "tcm.prescriptionLibrary/aiReports",
|
|
"generate_ai_reports": "tcm.prescriptionLibrary/generateAiReports",
|
|
"edit_ai_report": "tcm.prescriptionLibrary/editAiReport",
|
|
}
|
|
assert PRESCRIPTION_PERMISSIONS["delete"] == "cf.prescription/del"
|
|
assert PRESCRIPTION_PERMISSIONS["patch_patient"] == "tcm.prescription/patchPatient"
|
|
assert DIAGNOSIS_AI_PERMISSIONS == {
|
|
"ai_reports": "tcm.diagnosis/aiReports",
|
|
"generate_ai_reports": "tcm.diagnosis/generateAiReports",
|
|
"edit_ai_report": "tcm.diagnosis/editAiReport",
|
|
"analysis": "tcm.diagnosis/aiAnalysis",
|
|
"assistant": "tcm.diagnosis/aiAssistant",
|
|
}
|
|
|
|
|
|
def test_remote_prescription_library_ai_report_endpoints() -> None:
|
|
"""AI interpretation uses the same adminapi contract as the Vue library page."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client)
|
|
listed = repository.list_prescription_template_ai_reports(701)
|
|
generated = repository.generate_prescription_template_ai_reports(701)
|
|
edited = repository.edit_prescription_template_ai_report(
|
|
701, report_id=9, content='{"summary":"演示"}'
|
|
)
|
|
|
|
assert listed["prescription_id"] == 701
|
|
assert generated["status"] == "success"
|
|
assert edited["report"]["report_id"] == 9
|
|
assert client.get_calls[-1] == (
|
|
"tcm.prescriptionLibrary/aiReports",
|
|
{"id": 701},
|
|
)
|
|
assert (
|
|
"tcm.prescriptionLibrary/generateAiReports",
|
|
{"id": 701},
|
|
) in client.post_calls
|
|
assert (
|
|
"tcm.prescriptionLibrary/editAiReport",
|
|
{"id": 701, "report_id": 9, "content": '{"summary":"演示"}'},
|
|
) in client.post_calls
|
|
|
|
|
|
def test_remote_diagnosis_ai_report_endpoints() -> None:
|
|
"""Patient-profile AI reports use the diagnosis adminapi contract."""
|
|
|
|
client = RecordingClient()
|
|
repository = RemoteDoctorRepository(client)
|
|
listed = repository.list_diagnosis_ai_reports(501)
|
|
generated = repository.generate_diagnosis_ai_reports(501)
|
|
edited = repository.edit_diagnosis_ai_report(
|
|
501, report_id=9, content='{"summary":"演示"}'
|
|
)
|
|
|
|
assert listed["diagnosis_id"] == 501
|
|
assert generated["status"] == "success"
|
|
assert edited["report"]["report_id"] == 9
|
|
assert client.get_calls[-1] == (
|
|
"tcm.diagnosis/aiReports",
|
|
{"id": 501},
|
|
)
|
|
assert (
|
|
"tcm.diagnosis/generateAiReports",
|
|
{"id": 501},
|
|
) in client.post_calls
|
|
assert (
|
|
"tcm.diagnosis/editAiReport",
|
|
{"id": 501, "report_id": 9, "content": '{"summary":"演示"}'},
|
|
) in client.post_calls
|
|
|
|
|
|
def test_remote_diagnosis_ai_assistant_uses_first_party_endpoint_only() -> None:
|
|
class TimeoutRecordingClient(RecordingClient):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.timeouts: list[float | None] = []
|
|
|
|
def post(
|
|
self,
|
|
endpoint: str,
|
|
payload: dict[str, Any] | None = None,
|
|
*,
|
|
timeout: float | None = None,
|
|
) -> Any:
|
|
self.timeouts.append(timeout)
|
|
return super().post(endpoint, payload)
|
|
|
|
client = TimeoutRecordingClient()
|
|
repository = RemoteDoctorRepository(client)
|
|
|
|
result = repository.analyze_diagnosis_ai(
|
|
501,
|
|
"请给出用药调整建议",
|
|
task="medication_review",
|
|
)
|
|
|
|
assert result["answer"] == "服务端分析结果"
|
|
assert client.post_calls == [
|
|
(
|
|
"tcm.diagnosis/aiAssistant",
|
|
{"id": 501, "prompt": "请给出用药调整建议", "task": "medication_review"},
|
|
)
|
|
]
|
|
body = client.post_calls[0][1]
|
|
assert not ({"key", "api_key", "base_url", "provider", "model"} & body.keys())
|
|
assert client.timeouts == [105.0]
|
|
|
|
|
|
def test_remote_diagnosis_ai_analysis_uses_exact_post_contract() -> None:
|
|
"""The legacy default is qwen, followed by an explicit OpenAI request."""
|
|
|
|
class TimeoutRecordingClient(RecordingClient):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.timeouts: list[float | None] = []
|
|
|
|
def post(
|
|
self,
|
|
endpoint: str,
|
|
payload: dict[str, Any] | None = None,
|
|
*,
|
|
timeout: float | None = None,
|
|
) -> Any:
|
|
self.timeouts.append(timeout)
|
|
return super().post(endpoint, payload)
|
|
|
|
client = TimeoutRecordingClient()
|
|
repository = RemoteDoctorRepository(client)
|
|
|
|
qwen_result = repository.get_diagnosis_ai_analysis(501)
|
|
openai_result = repository.get_diagnosis_ai_analysis(501, model="openai")
|
|
|
|
assert client.post_calls == [
|
|
("tcm.diagnosis/aiAnalysis", {"id": 501, "model": "qwen"}),
|
|
("tcm.diagnosis/aiAnalysis", {"id": 501, "model": "openai"}),
|
|
]
|
|
assert client.timeouts == [105.0, 105.0]
|
|
assert qwen_result == {
|
|
"diagnosis_advice": "2 型糖尿病，血糖控制不佳",
|
|
"risk_assessment": [
|
|
{"label": "高血糖风险", "level": "high"},
|
|
{"label": "心血管风险", "level": "medium"},
|
|
],
|
|
"treatment_advice": "复核用药依从性并安排糖化血红蛋白检查。",
|
|
"model_key": "qwen",
|
|
"model_label": "千问",
|
|
"model_name": "qwen3.6-35b",
|
|
"generated_at": "2026-08-14 10:30:00",
|
|
}
|
|
assert openai_result == {
|
|
"diagnosis_advice": "2 型糖尿病，需结合客观检查复核",
|
|
"risk_assessment": [
|
|
{"label": "用药安全风险", "level": "medium"},
|
|
{"label": "肾功能风险", "level": "low"},
|
|
],
|
|
"treatment_advice": "复核近期检查趋势并评估联合用药安全性。",
|
|
"model_key": "openai",
|
|
"model_label": "OpenAI",
|
|
"model_name": "gpt-5.2",
|
|
"generated_at": "2026-08-14 10:31:00",
|
|
}
|
|
|
|
calls_before_validation = list(client.post_calls)
|
|
timeouts_before_validation = list(client.timeouts)
|
|
with pytest.raises(ValueError, match="qwen or openai"):
|
|
repository.get_diagnosis_ai_analysis(501, model="invalid") # type: ignore[arg-type]
|
|
with pytest.raises(ValueError, match="positive"):
|
|
repository.get_diagnosis_ai_analysis(0)
|
|
assert client.post_calls == calls_before_validation
|
|
assert client.timeouts == timeouts_before_validation
|
|
|
|
|
|
def test_demo_diagnosis_ai_analysis_matches_structured_contract() -> None:
|
|
repository = DemoDoctorRepository(today=date(2026, 8, 14))
|
|
|
|
qwen_result = repository.get_diagnosis_ai_analysis(501)
|
|
openai_result = repository.get_diagnosis_ai_analysis(501, model="openai")
|
|
|
|
expected_keys = {
|
|
"diagnosis_advice",
|
|
"risk_assessment",
|
|
"treatment_advice",
|
|
"model_key",
|
|
"model_label",
|
|
"model_name",
|
|
"generated_at",
|
|
}
|
|
for result in (qwen_result, openai_result):
|
|
assert set(result) == expected_keys
|
|
assert "肝郁脾虚证" in result["diagnosis_advice"]
|
|
assert result["treatment_advice"]
|
|
assert result["risk_assessment"]
|
|
assert all(
|
|
set(item) == {"label", "level"}
|
|
and item["level"] in {"high", "medium", "low"}
|
|
for item in result["risk_assessment"]
|
|
)
|
|
assert (qwen_result["model_key"], qwen_result["model_label"]) == ("qwen", "千问")
|
|
assert (openai_result["model_key"], openai_result["model_label"]) == (
|
|
"openai",
|
|
"OpenAI",
|
|
)
|
|
assert qwen_result["model_name"] != openai_result["model_name"]
|
|
assert qwen_result["diagnosis_advice"] != openai_result["diagnosis_advice"]
|
|
assert qwen_result["treatment_advice"] != openai_result["treatment_advice"]
|
|
|
|
|
|
def test_demo_mutates_prescriptions_orders_and_patient_workspaces() -> None:
|
|
"""Offline mode supports the full workflow rather than static placeholders."""
|
|
|
|
repository = DemoDoctorRepository(today=date(2026, 8, 10))
|
|
created = repository.create_prescription(
|
|
diagnosis_id=501,
|
|
patient_name="林晓岚",
|
|
phone="13800131203",
|
|
gender=2,
|
|
herbs=[{"medicine_id": 17, "name": "黄芪", "dosage": 20}],
|
|
doctor_name="陈医生（演示）",
|
|
doctor_signature="data:image/png;base64,demo",
|
|
)
|
|
updated = repository.update_prescription(created.id, {"clinical_diagnosis": "气虚证"})
|
|
repository.patch_prescription_patient(
|
|
created.id, patient_name="林晓岚（修正）", phone="13800131203", gender=2
|
|
)
|
|
audit = repository.audit_prescription(created.id, action="approve")
|
|
order = repository.create_prescription_order(
|
|
prescription_id=created.id,
|
|
diagnosis_id=501,
|
|
recipient_name="林晓岚（修正）",
|
|
recipient_phone="13800131203",
|
|
amount=268,
|
|
)
|
|
|
|
assert updated.clinical_diagnosis == "气虚证"
|
|
assert audit["audit_status"] == 1
|
|
assert repository.get_prescription(created.id).has_prescription_order
|
|
assert repository.get_prescription_order(order["id"])["prescription_id"] == created.id
|
|
assert repository.patient_orders().extend["summary"]["order_count"] == 2
|
|
|
|
assigned = repository.assign_patient(501, 2002)
|
|
repository.fill_patient_id_card(501, "410000199001010000")
|
|
appointment = repository.book_patient_appointment(
|
|
diagnosis_id=501,
|
|
appointment_date="2026-08-11",
|
|
appointment_time="15:00-15:30",
|
|
)
|
|
repository.update_diagnosis(501, {"chief_complaint": "乏力"})
|
|
|
|
assert assigned["assistant_name"] == "许医助"
|
|
assert repository.assign_history(501).total == 2
|
|
assert repository.patient_detail(501)["diagnosis"]["chief_complaint"] == "乏力"
|
|
assert repository.appointment_history(501).total == 2
|
|
repository.cancel_patient_appointment(appointment["id"])
|
|
assert repository.appointment_history(501).items[-1].status == 2
|
|
assert repository.list_medicines(name="黄芪").items[0]["name"] == "黄芪"
|
|
|
|
|
|
def test_reject_actions_require_a_remark() -> None:
|
|
"""Audit rejection mirrors the admin dialog's mandatory reason boundary."""
|
|
|
|
repository = DemoDoctorRepository(today=date(2026, 8, 10))
|
|
|
|
with pytest.raises(ValueError, match="remark"):
|
|
repository.audit_prescription(802, action="reject")
|
|
with pytest.raises(ValueError, match="remark"):
|
|
repository.audit_patient_order_payment(901, "reject")
|