"""Patient-level AI report contracts and reception history behaviour.""" from __future__ import annotations import os from datetime import date from typing import Any os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") import pytest from PySide6.QtWidgets import QApplication from doctor_workstation.core import PermissionSet from doctor_workstation.services.mock_repository import DemoDoctorRepository from doctor_workstation.services.repository import RemoteDoctorRepository from doctor_workstation.ui.pages import reception as reception_module from doctor_workstation.ui.pages.reception import ( AI_MEDICAL_DISCLAIMER, ReceptionPage, _generated_patient_report, _patient_report_rows, _ReceptionAiAnalysisDialog, ) @pytest.fixture(scope="module") def application() -> QApplication: return QApplication.instance() or QApplication([]) @pytest.fixture def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None: def run_immediately( function: Any, *args: Any, on_success: Any = None, on_error: Any = None, on_finished: Any = None, **kwargs: Any, ) -> object: try: result = function(*args, **kwargs) except Exception as error: if on_error: on_error(error) else: if on_success: on_success(result) finally: if on_finished: on_finished() return object() monkeypatch.setattr(reception_module, "run_async", run_immediately) def _detail(appointment_id: int, patient_id: int, diagnosis_id: int) -> dict[str, Any]: return { "appointment": { "id": appointment_id, "patient_id": patient_id, "patient_name": "快照患者", "status": 1, "appointment_date": date.today().isoformat(), }, "patient": {"id": patient_id, "patient_name": "快照患者", "age": 48}, "diagnosis": { "id": diagnosis_id, "patient_id": patient_id, "clinical_diagnosis": "气阴两虚证", }, } def _snapshot(model: str, version: int, stamp: str) -> dict[str, Any]: label = "OpenAI" if model == "openai" else "千问" return { "id": version * 10 + (2 if model == "openai" else 1), "patient_id": 301, "model_key": model, "model_label": label, "model_name": "gpt-demo" if model == "openai" else "qwen-demo", "version": version, "generated_at": stamp, "report": { "diagnosis": f"{label}第 {version} 版诊断建议", "risk_assessment": [{"label": "随访风险", "level": "low"}], "treatment_advice": f"{label}第 {version} 版治疗建议", "disclaimer": "服务端免责声明", }, } def test_remote_patient_report_contract_sends_only_patient_and_model() -> None: class Client: token = "token" def __init__(self) -> None: self.calls: list[tuple[str, str, dict[str, Any]]] = [] def get(self, endpoint: str, params: dict[str, Any], **_kwargs: Any) -> Any: self.calls.append(("get", endpoint, dict(params))) return {"patient_id": params["patient_id"], "reports": []} def post(self, endpoint: str, body: dict[str, Any], **_kwargs: Any) -> Any: self.calls.append(("post", endpoint, dict(body))) return {"patient_id": body["patient_id"], "reports": []} client = Client() repository = RemoteDoctorRepository(client) # type: ignore[arg-type] repository.list_patient_ai_reports(301) repository.generate_patient_ai_report(301, model="qwen") assert client.calls == [ ("get", "tcm.diagnosis/patientAiReports", {"patient_id": 301}), ( "post", "tcm.diagnosis/generatePatientAiReport", {"patient_id": 301, "model": "qwen"}, ), ] assert not ({"key", "api_key", "base_url", "provider"} & client.calls[-1][2].keys()) def test_demo_patient_history_has_two_versions_and_generation_appends() -> None: repository = DemoDoctorRepository() before = repository.list_patient_ai_reports(301) assert len(before["reports"]) == 4 assert [row["version"] for row in before["reports"] if row["model_key"] == "qwen"] == [2, 1] generated = repository.generate_patient_ai_report(301, model="qwen") assert "reports" not in generated assert "latest_by_model" not in generated assert generated["generated_report"]["version"] == 3 assert generated["generated_report"] == generated["report"] assert generated["disclaimer"] == AI_MEDICAL_DISCLAIMER assert generated["generated_report"]["disclaimer"] == AI_MEDICAL_DISCLAIMER assert isinstance(generated["source_summary"], dict) assert generated["source_summary"] == generated["generated_report"]["source_summary"] assert len(repository.list_patient_ai_reports(301)["reports"]) == 5 assert repository.list_patient_ai_reports(301)["disclaimer"] == AI_MEDICAL_DISCLAIMER def test_saved_history_is_rendered_without_automatic_generation( application: QApplication, immediate_async: None, ) -> None: detail = _detail(101, 301, 501) reports = [ _snapshot("qwen", 2, "2026-08-13 15:42:00"), _snapshot("openai", 2, "2026-08-13 15:43:00"), _snapshot("qwen", 1, "2026-08-12 09:18:00"), _snapshot("openai", 1, "2026-08-12 09:19:00"), ] class Repository: list_calls: list[int] = [] generate_calls: list[tuple[int, str]] = [] def get_reception(self, appointment_id: int) -> dict[str, Any]: assert appointment_id == 101 return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: self.list_calls.append(patient_id) return {"patient_id": patient_id, "reports": reports} def generate_patient_ai_report(self, patient_id: int, *, model: str) -> dict[str, Any]: self.generate_calls.append((patient_id, model)) raise AssertionError("saved history must not auto-generate") repository = Repository() page = ReceptionPage( repository, PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(detail["appointment"]) application.processEvents() assert repository.list_calls == [301] assert repository.generate_calls == [] assert page.ai_summary_label.text() == "千问第 2 版诊断建议" assert [len(page._ai_analysis_histories[key]) for key in ("qwen", "openai")] == [2, 2] assert "第 2 版" in page.ai_analysis_snapshot_meta.text() assert page.ai_analysis_disclaimer.text() == AI_MEDICAL_DISCLAIMER assert not page.ai_analysis_disclaimer.isVisibleTo(page) assert page.ai_analysis_history_button.objectName() == "ReceptionAiHistoryButton" assert page.ai_analysis_regenerate_button.objectName() == "ReceptionAiRegenerateButton" dialog = _ReceptionAiAnalysisDialog(page._ai_analysis_histories, preferred_model="qwen") assert dialog.history_selector.count() == 2 assert dialog.disclaimer_label.text() == AI_MEDICAL_DISCLAIMER dialog.history_selector.setCurrentIndex(1) assert "第 1 版诊断建议" in dialog.diagnosis_label.text() dialog.close() page.close() def test_empty_database_and_manual_refresh_append_qwen_then_openai( application: QApplication, immediate_async: None, ) -> None: detail = _detail(102, 302, 502) class Repository: def __init__(self) -> None: self.reports: list[dict[str, Any]] = [] self.calls: list[tuple[str, Any]] = [] def get_reception(self, appointment_id: int) -> dict[str, Any]: return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: self.calls.append(("list", patient_id)) return {"patient_id": patient_id, "reports": list(self.reports)} def generate_patient_ai_report(self, patient_id: int, *, model: str) -> dict[str, Any]: self.calls.append(("generate", model)) version = 1 + sum(row["model_key"] == model for row in self.reports) row = _snapshot(model, version, f"2026-08-14 10:0{len(self.reports)}:00") row["patient_id"] = patient_id self.reports.append(row) return { "patient_id": patient_id, "generated_report": row, "report": row, } repository = Repository() page = ReceptionPage( repository, PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(detail["appointment"]) application.processEvents() assert repository.calls == [("list", 302), ("generate", "qwen"), ("generate", "openai")] assert [len(page._ai_analysis_histories[key]) for key in ("qwen", "openai")] == [1, 1] page.ai_analysis_regenerate_button.click() application.processEvents() assert repository.calls[-2:] == [("generate", "qwen"), ("generate", "openai")] assert [len(page._ai_analysis_histories[key]) for key in ("qwen", "openai")] == [2, 2] page.close() def test_openai_failure_keeps_new_qwen_snapshot( application: QApplication, immediate_async: None, ) -> None: detail = _detail(103, 303, 503) class Repository: def get_reception(self, appointment_id: int) -> dict[str, Any]: return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: return {"patient_id": patient_id, "reports": []} def generate_patient_ai_report(self, patient_id: int, *, model: str) -> dict[str, Any]: if model == "openai": raise RuntimeError("OpenAI 暂时不可用") row = _snapshot("qwen", 1, "2026-08-14 10:30:00") row["patient_id"] = patient_id return { "patient_id": patient_id, "generated_report": row, "report": row, } page = ReceptionPage( Repository(), PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(detail["appointment"]) application.processEvents() assert page._ai_analysis_model_states["qwen"] == "success" assert page.ai_summary_label.text() == "千问第 1 版诊断建议" assert len(page._ai_analysis_histories["qwen"]) == 1 assert page._ai_analysis_model_states["openai"] == "error" assert "千问新快照已保留" in page.ai_analysis_secondary_status.text() page.close() def test_late_patient_history_response_is_discarded_after_switch( application: QApplication, monkeypatch: pytest.MonkeyPatch, ) -> None: first = _detail(104, 304, 504) second = _detail(105, 305, 505) jobs: list[dict[str, Any]] = [] def queue(function: Any, *args: Any, **options: Any) -> object: jobs.append({"function": function, "args": args, **options}) return object() monkeypatch.setattr(reception_module, "run_async", queue) class Repository: def get_reception(self, appointment_id: int) -> dict[str, Any]: return first if appointment_id == 104 else second def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: row = _snapshot("qwen", 1, f"2026-08-14 10:{patient_id - 300:02d}:00") row["patient_id"] = patient_id row["report"]["diagnosis"] = f"患者 {patient_id} 的报告" return {"patient_id": patient_id, "reports": [row]} def generate_patient_ai_report(self, patient_id: int, *, model: str) -> dict[str, Any]: raise AssertionError("history exists") def finish(job: dict[str, Any]) -> None: result = job["function"](*job.get("args", ())) if job.get("on_success"): job["on_success"](result) if job.get("on_finished"): job["on_finished"]() page = ReceptionPage( Repository(), PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(first["appointment"]) finish(jobs[0]) first_history_job = jobs[1] page._select_record(second["appointment"]) finish(jobs[2]) second_history_job = jobs[3] finish(second_history_job) assert page.ai_summary_label.text() == "患者 305 的报告" finish(first_history_job) assert page._ai_analysis_patient_id == 305 assert page.ai_summary_label.text() == "患者 305 的报告" page.close() def test_get_history_requires_exact_top_level_and_row_patient_ids() -> None: row = _snapshot("qwen", 1, "2026-08-14 11:00:00") valid = {"patient_id": 301, "reports": [row]} rows = _patient_report_rows(valid, expected_patient_id=301) assert rows is not None and len(rows) == 1 invalid_top_level_ids: tuple[Any, ...] = (None, 0, -1, True, "301", 302) for patient_id in invalid_top_level_ids: assert ( _patient_report_rows( {"patient_id": patient_id, "reports": [row]}, expected_patient_id=301, ) is None ) wrong_row = dict(row, patient_id=302) assert ( _patient_report_rows( {"patient_id": 301, "reports": [wrong_row]}, expected_patient_id=301, ) is None ) assert ( _patient_report_rows( { "patient_id": 302, "data": {"patient_id": 301, "reports": [row]}, }, expected_patient_id=301, ) is None ) def test_post_accepts_only_the_current_persisted_snapshot() -> None: valid = _snapshot("qwen", 3, "2026-08-14 11:05:00") accepted = _generated_patient_report( {"patient_id": 301, "generated_report": valid}, expected_patient_id=301, expected_model="qwen", ) assert accepted is not None and accepted["id"] == valid["id"] invalid_payloads = ( {"patient_id": 301, "reports": [valid], "report": valid}, {"patient_id": 301, "generated_report": {}, "reports": [valid]}, {"patient_id": 302, "generated_report": valid}, { "patient_id": 301, "generated_report": dict(valid, patient_id=302), }, {"patient_id": 301, "generated_report": dict(valid, id=0)}, {"patient_id": 301, "generated_report": dict(valid, id="31")}, { "patient_id": 301, "generated_report": dict(valid, model_key="openai"), }, ) for payload in invalid_payloads: assert ( _generated_patient_report( payload, expected_patient_id=301, expected_model="qwen", ) is None ) def test_stale_post_history_cannot_fake_generation_success( application: QApplication, immediate_async: None, ) -> None: detail = _detail(106, 306, 506) class Repository: def __init__(self) -> None: self.generate_calls: list[str] = [] def get_reception(self, appointment_id: int) -> dict[str, Any]: return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: return {"patient_id": patient_id, "reports": []} def generate_patient_ai_report( self, patient_id: int, *, model: str, ) -> dict[str, Any]: self.generate_calls.append(model) old = _snapshot("qwen", 9, "2026-08-13 08:00:00") old["patient_id"] = patient_id return { "patient_id": patient_id, "generated_report": None, "reports": [old], "report": old, } repository = Repository() page = ReceptionPage( repository, PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(detail["appointment"]) application.processEvents() assert repository.generate_calls == ["qwen"] assert not any(page._ai_analysis_histories.values()) assert page._ai_analysis_model_states["qwen"] == "error" page.close() def test_patient_report_generation_requires_read_and_generate_permissions( application: QApplication, immediate_async: None, ) -> None: detail = _detail(107, 307, 507) row = _snapshot("qwen", 1, "2026-08-14 11:10:00") row["patient_id"] = 307 class Repository: def __init__(self) -> None: self.list_calls = 0 self.generate_calls = 0 def get_reception(self, appointment_id: int) -> dict[str, Any]: return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: self.list_calls += 1 return {"patient_id": patient_id, "reports": [row]} def generate_patient_ai_report( self, patient_id: int, *, model: str, ) -> dict[str, Any]: self.generate_calls += 1 return {"patient_id": patient_id, "generated_report": row} cases = ( ([], False, 0), (["tcm.diagnosis/patientAiReports"], False, 1), (["tcm.diagnosis/generatePatientAiReport"], False, 0), (["tcm.diagnosis/aiAnalysis"], False, 0), ( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ], True, 1, ), ) for permissions, expected_enabled, expected_list_calls in cases: repository = Repository() page = ReceptionPage(repository, PermissionSet(permissions)) page._select_record(detail["appointment"]) application.processEvents() assert page.ai_analysis_regenerate_button.isEnabled() is expected_enabled assert repository.list_calls == expected_list_calls assert repository.generate_calls == 0 page.close() def test_ui_never_displays_internal_prompt_version( application: QApplication, immediate_async: None, ) -> None: detail = _detail(108, 308, 508) reports = [ _snapshot("qwen", 2, "2026-08-14 11:20:00"), _snapshot("qwen", 1, "2026-08-13 11:20:00"), ] for row in reports: row["patient_id"] = 308 row.pop("version") row["prompt_version"] = "patient-longitudinal-report-internal-v99" class Repository: def get_reception(self, appointment_id: int) -> dict[str, Any]: return detail def list_patient_ai_reports(self, patient_id: int) -> dict[str, Any]: return {"patient_id": patient_id, "reports": reports} def generate_patient_ai_report( self, patient_id: int, *, model: str, ) -> dict[str, Any]: raise AssertionError("saved history must not auto-generate") page = ReceptionPage( Repository(), PermissionSet( [ "tcm.diagnosis/patientAiReports", "tcm.diagnosis/generatePatientAiReport", ] ), ) page._select_record(detail["appointment"]) application.processEvents() assert page.ai_analysis_snapshot_meta.text().startswith("第 2 版") assert "internal-v99" not in page.ai_analysis_snapshot_meta.text() assert "internal-v99" not in page.ai_analysis_snapshot_meta.toolTip() dialog = _ReceptionAiAnalysisDialog(page._ai_analysis_histories, preferred_model="qwen") assert dialog.history_selector.itemText(0).startswith("第 2 版") assert dialog.history_selector.itemText(1).startswith("第 1 版") assert "internal-v99" not in dialog.meta_label.text() dialog.close() page.close() def test_patient_ai_disclaimer_remains_the_unified_text() -> None: assert AI_MEDICAL_DISCLAIMER == ( "仅供临床辅助参考，不可替代医生诊断，不得直接用于开方、用药调整或其他医疗决策。" "系统未对舌像、报告附件或视频画面进行视觉诊断；仅分析已录入、归档或转写的文字及附件元数据。" )