"""No-network contracts for the audited five doctor workspaces.""" 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, ApiHttpError, 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_reception_daily_records_use_admin_endpoints_exactly() -> None: client = RecordingClient() repository = RemoteDoctorRepository(client) # type: ignore[arg-type] repository.list_appointments( status=1, start_date="2026-08-11", end_date="2026-08-17", include_status_counts=1, page_no=1, page_size=15, ) repository.get_reception(71) repository.get_tracking_window( 271, start_date="2026-08-11", end_date="2026-08-17", ) repository.list_tracking_notes(271) assert client.get_calls[-4:] == [ ( "doctor.appointment/lists", { "status": 1, "start_date": "2026-08-11", "end_date": "2026-08-17", "include_status_counts": 1, "page_no": 1, "page_size": 15, }, ), ("doctor.appointment/reception", {"id": 71}), ( "tcm.diagnosis/trackingWindow", { "id": 271, "start_date": "2026-08-11", "end_date": "2026-08-17", }, ), ("tcm.diagnosis/trackingNotes", {"diagnosis_id": 271}), ] 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, "appointment_id": 0}, ) ] def test_remote_call_record_identity_is_reused_for_room_recording_and_end() -> None: """Room binding and COS finalization must never select a different latest call.""" client = RecordingClient() repository = RemoteDoctorRepository(client) # type: ignore[arg-type] repository.bind_call_room(501, " room-901 ", call_record_id=901) repository.end_call(501, call_record_id=901) assert client.post_calls == [ ( "tcm.diagnosis/bindCallRoom", {"diagnosis_id": 501, "room_id": "room-901", "call_record_id": 901}, ), ( "tcm.diagnosis/endCall", {"diagnosis_id": 501, "call_record_id": 901}, ), ] def test_remote_local_audio_upload_keeps_exact_call_identity_and_mime( tmp_path: Path, ) -> None: class MultipartClient(RecordingClient): def __init__(self) -> None: super().__init__() self.multipart_calls: list[ tuple[str, dict[str, tuple[str, bytes, str]], dict[str, Any]] ] = [] def post_multipart( self, endpoint: str, *, files: dict[str, tuple[str, bytes, str]], data: dict[str, Any], ) -> dict[str, Any]: self.multipart_calls.append((endpoint, files, data)) completed = int(data["chunk_index"]) == int(data["chunk_total"]) - 1 return { "call_record_id": 901, "completed": completed, "file_url": "https://cos.example.test/calls/local-audio.webm" if completed else "", "media_kind": "local_audio", } recording = tmp_path / "local-audio.webm" recording.write_bytes(b"a" * (4 * 1024 * 1024 + 3)) client = MultipartClient() repository = RemoteDoctorRepository(client) # type: ignore[arg-type] result = repository.upload_call_recording( recording, 501, call_record_id=901, mime_type="audio/webm;codecs=opus", ) assert result["completed"] is True assert result["call_record_id"] == 901 assert result["media_kind"] == "local_audio" assert len(client.multipart_calls) == 2 assert all(call[0] == "tcm.diagnosis/uploadCallRecording" for call in client.multipart_calls) for _endpoint, files, data in client.multipart_calls: assert data["diagnosis_id"] == 501 assert data["call_record_id"] == 901 assert data["mime_type"] == "audio/webm;codecs=opus" assert str(data["upload_id"]).startswith("local_audio_") assert files["file"][0] == "local-audio.webm" assert files["file"][2] == "audio/webm;codecs=opus" @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_ai_post_type_error_after_dispatch_is_never_retried() -> None: class FailingAfterDispatchClient: token = "token" def __init__(self) -> None: self.calls = 0 def post( self, endpoint: str, payload: dict[str, Any] | None = None, *, timeout: float | None = None, ) -> Any: assert endpoint == "tcm.diagnosis/generatePatientAiReport" assert payload == {"patient_id": 301, "model": "qwen"} assert timeout == 105.0 self.calls += 1 raise TypeError("transport failed after dispatch") client = FailingAfterDispatchClient() with pytest.raises(TypeError, match="after dispatch"): RemoteDoctorRepository(client).generate_patient_ai_report(301, model="qwen") assert client.calls == 1 def test_remote_diagnosis_ai_stream_normalises_chunks_in_order() -> None: class StreamingClient(RecordingClient): def post_event_stream(self, endpoint: str, payload: dict[str, Any], **kwargs: Any): assert endpoint == "tcm.diagnosis/aiAssistantStream" assert payload == {"id": 501, "prompt": "请辨证", "task": "tcm_pattern"} assert kwargs["timeout"] == 105.0 yield {"event": "start", "data": {"model_key": "qwen"}} yield {"event": "delta", "data": {"content": "肝郁"}} yield {"event": "delta", "data": {"delta": "脾虚"}} yield {"event": "done", "data": {"model_label": "千问"}} client = StreamingClient() events = list( RemoteDoctorRepository(client).stream_diagnosis_ai( 501, "请辨证", task="tcm_pattern", ) ) assert [event["event"] for event in events] == ["start", "delta", "delta", "done"] assert "".join(event.get("text", "") for event in events) == "肝郁脾虚" assert client.post_calls == [] def test_remote_diagnosis_ai_stream_falls_back_once_but_not_for_error_event() -> None: class MissingStreamClient(RecordingClient): def post_event_stream(self, *args: Any, **kwargs: Any): raise ApiHttpError("missing", status_code=404) missing_client = MissingStreamClient() events = list( RemoteDoctorRepository(missing_client).stream_diagnosis_ai(501, "请分析") ) assert [event["event"] for event in events] == ["start", "delta", "done"] assert events[1]["text"] == "服务端分析结果" assert [call[0] for call in missing_client.post_calls] == [ "tcm.diagnosis/aiAssistant" ] class ErrorStreamClient(RecordingClient): def post_event_stream(self, *args: Any, **kwargs: Any): yield {"event": "error", "data": {"message": "模型繁忙"}} error_client = ErrorStreamClient() with pytest.raises(ApiBusinessError, match="模型繁忙"): list(RemoteDoctorRepository(error_client).stream_diagnosis_ai(501, "请分析")) assert error_client.post_calls == [] 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")