897 lines
31 KiB
Python
897 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
import pytest
|
|
from PySide6.QtWidgets import (
|
|
QApplication,
|
|
QLabel,
|
|
QLineEdit,
|
|
QPushButton,
|
|
QScrollArea,
|
|
QTextBrowser,
|
|
QTextEdit,
|
|
QWidget,
|
|
)
|
|
|
|
from doctor_workstation.core import PermissionSet
|
|
from doctor_workstation.ui.dialogs import ai_consult as ai_consult_module
|
|
from doctor_workstation.ui.dialogs import prescription as prescription_module
|
|
from doctor_workstation.ui.dialogs.ai_consult import AiConsultDialog
|
|
|
|
|
|
@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: Callable[..., Any],
|
|
*args: Any,
|
|
on_success: Callable[[Any], Any] | None = None,
|
|
on_error: Callable[[Exception], Any] | None = None,
|
|
on_finished: Callable[[], Any] | None = None,
|
|
**_kwargs: Any,
|
|
) -> object:
|
|
try:
|
|
result = function(*args)
|
|
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(ai_consult_module, "run_async", run_immediately)
|
|
|
|
|
|
class DeferredAsync:
|
|
def __init__(self) -> None:
|
|
self.pending: list[dict[str, Any]] = []
|
|
|
|
def __call__(
|
|
self,
|
|
function: Callable[..., Any],
|
|
*args: Any,
|
|
on_success: Callable[[Any], Any] | None = None,
|
|
on_error: Callable[[Exception], Any] | None = None,
|
|
on_finished: Callable[[], Any] | None = None,
|
|
**_kwargs: Any,
|
|
) -> object:
|
|
self.pending.append(
|
|
{
|
|
"function": function,
|
|
"args": args,
|
|
"on_success": on_success,
|
|
"on_error": on_error,
|
|
"on_finished": on_finished,
|
|
}
|
|
)
|
|
return object()
|
|
|
|
def complete(self, index: int) -> None:
|
|
pending = self.pending[index]
|
|
try:
|
|
result = pending["function"](*pending["args"])
|
|
except Exception as error:
|
|
if pending["on_error"]:
|
|
pending["on_error"](error)
|
|
else:
|
|
if pending["on_success"]:
|
|
pending["on_success"](result)
|
|
finally:
|
|
if pending["on_finished"]:
|
|
pending["on_finished"]()
|
|
|
|
|
|
def _detail(diagnosis_id: int, marker: str) -> dict[str, Any]:
|
|
diagnosis = {
|
|
"id": diagnosis_id,
|
|
"patient_id": diagnosis_id + 1000,
|
|
"patient_name": f"{marker}患者",
|
|
"phone": "13800138000",
|
|
"id_card": "110105199203071234",
|
|
"gender": 0,
|
|
"age": 34,
|
|
"region": f"{marker}杭州",
|
|
"address": f"{marker}健康路 8 号",
|
|
"height": 162,
|
|
"weight": 54.5,
|
|
"bmi": 20.8,
|
|
"systolic_pressure": 146,
|
|
"diastolic_pressure": 92,
|
|
"fasting_blood_sugar": 8.2,
|
|
"chief_complaint": f"{marker}主诉口渴乏力",
|
|
"present_illness": f"{marker}现病史半年血糖波动",
|
|
"past_history": f"{marker}既往高血压五年",
|
|
"allergy_history": f"{marker}青霉素过敏",
|
|
"family_history": f"{marker}父亲糖尿病",
|
|
"clinical_diagnosis": f"{marker}气阴两虚",
|
|
"diabetes_discovery_year": 6,
|
|
"current_medications": [f"{marker}二甲双胍", "阿卡波糖"],
|
|
"smoking": "不吸烟",
|
|
"sleep_condition": [f"{marker}易醒", "多梦"],
|
|
"local_hospital_diagnosis": [f"{marker}2 型糖尿病", "高血压"],
|
|
"diet_condition": [f"{marker}偏甜", "夜宵"],
|
|
"body_feeling": [f"{marker}乏力", "四肢沉重"],
|
|
"tongue": f"{marker}舌淡红",
|
|
"tongue_coating": f"{marker}苔薄白",
|
|
"pulse": f"{marker}脉细",
|
|
"remark": f"{marker}继续监测",
|
|
"latest_prescription_order": {
|
|
"id": f"{marker}-RX-09",
|
|
"status_text": "待配药",
|
|
},
|
|
}
|
|
for index in range(12):
|
|
diagnosis[f"custom_field_{index}"] = f"{marker}扩展病历字段 {index}"
|
|
return {
|
|
"diagnosis": diagnosis,
|
|
"patient": {
|
|
"id": diagnosis_id + 1000,
|
|
"patient_name": f"{marker}患者",
|
|
"phone": "13800138000",
|
|
"id_card": "110105199203071234",
|
|
"gender": 0,
|
|
"age": 34,
|
|
"region": f"{marker}杭州",
|
|
"address": f"{marker}健康路 8 号",
|
|
},
|
|
"appointment": {"doctor_name": f"{marker}陈医生"},
|
|
}
|
|
|
|
|
|
class WorkspaceRepository:
|
|
def __init__(self, *, include_foreign_rows: bool = True) -> None:
|
|
self.details = {501: _detail(501, "甲"), 502: _detail(502, "乙")}
|
|
self.include_foreign_rows = include_foreign_rows
|
|
self.failures: set[tuple[str, int]] = set()
|
|
self.calls: list[tuple[str, int]] = []
|
|
self.prescription_detail_calls: list[int] = []
|
|
self.prescription_overrides: dict[int, dict[str, Any]] = {}
|
|
self.report_payload: Any = []
|
|
|
|
def _check(self, name: str, diagnosis_id: int) -> None:
|
|
self.calls.append((name, diagnosis_id))
|
|
if (name, diagnosis_id) in self.failures:
|
|
raise RuntimeError(f"{name} 暂时不可用")
|
|
|
|
def get_diagnosis_detail(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
readonly: bool = False,
|
|
) -> dict[str, Any]:
|
|
del readonly
|
|
self._check("get_diagnosis_detail", diagnosis_id)
|
|
return self.details[diagnosis_id]
|
|
|
|
def list_im_chat_messages(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
only_archived: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
del only_archived
|
|
self._check("list_im_chat_messages", diagnosis_id)
|
|
return []
|
|
|
|
def list_patient_ai_reports(self, patient_id: int) -> Any:
|
|
self.calls.append(("list_patient_ai_reports", patient_id))
|
|
return self.report_payload
|
|
|
|
def get_doctor_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
self._check("get_doctor_notes", diagnosis_id)
|
|
marker = "甲" if diagnosis_id == 501 else "乙"
|
|
rows = [
|
|
{
|
|
"id": diagnosis_id * 10 + 1,
|
|
"diagnosis_id": diagnosis_id,
|
|
"create_time": "2026-08-18 09:20",
|
|
"content": f"{marker}医生检查记录",
|
|
"tongue_images": [
|
|
{
|
|
"name": f"{marker}舌苔照片.jpg",
|
|
"url": f"https://media.example.invalid/{marker}/tongue.jpg",
|
|
}
|
|
],
|
|
"report_files": [
|
|
{
|
|
"name": f"{marker}血糖报告.pdf",
|
|
"url": f"https://media.example.invalid/{marker}/report.pdf",
|
|
},
|
|
{
|
|
"name": f"{marker}本地危险附件.pdf",
|
|
"url": "file:///C:/private/unsafe.pdf",
|
|
},
|
|
],
|
|
}
|
|
]
|
|
if self.include_foreign_rows:
|
|
rows.append(
|
|
{
|
|
"id": 99901,
|
|
"diagnosis_id": 999,
|
|
"content": "错误诊单附件哨兵",
|
|
"tongue_images": [
|
|
{
|
|
"name": "错误诊单舌苔.jpg",
|
|
"url": "https://media.example.invalid/wrong.jpg",
|
|
}
|
|
],
|
|
}
|
|
)
|
|
return rows
|
|
|
|
def get_tracking_window(self, diagnosis_id: int) -> dict[str, Any]:
|
|
self._check("get_tracking_window", diagnosis_id)
|
|
marker = "甲" if diagnosis_id == 501 else "乙"
|
|
blood_records = [
|
|
{
|
|
"id": diagnosis_id * 10 + 2,
|
|
"diagnosis_id": diagnosis_id,
|
|
"record_date": "2026-08-18",
|
|
"fasting_blood_sugar": f"{marker}8.2",
|
|
"postprandial_blood_sugar": f"{marker}12.4",
|
|
"systolic_pressure": f"{marker}146",
|
|
"diastolic_pressure": f"{marker}92",
|
|
}
|
|
]
|
|
if self.include_foreign_rows:
|
|
blood_records.append(
|
|
{
|
|
"id": 99902,
|
|
"diagnosis_id": 999,
|
|
"record_date": "2026-08-18",
|
|
"fasting_blood_sugar": "错误诊单血糖 19.9",
|
|
}
|
|
)
|
|
return {
|
|
"diagnosis_id": diagnosis_id,
|
|
"blood_records": blood_records,
|
|
"diet_records": [
|
|
{
|
|
"id": diagnosis_id * 10 + 3,
|
|
"diagnosis_id": diagnosis_id,
|
|
"record_date": "2026-08-18",
|
|
"breakfast_foods": f"{marker}燕麦鸡蛋",
|
|
"lunch_foods": f"{marker}杂粮饭",
|
|
}
|
|
],
|
|
"exercise_records": [
|
|
{
|
|
"id": diagnosis_id * 10 + 4,
|
|
"diagnosis_id": diagnosis_id,
|
|
"record_date": "2026-08-17",
|
|
"exercise_type": f"{marker}散步",
|
|
"duration": 35,
|
|
}
|
|
],
|
|
}
|
|
|
|
def list_prescriptions_by_diagnosis(
|
|
self,
|
|
diagnosis_id: int,
|
|
) -> list[dict[str, Any]]:
|
|
self._check("list_prescriptions_by_diagnosis", diagnosis_id)
|
|
marker = "甲" if diagnosis_id == 501 else "乙"
|
|
return [
|
|
{
|
|
"id": diagnosis_id * 10 + index,
|
|
"diagnosis_id": diagnosis_id,
|
|
"sn": f"{marker}-RX-{index}",
|
|
"prescription_date": f"2026-08-{10 + index}",
|
|
"prescription_summary": f"{marker}方剂 {index}",
|
|
"doctor_name": f"{marker}陈医生",
|
|
"status_text": "已审核",
|
|
"herbs": [{"name": f"{marker}黄芪", "dosage": index * 5, "unit": "g"}],
|
|
}
|
|
for index in range(1, 4)
|
|
]
|
|
|
|
def get_prescription(self, prescription_id: int) -> dict[str, Any]:
|
|
self.prescription_detail_calls.append(prescription_id)
|
|
if prescription_id in self.prescription_overrides:
|
|
return self.prescription_overrides[prescription_id]
|
|
diagnosis_id = prescription_id // 10
|
|
return {
|
|
"id": prescription_id,
|
|
"diagnosis_id": diagnosis_id,
|
|
"sn": f"FULL-{prescription_id}",
|
|
"clinical_diagnosis": "气阴两虚",
|
|
"herbs": [{"name": "黄芪", "dosage": 15, "unit": "g"}],
|
|
}
|
|
|
|
|
|
def _permissions() -> PermissionSet:
|
|
return PermissionSet(["tcm.diagnosis/aiAssistant", "cf.prescription/read"])
|
|
|
|
|
|
def _pane_text(widget: QWidget) -> str:
|
|
parts = [child.text() for child in widget.findChildren(QLabel)]
|
|
parts.extend(child.text() for child in widget.findChildren(QPushButton))
|
|
parts.extend(child.text() for child in widget.findChildren(QLineEdit))
|
|
parts.extend(child.toPlainText() for child in widget.findChildren(QTextBrowser))
|
|
parts.extend(child.toPlainText() for child in widget.findChildren(QTextEdit))
|
|
return "\n".join(part for part in parts if part)
|
|
|
|
|
|
def _open_dialog(
|
|
application: QApplication,
|
|
repository: WorkspaceRepository,
|
|
diagnosis_id: int = 501,
|
|
) -> AiConsultDialog:
|
|
dialog = AiConsultDialog(repository, _permissions())
|
|
dialog.open_for(diagnosis_id=diagnosis_id, patient_id=diagnosis_id + 1000)
|
|
dialog.show()
|
|
application.processEvents()
|
|
return dialog
|
|
|
|
|
|
def test_case_tab_renders_complete_owned_detail_as_readable_chinese(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
dialog = _open_dialog(application, WorkspaceRepository())
|
|
pane = dialog.records["病历资料"]
|
|
dialog.tabs.setCurrentIndex(1)
|
|
application.processEvents()
|
|
|
|
assert pane.findChild(QWidget, "AiConsultCaseGrid") is not None
|
|
text = _pane_text(pane)
|
|
for sentinel in (
|
|
"甲主诉口渴乏力",
|
|
"甲现病史半年血糖波动",
|
|
"甲既往高血压五年",
|
|
"甲青霉素过敏",
|
|
"甲父亲糖尿病",
|
|
"甲气阴两虚",
|
|
"甲2 型糖尿病",
|
|
"高血压",
|
|
"甲偏甜",
|
|
"夜宵",
|
|
"甲-RX-09",
|
|
"待配药",
|
|
):
|
|
assert sentinel in text
|
|
assert "['" not in text
|
|
assert "{'" not in text
|
|
|
|
scroll = pane.findChild(QScrollArea)
|
|
assert scroll is not None and scroll.widgetResizable()
|
|
assert pane.geometry().isValid() and scroll.viewport().geometry().isValid()
|
|
assert scroll.verticalScrollBar().maximum() > 0
|
|
dialog.close()
|
|
|
|
|
|
def test_all_four_record_tabs_use_the_selected_diagnosis_id(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
dialog = _open_dialog(application, repository, diagnosis_id=501)
|
|
|
|
for method in (
|
|
"get_diagnosis_detail",
|
|
"get_doctor_notes",
|
|
"get_tracking_window",
|
|
"list_prescriptions_by_diagnosis",
|
|
):
|
|
assert (method, 501) in repository.calls
|
|
assert all(
|
|
called_id == 501
|
|
for called_method, called_id in repository.calls
|
|
if called_method == method
|
|
)
|
|
assert ("list_patient_ai_reports", 1501) in repository.calls
|
|
dialog.close()
|
|
|
|
|
|
def test_seed_cannot_replace_the_authoritative_patient_id(
|
|
application: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
deferred = DeferredAsync()
|
|
monkeypatch.setattr(ai_consult_module, "run_async", deferred)
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
dialog = AiConsultDialog(repository, _permissions())
|
|
|
|
dialog.open_for(
|
|
diagnosis_id=501,
|
|
patient_id=1501,
|
|
seed={"patient_id": 501, "patient_name": "错误种子"},
|
|
)
|
|
|
|
assert dialog.patient_id == 1501
|
|
deferred.complete(0)
|
|
application.processEvents()
|
|
assert dialog.patient_id == 1501
|
|
assert ("list_patient_ai_reports", 1501) in repository.calls
|
|
dialog.close()
|
|
|
|
|
|
def test_detail_failure_or_wrong_owner_never_requests_patient_reports(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
failed_repository = WorkspaceRepository(include_foreign_rows=False)
|
|
failed_repository.failures.add(("get_diagnosis_detail", 501))
|
|
failed = _open_dialog(application, failed_repository)
|
|
assert all(
|
|
method != "list_patient_ai_reports"
|
|
for method, _owner in failed_repository.calls
|
|
)
|
|
failed.close()
|
|
|
|
wrong_repository = WorkspaceRepository(include_foreign_rows=False)
|
|
wrong_repository.details[501] = _detail(999, "越权")
|
|
wrong = _open_dialog(application, wrong_repository)
|
|
assert all(
|
|
method != "list_patient_ai_reports"
|
|
for method, _owner in wrong_repository.calls
|
|
)
|
|
wrong.close()
|
|
|
|
|
|
def test_patient_report_response_owner_must_match_exactly(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
repository.report_payload = {
|
|
"patient_id": "1501",
|
|
"reports": [
|
|
{
|
|
"patient_id": 1501,
|
|
"report": {"diagnosis": "不应显示的越权报告"},
|
|
}
|
|
],
|
|
}
|
|
dialog = _open_dialog(application, repository)
|
|
|
|
assert ("list_patient_ai_reports", 1501) in repository.calls
|
|
assert "不应显示的越权报告" not in _pane_text(dialog)
|
|
assert not ai_consult_module._report_response_matches_patient(
|
|
repository.report_payload,
|
|
1501,
|
|
)
|
|
dialog.close()
|
|
|
|
|
|
def test_exam_tab_filters_foreign_attachments_and_blocks_file_urls(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
opened: list[str] = []
|
|
monkeypatch.setattr(
|
|
ai_consult_module,
|
|
"open_safe_http_url",
|
|
lambda target: opened.append(target) or True,
|
|
)
|
|
dialog = _open_dialog(application, WorkspaceRepository())
|
|
pane = dialog.records["检查检验"]
|
|
dialog.tabs.setCurrentIndex(2)
|
|
application.processEvents()
|
|
|
|
assert pane.findChild(QWidget, "AiConsultExamTimeline") is not None
|
|
text = _pane_text(pane)
|
|
assert "甲舌苔照片.jpg" in text
|
|
assert "甲血糖报告.pdf" in text
|
|
assert "甲本地危险附件.pdf" in text
|
|
assert "错误诊单附件哨兵" not in text
|
|
assert "错误诊单舌苔.jpg" not in text
|
|
|
|
buttons = pane.findChildren(QPushButton, "AiConsultMediaOpen")
|
|
assert len(buttons) == 3
|
|
thumbnails = pane.findChildren(QPushButton, "AiConsultTongueThumb")
|
|
assert len(thumbnails) == 1
|
|
assert thumbnails[0].isEnabled()
|
|
assert thumbnails[0].accessibleName() == "舌苔图片点击查看"
|
|
assert thumbnails[0].property("loadState") == "blocked"
|
|
assert pane.state_label.property("state") == "warning" # type: ignore[attr-defined]
|
|
assert pane.retry_button.isHidden() # type: ignore[attr-defined]
|
|
unsafe = next(button for button in buttons if "本地危险附件" in button.text())
|
|
assert not unsafe.isEnabled()
|
|
for button in buttons:
|
|
button.click()
|
|
assert len(opened) == 2
|
|
assert all(target.startswith(("http://", "https://")) for target in opened)
|
|
assert all(not target.startswith("file:") for target in opened)
|
|
dialog.close()
|
|
|
|
|
|
def test_tongue_thumbnail_auto_get_requires_configured_https_origin(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
requested: list[str] = []
|
|
|
|
class RecordingRemoteImageButton(QPushButton):
|
|
def __init__(self, source: str, **kwargs: Any) -> None:
|
|
super().__init__(kwargs.get("parent"))
|
|
requested.append(source)
|
|
self.setObjectName(str(kwargs.get("object_name") or ""))
|
|
self.setAccessibleName(
|
|
str(kwargs.get("fallback_text") or "").replace("\n", "")
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
ai_consult_module,
|
|
"_RemoteImageButton",
|
|
RecordingRemoteImageButton,
|
|
)
|
|
|
|
untrusted = _open_dialog(application, WorkspaceRepository())
|
|
assert requested == []
|
|
untrusted.close()
|
|
|
|
trusted_repository = WorkspaceRepository()
|
|
trusted_repository.trusted_media_domains = ["media.example.invalid"]
|
|
assert not ai_consult_module._trusted_thumbnail_url(
|
|
trusted_repository,
|
|
"http://media.example.invalid/甲/tongue.jpg",
|
|
)
|
|
assert not ai_consult_module._trusted_thumbnail_url(
|
|
trusted_repository,
|
|
"https://sub.media.example.invalid/甲/tongue.jpg",
|
|
)
|
|
trusted = _open_dialog(application, trusted_repository)
|
|
assert requested == ["https://media.example.invalid/甲/tongue.jpg"]
|
|
trusted.close()
|
|
|
|
|
|
def test_three_prescription_cards_open_exact_details_and_reject_wrong_or_late_ids(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
opened: list[int] = []
|
|
|
|
class FakePrescriptionDetailDialog:
|
|
def __init__(self, prescription: Any, **_kwargs: Any) -> None:
|
|
self.prescription = prescription
|
|
|
|
def exec(self) -> None:
|
|
opened.append(int(self.prescription["id"]))
|
|
|
|
monkeypatch.setattr(
|
|
prescription_module,
|
|
"PrescriptionDetailDialog",
|
|
FakePrescriptionDetailDialog,
|
|
)
|
|
dialog = _open_dialog(application, repository)
|
|
pane = dialog.records["处方记录"]
|
|
dialog.tabs.setCurrentIndex(3)
|
|
application.processEvents()
|
|
|
|
cards = pane.findChildren(QWidget, "AiConsultPrescriptionCard")
|
|
buttons = sorted(
|
|
pane.findChildren(QPushButton, "AiConsultPrescriptionOpen"),
|
|
key=lambda button: int(button.property("prescriptionId")),
|
|
)
|
|
expected_ids = [5011, 5012, 5013]
|
|
assert len(cards) == len(buttons) == 3
|
|
assert [int(button.property("prescriptionId")) for button in buttons] == expected_ids
|
|
assert all(button.text() == "查看详情" for button in buttons)
|
|
for button in buttons:
|
|
button.click()
|
|
assert repository.prescription_detail_calls == expected_ids
|
|
assert opened == expected_ids
|
|
|
|
repository.prescription_overrides[5011] = {
|
|
"id": 9999,
|
|
"diagnosis_id": 501,
|
|
}
|
|
buttons[0].click()
|
|
assert repository.prescription_detail_calls[-1] == 5011
|
|
assert opened == expected_ids
|
|
repository.prescription_overrides.pop(5011)
|
|
|
|
repository.prescription_overrides[5011] = {"id": 5011}
|
|
buttons[0].click()
|
|
assert repository.prescription_detail_calls[-1] == 5011
|
|
assert opened == expected_ids
|
|
repository.prescription_overrides.pop(5011)
|
|
|
|
calls_before_unowned_source = list(repository.prescription_detail_calls)
|
|
dialog._open_prescription_detail({"id": 5011})
|
|
assert repository.prescription_detail_calls == calls_before_unowned_source
|
|
|
|
deferred = DeferredAsync()
|
|
monkeypatch.setattr(ai_consult_module, "run_async", deferred)
|
|
buttons[0].click()
|
|
buttons[1].click()
|
|
assert len(deferred.pending) == 2
|
|
deferred.complete(1)
|
|
application.processEvents()
|
|
deferred.complete(0)
|
|
application.processEvents()
|
|
assert repository.prescription_detail_calls[-2:] == [5012, 5011]
|
|
assert opened == [*expected_ids, 5012]
|
|
dialog.close()
|
|
|
|
|
|
def test_health_tab_masks_sensitive_patient_data_and_renders_tracking_window(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
dialog = _open_dialog(application, WorkspaceRepository())
|
|
pane = dialog.records["健康档案"]
|
|
dialog.tabs.setCurrentIndex(4)
|
|
application.processEvents()
|
|
|
|
assert pane.findChild(QWidget, "AiConsultHealthGrid") is not None
|
|
text = _pane_text(pane)
|
|
for sentinel in (
|
|
"甲患者",
|
|
"甲杭州",
|
|
"甲健康路 8 号",
|
|
"138****8000",
|
|
"110***********1234",
|
|
"甲8.2",
|
|
"甲12.4",
|
|
"甲146",
|
|
"甲92",
|
|
"甲燕麦鸡蛋",
|
|
"甲杂粮饭",
|
|
"甲散步",
|
|
"35",
|
|
"甲气阴两虚",
|
|
"甲二甲双胍",
|
|
"阿卡波糖",
|
|
"不吸烟",
|
|
"甲易醒",
|
|
"多梦",
|
|
):
|
|
assert sentinel in text
|
|
assert pane.findChild(QWidget, "AiConsultDiagnosisHealthSummary") is not None
|
|
assert "13800138000" not in text
|
|
assert "110105199203071234" not in text
|
|
assert "错误诊单血糖 19.9" not in text
|
|
assert pane.state_label.property("state") == "warning" # type: ignore[attr-defined]
|
|
assert pane.retry_button.isHidden() # type: ignore[attr-defined]
|
|
dialog.close()
|
|
|
|
|
|
def test_ownerless_notes_prescriptions_and_tracking_rows_fail_closed_as_warning(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
class OwnerlessRepository(WorkspaceRepository):
|
|
def get_doctor_notes(self, diagnosis_id: int) -> list[dict[str, Any]]:
|
|
rows = super().get_doctor_notes(diagnosis_id)
|
|
for row in rows:
|
|
row.pop("diagnosis_id", None)
|
|
return rows
|
|
|
|
def list_prescriptions_by_diagnosis(
|
|
self,
|
|
diagnosis_id: int,
|
|
) -> list[dict[str, Any]]:
|
|
rows = super().list_prescriptions_by_diagnosis(diagnosis_id)
|
|
for row in rows:
|
|
row.pop("diagnosis_id", None)
|
|
return rows
|
|
|
|
def get_tracking_window(self, diagnosis_id: int) -> dict[str, Any]:
|
|
result = super().get_tracking_window(diagnosis_id)
|
|
for key in ("blood_records", "diet_records", "exercise_records"):
|
|
for row in result[key]:
|
|
row.pop("diagnosis_id", None)
|
|
return result
|
|
|
|
dialog = _open_dialog(
|
|
application,
|
|
OwnerlessRepository(include_foreign_rows=False),
|
|
)
|
|
exam = dialog.records["检查检验"]
|
|
prescriptions = dialog.records["处方记录"]
|
|
health = dialog.records["健康档案"]
|
|
|
|
assert "甲医生检查记录" not in _pane_text(exam)
|
|
assert not prescriptions.findChildren(QWidget, "AiConsultPrescriptionCard")
|
|
assert "甲燕麦鸡蛋" not in _pane_text(health)
|
|
for pane in (exam, prescriptions, health):
|
|
assert pane.state_label.property("state") == "warning" # type: ignore[attr-defined]
|
|
assert pane.retry_button.isHidden() # type: ignore[attr-defined]
|
|
dialog.close()
|
|
|
|
|
|
def test_ownerless_im_messages_fail_closed_before_chat_rendering(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
class OwnerlessMessageRepository(WorkspaceRepository):
|
|
def list_im_chat_messages(
|
|
self,
|
|
diagnosis_id: int,
|
|
*,
|
|
only_archived: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
del diagnosis_id, only_archived
|
|
return [
|
|
{
|
|
"msg_id": "ownerless-message",
|
|
"msg_type": "text",
|
|
"text": "不应展示的无归属会话",
|
|
"is_from_doctor": False,
|
|
}
|
|
]
|
|
|
|
dialog = _open_dialog(application, OwnerlessMessageRepository())
|
|
chat_text = "\n".join(
|
|
label.text() for label in dialog.chat_host.findChildren(QLabel)
|
|
)
|
|
|
|
assert "不应展示的无归属会话" not in chat_text
|
|
dialog.close()
|
|
|
|
|
|
def test_tracking_response_without_diagnosis_owner_is_filtered_without_retry(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
class OwnerlessTrackingRepository(WorkspaceRepository):
|
|
def get_tracking_window(self, diagnosis_id: int) -> dict[str, Any]:
|
|
result = super().get_tracking_window(diagnosis_id)
|
|
result.pop("diagnosis_id", None)
|
|
return result
|
|
|
|
dialog = _open_dialog(
|
|
application,
|
|
OwnerlessTrackingRepository(include_foreign_rows=False),
|
|
)
|
|
pane = dialog.records["健康档案"]
|
|
|
|
assert "甲燕麦鸡蛋" not in _pane_text(pane)
|
|
assert pane.state_label.property("state") == "warning" # type: ignore[attr-defined]
|
|
assert pane.retry_button.isHidden() # type: ignore[attr-defined]
|
|
dialog.close()
|
|
|
|
|
|
def test_late_workspace_a_response_cannot_pollute_selected_workspace_b(
|
|
application: QApplication,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
deferred = DeferredAsync()
|
|
monkeypatch.setattr(ai_consult_module, "run_async", deferred)
|
|
dialog = AiConsultDialog(repository, _permissions())
|
|
dialog.open_for(diagnosis_id=501, patient_id=1501)
|
|
dialog.open_for(diagnosis_id=502, patient_id=1502)
|
|
dialog.show()
|
|
assert len(deferred.pending) == 2
|
|
|
|
deferred.complete(1)
|
|
application.processEvents()
|
|
deferred.complete(0)
|
|
application.processEvents()
|
|
for title in ("病历资料", "检查检验", "处方记录", "健康档案"):
|
|
text = _pane_text(dialog.records[title])
|
|
assert "乙" in text
|
|
assert "甲主诉口渴乏力" not in text
|
|
assert "甲医生检查记录" not in text
|
|
assert "甲-RX-1" not in text
|
|
assert "甲燕麦鸡蛋" not in text
|
|
assert dialog.diagnosis_id == 502
|
|
assert dialog._detail["diagnosis"]["id"] == 502
|
|
dialog.close()
|
|
|
|
|
|
def test_mismatched_detail_owner_fails_closed_across_all_record_tabs(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
) -> None:
|
|
repository = WorkspaceRepository()
|
|
repository.details[501] = _detail(999, "越权")
|
|
dialog = _open_dialog(application, repository)
|
|
|
|
forbidden = (
|
|
"越权主诉口渴乏力",
|
|
"甲医生检查记录",
|
|
"甲舌苔照片.jpg",
|
|
"甲-RX-1",
|
|
"甲燕麦鸡蛋",
|
|
)
|
|
for title in ("病历资料", "检查检验", "处方记录", "健康档案"):
|
|
pane = dialog.records[title]
|
|
text = _pane_text(pane)
|
|
assert all(sentinel not in text for sentinel in forbidden)
|
|
state = pane.state_label # type: ignore[attr-defined]
|
|
retry = pane.retry_button # type: ignore[attr-defined]
|
|
assert state.objectName() == "AiConsultRecordState"
|
|
assert retry.objectName() == "AiConsultRecordRetry"
|
|
assert state is not None and state.property("state") == "warning"
|
|
assert retry is not None and retry.isHidden()
|
|
assert all(
|
|
method != "list_patient_ai_reports" for method, _owner in repository.calls
|
|
)
|
|
dialog.close()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("method", "error_tabs", "success_sentinels"),
|
|
[
|
|
(
|
|
"get_diagnosis_detail",
|
|
{"病历资料", "健康档案"},
|
|
{"检查检验": "甲医生检查记录", "处方记录": "甲-RX-1"},
|
|
),
|
|
(
|
|
"get_doctor_notes",
|
|
{"检查检验"},
|
|
{
|
|
"病历资料": "甲主诉口渴乏力",
|
|
"处方记录": "甲-RX-1",
|
|
"健康档案": "甲燕麦鸡蛋",
|
|
},
|
|
),
|
|
(
|
|
"get_tracking_window",
|
|
{"健康档案"},
|
|
{
|
|
"病历资料": "甲主诉口渴乏力",
|
|
"检查检验": "甲医生检查记录",
|
|
"处方记录": "甲-RX-1",
|
|
},
|
|
),
|
|
(
|
|
"list_prescriptions_by_diagnosis",
|
|
{"处方记录"},
|
|
{
|
|
"病历资料": "甲主诉口渴乏力",
|
|
"检查检验": "甲医生检查记录",
|
|
"健康档案": "甲燕麦鸡蛋",
|
|
},
|
|
),
|
|
],
|
|
)
|
|
def test_one_failed_source_has_local_error_retry_and_preserves_other_sections(
|
|
application: QApplication,
|
|
immediate_async: None,
|
|
method: str,
|
|
error_tabs: set[str],
|
|
success_sentinels: dict[str, str],
|
|
) -> None:
|
|
repository = WorkspaceRepository(include_foreign_rows=False)
|
|
repository.failures.add((method, 501))
|
|
dialog = _open_dialog(application, repository)
|
|
|
|
for title in error_tabs:
|
|
pane = dialog.records[title]
|
|
state = pane.state_label # type: ignore[attr-defined]
|
|
retry = pane.retry_button # type: ignore[attr-defined]
|
|
assert state.objectName() == "AiConsultRecordState"
|
|
assert retry.objectName() == "AiConsultRecordRetry"
|
|
assert state is not None and state.property("state") == "error"
|
|
assert retry is not None and not retry.isHidden() and retry.isEnabled()
|
|
for title, sentinel in success_sentinels.items():
|
|
assert sentinel in _pane_text(dialog.records[title])
|
|
state = dialog.records[title].state_label # type: ignore[attr-defined]
|
|
assert state is not None and state.property("state") != "error", (
|
|
title,
|
|
state.text(),
|
|
state.property("state"),
|
|
)
|
|
|
|
repository.failures.clear()
|
|
dialog.records[next(iter(error_tabs))].retry_button.click() # type: ignore[attr-defined]
|
|
application.processEvents()
|
|
for pane in dialog.records.values():
|
|
state = pane.state_label # type: ignore[attr-defined]
|
|
assert state is not None and state.property("state") != "error"
|
|
dialog.close()
|