599 lines
20 KiB
Python
599 lines
20 KiB
Python
"""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 == (
|
|
"仅供临床辅助参考，不可替代医生诊断，不得直接用于开方、用药调整或其他医疗决策。"
|
|
"系统未对舌像、报告附件或视频画面进行视觉诊断；仅分析已录入、归档或转写的文字及附件元数据。"
|
|
)
|