1994 lines
67 KiB
Python
1994 lines
67 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
from datetime import date
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import httpx
|
||
import pytest
|
||
from PySide6.QtCore import QDate, QPoint, Qt
|
||
from PySide6.QtGui import QPalette
|
||
from PySide6.QtWidgets import QApplication, QLabel, QPushButton, QScrollArea
|
||
|
||
from doctor_workstation.core import PermissionSet
|
||
from doctor_workstation.services.api_client import ApiClient
|
||
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 (
|
||
NOTE_LIMIT,
|
||
QueueRow,
|
||
ReceptionPage,
|
||
_is_image_attachment,
|
||
)
|
||
from doctor_workstation.ui.widgets import StatusBadge
|
||
|
||
|
||
@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)
|
||
|
||
|
||
@pytest.fixture
|
||
def queued_async(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
|
||
"""Capture asynchronous work so tests can advance each model independently."""
|
||
|
||
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)
|
||
return jobs
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("path", "expected"),
|
||
[
|
||
("https://cdn.test/tongue.JPG?token=1", True),
|
||
("https://cdn.test/report.webp", True),
|
||
("https://cdn.test/report.pdf", False),
|
||
],
|
||
)
|
||
def test_note_attachment_preview_type_is_extension_aware(path: str, expected: bool) -> None:
|
||
assert _is_image_attachment(path) is expected
|
||
|
||
|
||
def test_note_attachments_expand_images_and_keep_file_open(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
page = ReceptionPage(DemoDoctorRepository(), PermissionSet([]))
|
||
page._render_notes(
|
||
[
|
||
{
|
||
"id": 1,
|
||
"note_date": "2026-08-12",
|
||
"tongue_images": ["https://cdn.test/tongue.jpg"],
|
||
"report_files": [
|
||
"https://cdn.test/check.png",
|
||
"https://cdn.test/check.pdf",
|
||
],
|
||
}
|
||
]
|
||
)
|
||
|
||
previews = page.notes_container.findChildren(
|
||
QPushButton, "NoteAttachmentPreview"
|
||
)
|
||
assert len(previews) == 2
|
||
assert all(button.text() == "" for button in previews)
|
||
assert all(not button.icon().isNull() for button in previews)
|
||
labels = [button.text() for button in page.notes_container.findChildren(QPushButton)]
|
||
assert labels.count("打开") == 1
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_status_badge_is_not_clipped_in_narrow_panel(
|
||
application: QApplication,
|
||
) -> None:
|
||
row = QueueRow(
|
||
{
|
||
"patient_name": "张蒙",
|
||
"status": 1,
|
||
"status_desc": "问诊中",
|
||
"appointment_time": "12:45:00",
|
||
"gender": 1,
|
||
"age": 36,
|
||
"assistant_name": "苏亚梅",
|
||
}
|
||
)
|
||
row.setFixedWidth(280)
|
||
row.show()
|
||
application.processEvents()
|
||
|
||
badge = row.findChild(StatusBadge)
|
||
assert badge is not None
|
||
assert badge.height() >= 24
|
||
assert badge.width() >= 54
|
||
assert badge.geometry().right() < row.width()
|
||
|
||
row.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_condition_summary_never_exposes_raw_dict(
|
||
application: QApplication,
|
||
) -> None:
|
||
row = QueueRow(
|
||
{
|
||
"patient_name": "结构患者",
|
||
"status": 1,
|
||
"gender": 1,
|
||
"age": 52,
|
||
"diagnosis": {
|
||
"clinical_diagnosis": "2 型糖尿病",
|
||
"chief_complaint": "口渴多饮",
|
||
"disease_course_text": "病程 10 年",
|
||
},
|
||
}
|
||
)
|
||
summary = row.findChild(QLabel, "ReceptionQueueSubline")
|
||
assert summary is not None
|
||
assert summary.text() == "2 型糖尿病 · 病程 10 年"
|
||
assert "{" not in summary.text()
|
||
|
||
fallback = QueueRow(
|
||
{
|
||
"patient_name": "待完善患者",
|
||
"status": 1,
|
||
"gender": 2,
|
||
"age": 40,
|
||
"diagnosis": {"unknown": {"raw": "不能泄露"}},
|
||
}
|
||
)
|
||
fallback_summary = fallback.findChild(QLabel, "ReceptionQueueSubline")
|
||
assert fallback_summary is not None
|
||
assert fallback_summary.text() == "暂无病情摘要"
|
||
assert "{" not in fallback_summary.text()
|
||
|
||
row.close()
|
||
fallback.close()
|
||
application.processEvents()
|
||
|
||
|
||
def _detail(
|
||
appointment_id: int,
|
||
*,
|
||
name: str,
|
||
status: int = 1,
|
||
phone: str = "13800138000",
|
||
) -> dict[str, Any]:
|
||
patient_id = appointment_id + 100
|
||
diagnosis_id = appointment_id + 200
|
||
return {
|
||
"appointment": {
|
||
"id": appointment_id,
|
||
"patient_id": patient_id,
|
||
"patient_name": name,
|
||
"status": status,
|
||
"appointment_date": date.today().isoformat(),
|
||
"appointment_time": "09:30",
|
||
"doctor_name": "张医生",
|
||
"assistant_name": "李医助",
|
||
"appointment_type_text": "复诊",
|
||
"channel_text": "线上",
|
||
"remark": "准时到诊",
|
||
},
|
||
"patient": {
|
||
"id": patient_id,
|
||
"phone": phone,
|
||
"gender": 2,
|
||
"age": 42,
|
||
"height": 165,
|
||
"weight": 55,
|
||
"region_text": "浙江省杭州市",
|
||
},
|
||
"diagnosis": {
|
||
"id": diagnosis_id,
|
||
"patient_id": patient_id,
|
||
"patient_name": name,
|
||
"phone": phone,
|
||
"chief_complaint": "反复口渴",
|
||
"present_illness": "持续两周",
|
||
"clinical_diagnosis": "消渴",
|
||
},
|
||
}
|
||
|
||
|
||
def _analysis_payload(
|
||
label: str = "高血糖风险",
|
||
*,
|
||
model: str = "qwen",
|
||
) -> dict[str, Any]:
|
||
is_openai = model == "openai"
|
||
return {
|
||
"diagnosis_advice": (
|
||
"OpenAI：建议结合客观检查复核当前糖尿病控制情况"
|
||
if is_openai
|
||
else "2 型糖尿病，血糖控制不佳"
|
||
),
|
||
"risk_assessment": [
|
||
{"label": label, "level": "high"},
|
||
{"label": "心血管风险", "level": "medium"},
|
||
{"label": "肾脏风险", "level": "low"},
|
||
],
|
||
"treatment_advice": (
|
||
"OpenAI：复核近期检查趋势并评估联合用药安全性。"
|
||
if is_openai
|
||
else "建议调整降糖方案，考虑联合用药，加强生活方式干预。"
|
||
),
|
||
"model_key": model,
|
||
"model_label": "OpenAI" if is_openai else "千问",
|
||
"model_name": "gpt-5.2" if is_openai else "qwen3.6-35b",
|
||
"generated_at": "2026-08-14 10:31:00" if is_openai else "2026-08-14 10:30:00",
|
||
}
|
||
|
||
|
||
def test_queue_uses_admin_same_day_contract(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
return {"lists": [], "count": 0}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.search_edit.setText(" 王小明 ")
|
||
page.refresh()
|
||
|
||
assert calls == [
|
||
{
|
||
"status": 1,
|
||
"start_date": date.today().isoformat(),
|
||
"end_date": date.today().isoformat(),
|
||
"page_no": 1,
|
||
"page_size": 15,
|
||
"patient_name": "王小明",
|
||
}
|
||
]
|
||
|
||
page.queue_tabs.setCurrentIndex(1)
|
||
assert calls[-1]["status"] == 4
|
||
assert calls[-1]["start_date"] == calls[-1]["end_date"]
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_date_picker_filters_the_selected_day(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
return {"lists": [], "count": 0}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.show()
|
||
application.processEvents()
|
||
page.queue_date_button.click()
|
||
calendar = page.queue_date_button.calendarWidget()
|
||
assert calendar.isVisible()
|
||
page.queue_date_button._pick_date(QDate(2026, 8, 1))
|
||
assert not calendar.isVisible()
|
||
assert page.queue_date_button.text() == "2026-08-01"
|
||
assert page._queue_date == "2026-08-01"
|
||
assert calls[-1]["start_date"] == "2026-08-01"
|
||
assert calls[-1]["end_date"] == "2026-08-01"
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_silent_queue_polls_reuse_rows_and_do_not_restart_detail_or_ai(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(10, name="轮询患者")
|
||
|
||
class Repository:
|
||
queue_calls = 0
|
||
detail_calls = 0
|
||
analysis_calls: list[tuple[int, str]]
|
||
|
||
def __init__(self) -> None:
|
||
self.analysis_calls = []
|
||
|
||
def list_appointments(self, **_kwargs: Any) -> dict[str, Any]:
|
||
self.queue_calls += 1
|
||
return {
|
||
"lists": [
|
||
{
|
||
**detail["appointment"],
|
||
"diagnosis_id": detail["diagnosis"]["id"],
|
||
"clinical_diagnosis": "消渴",
|
||
"server_revision": self.queue_calls,
|
||
}
|
||
],
|
||
"count": 20 + self.queue_calls,
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 10
|
||
self.detail_calls += 1
|
||
return detail
|
||
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
assert diagnosis_id == 210
|
||
self.analysis_calls.append((diagnosis_id, model))
|
||
return _analysis_payload(model=model)
|
||
|
||
repository = Repository()
|
||
page = ReceptionPage(
|
||
repository,
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page.refresh()
|
||
first_item = page.queue_list.item(0)
|
||
first_row = page.queue_list.itemWidget(first_item)
|
||
|
||
for _index in range(3):
|
||
page.poll_timer.timeout.emit()
|
||
|
||
assert repository.queue_calls == 4
|
||
assert repository.detail_calls == 1
|
||
assert repository.analysis_calls == [(210, "qwen"), (210, "openai")]
|
||
assert page.queue_list.item(0) is first_item
|
||
assert page.queue_list.itemWidget(first_item) is first_row
|
||
assert first_item.data(Qt.ItemDataRole.UserRole)["server_revision"] == 4
|
||
assert page.queue_summary.text() == "已加载 1 / 共 24 位患者"
|
||
|
||
page.refresh()
|
||
assert repository.detail_calls == 2
|
||
assert repository.analysis_calls == [(210, "qwen"), (210, "openai")]
|
||
assert page.queue_list.item(0) is first_item
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_timer_poll_does_not_supersede_an_in_flight_queue_request(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
jobs.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(object(), PermissionSet([]))
|
||
page.refresh()
|
||
generation = page._queue_generation
|
||
|
||
for _index in range(3):
|
||
page.poll_timer.timeout.emit()
|
||
|
||
assert len(jobs) == 1
|
||
assert page._queue_generation == generation
|
||
|
||
jobs[0]["on_success"]({"lists": [], "count": 0})
|
||
jobs[0]["on_finished"]()
|
||
assert not page._queue_loading
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_silent_queue_error_does_not_replace_the_visible_banner(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
jobs.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(object(), PermissionSet([]))
|
||
|
||
page.refresh(silent=True)
|
||
jobs[0]["on_error"](RuntimeError("background unavailable"))
|
||
assert page.queue_banner.label.text() == ""
|
||
jobs[0]["on_finished"]()
|
||
|
||
page.refresh()
|
||
jobs[1]["on_error"](RuntimeError("foreground unavailable"))
|
||
assert page.queue_banner.label.text()
|
||
assert page.queue_banner.property("kind") == "danger"
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_fast_patient_switch_rejects_late_detail(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
callbacks: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
callbacks.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(object(), PermissionSet(["*"]))
|
||
first = {"id": 11, "patient_id": 111, "diagnosis_id": 211, "patient_name": "甲患者"}
|
||
second = {"id": 22, "patient_id": 122, "diagnosis_id": 222, "patient_name": "乙患者"}
|
||
|
||
page._select_record(first)
|
||
page.note_edit.setPlainText("甲患者的未保存草稿")
|
||
page._pending_report_files = [r"C:\records\first.pdf"]
|
||
page._select_record(second)
|
||
assert len(callbacks) == 2
|
||
assert page.patient_name_label.text() == "乙患者"
|
||
assert page.note_edit.toPlainText() == ""
|
||
assert page._pending_report_files == []
|
||
|
||
callbacks[0]["on_success"]({"detail": _detail(11, name="甲患者")})
|
||
callbacks[0]["on_finished"]()
|
||
assert page._selected_appointment_id == 22
|
||
assert page.patient_name_label.text() == "乙患者"
|
||
assert page._selected_detail is None
|
||
|
||
callbacks[1]["on_success"]({"detail": _detail(22, name="乙患者")})
|
||
callbacks[1]["on_finished"]()
|
||
assert page._selected_detail == _detail(22, name="乙患者")
|
||
assert page.patient_name_label.text() == "乙患者"
|
||
assert "反复口渴" in page.diagnosis_text.text()
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_medication_case_prioritizes_clinical_information_and_keeps_plain_summary(
|
||
application: QApplication,
|
||
) -> None:
|
||
page = ReceptionPage(object(), PermissionSet([]))
|
||
detail = _detail(31, name="层级患者")
|
||
diagnosis = dict(detail["diagnosis"])
|
||
diagnosis.update(
|
||
{
|
||
"diagnosis_date": 0,
|
||
"diagnosis_type_text": "复诊",
|
||
"systolic": 168,
|
||
"diastolic": 102,
|
||
"blood_pressure_status": "偏高",
|
||
"fasting_blood_sugar": "空腹6.8",
|
||
"fasting_blood_sugar_status": "偏高",
|
||
"current_medications": ["二甲双胍缓释片早晚分服", "格列吡嗪餐前服用"],
|
||
"allergy_history_text": "青霉素过敏",
|
||
"symptoms": "口干口苦、多饮多汗、腰膝酸软。" * 14,
|
||
"tongue": "舌红、苔黄腻",
|
||
"pulse": "脉弦滑",
|
||
"past_history_text": "高血压、高脂血症",
|
||
"remark": "重点复核用药剂量与低血糖风险。【完整病例末尾】",
|
||
}
|
||
)
|
||
page.resize(1280, 800)
|
||
page.show()
|
||
application.processEvents()
|
||
page.detail_stack.setCurrentIndex(1)
|
||
medication_tab = next(
|
||
index for index in range(page.detail_tabs.count())
|
||
if page.detail_tabs.tabText(index) == "用药记录"
|
||
)
|
||
page.detail_tabs.setCurrentIndex(medication_tab)
|
||
application.processEvents()
|
||
page._render_case(
|
||
{**detail["appointment"], "has_prescription": True},
|
||
detail["patient"],
|
||
diagnosis,
|
||
)
|
||
application.processEvents()
|
||
|
||
assert page.diagnosis_card.objectName() == "ReceptionMedicationCaseCard"
|
||
assert page.diagnosis_card.accessibleName() == "完整病例"
|
||
assert page.medication_scroll.objectName() == "ReceptionMedicationScroll"
|
||
assert page.medication_scroll.accessibleName() == "用药记录与完整病例滚动区域"
|
||
assert page.medication_scroll.horizontalScrollBar().maximum() == 0
|
||
assert page.medication_scroll.verticalScrollBar().maximum() > 0
|
||
|
||
assert page.diagnosis_text.isHidden()
|
||
assert page.diagnosis_text.textFormat() == Qt.TextFormat.PlainText
|
||
assert page._case_summary_text == page.diagnosis_text.text()
|
||
assert "<" not in page._case_summary_text
|
||
assert "【完整病例末尾】" in page._case_summary_text
|
||
assert page.case_metric_labels["diagnosis_date"].text() == "未记录"
|
||
assert page.case_metric_labels["fasting_glucose"].text() == "6.8 mmol/L"
|
||
assert page.case_metric_labels["fasting_glucose"].property("alert") is True
|
||
assert page.case_metric_labels["blood_pressure"].text() == "168/102 mmHg"
|
||
|
||
medication = page.case_section_frames["medication"]
|
||
assert medication.isVisible()
|
||
medication_captions = [
|
||
label.text()
|
||
for label in medication.findChildren(QLabel, "ReceptionCaseFieldCaption")
|
||
]
|
||
medication_values = [
|
||
label
|
||
for label in medication.findChildren(QLabel, "ReceptionCaseFieldValue")
|
||
]
|
||
assert medication_captions == ["当前用药", "过敏史"]
|
||
assert medication_values[0].text().startswith("二甲双胍缓释片")
|
||
assert medication_values[0].property("important") is True
|
||
assert medication_values[1].property("risk") is True
|
||
|
||
title = page.findChild(QLabel, "ReceptionMedicationCaseTitle")
|
||
caption = medication.findChild(QLabel, "ReceptionCaseFieldCaption")
|
||
value = medication.findChild(QLabel, "ReceptionCaseFieldValue")
|
||
assert title is not None and title.font().pixelSize() == 18
|
||
assert caption is not None and caption.font().pixelSize() == 12
|
||
assert value is not None and value.font().pixelSize() == 14
|
||
assert title.font().weight() >= 700
|
||
assert caption.font().weight() >= 600
|
||
assert value.font().weight() >= 700
|
||
assert title.palette().color(QPalette.ColorRole.WindowText).name() == "#17264d"
|
||
assert caption.palette().color(QPalette.ColorRole.WindowText).name() == "#617092"
|
||
assert value.palette().color(QPalette.ColorRole.WindowText).name() == "#253a83"
|
||
assert value.wordWrap()
|
||
assert value.textInteractionFlags() & Qt.TextInteractionFlag.TextSelectableByMouse
|
||
|
||
for _ in range(5):
|
||
page.medication_scroll.verticalScrollBar().setValue(
|
||
page.medication_scroll.verticalScrollBar().maximum()
|
||
)
|
||
application.processEvents()
|
||
assert (
|
||
page.medication_scroll.verticalScrollBar().value()
|
||
== page.medication_scroll.verticalScrollBar().maximum()
|
||
)
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_current_detail_finishes_without_waiting_for_stale_patient_request(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
jobs.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(object(), PermissionSet(["*"]))
|
||
first = {"id": 71, "patient_id": 171, "diagnosis_id": 271, "patient_name": "慢患者"}
|
||
second = {"id": 72, "patient_id": 172, "diagnosis_id": 272, "patient_name": "当前患者"}
|
||
|
||
page._select_record(first)
|
||
page._select_record(second)
|
||
jobs[1]["on_success"]({"detail": _detail(72, name="当前患者")})
|
||
jobs[1]["on_finished"]()
|
||
|
||
assert not page._detail_loading
|
||
assert page.ai_button.isEnabled()
|
||
|
||
jobs[0]["on_finished"]()
|
||
assert not page._detail_loading
|
||
assert page._selected_appointment_id == 72
|
||
assert page.ai_button.isEnabled()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_load_more_accumulates_to_total_boundary(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
all_rows = [
|
||
{
|
||
"id": index,
|
||
"patient_id": 1000 + index,
|
||
"diagnosis_id": 2000 + index,
|
||
"patient_name": f"患者{index:02d}",
|
||
"status": 1,
|
||
}
|
||
for index in range(1, 23)
|
||
]
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
start = (kwargs["page_no"] - 1) * kwargs["page_size"]
|
||
return {
|
||
"lists": all_rows[start : start + kwargs["page_size"]],
|
||
"count": len(all_rows),
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
row = all_rows[appointment_id - 1]
|
||
return {
|
||
"appointment": row,
|
||
"diagnosis": {"id": row["diagnosis_id"], "patient_id": row["patient_id"]},
|
||
}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.refresh()
|
||
assert page.queue_list.count() == 15
|
||
assert page._queue_has_more()
|
||
assert not hasattr(page, "load_more_button")
|
||
assert not hasattr(page, "add_patient_button")
|
||
|
||
page._load_more()
|
||
assert [call["page_no"] for call in calls] == [1, 2]
|
||
assert all(call["page_size"] == 15 for call in calls)
|
||
assert page.queue_list.count() == 22
|
||
assert page.queue_summary.text() == "已加载 22 / 共 22 位患者"
|
||
assert not page._queue_has_more()
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_silent_poll_keeps_already_loaded_queue_pages(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
all_rows = [
|
||
{
|
||
"id": index,
|
||
"patient_id": 1000 + index,
|
||
"diagnosis_id": 2000 + index,
|
||
"patient_name": f"患者{index:02d}",
|
||
"status": 1,
|
||
}
|
||
for index in range(1, 28)
|
||
]
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(dict(kwargs))
|
||
start = (kwargs["page_no"] - 1) * kwargs["page_size"]
|
||
return {
|
||
"lists": all_rows[start : start + kwargs["page_size"]],
|
||
"count": len(all_rows),
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
row = all_rows[appointment_id - 1]
|
||
return {
|
||
"appointment": row,
|
||
"diagnosis": {"id": row["diagnosis_id"], "patient_id": row["patient_id"]},
|
||
}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.refresh()
|
||
page._load_more()
|
||
assert page.queue_list.count() == 27
|
||
assert page._queue_page == 2
|
||
calls.clear()
|
||
page.refresh(silent=True)
|
||
assert page.queue_list.count() == 27
|
||
assert page._queue_page == 2
|
||
assert calls[-1]["page_no"] == 1
|
||
assert calls[-1]["page_size"] >= 27
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_shows_loading_indicator_while_page_request_is_open(
|
||
application: QApplication,
|
||
queued_async: list[dict[str, Any]],
|
||
) -> None:
|
||
class Repository:
|
||
def list_appointments(self, **_kwargs: Any) -> dict[str, Any]:
|
||
return {
|
||
"lists": [
|
||
{"id": 1, "patient_name": "加载患者", "status": 1},
|
||
],
|
||
"count": 16,
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
return {
|
||
"appointment": {
|
||
"id": appointment_id,
|
||
"patient_id": 1001,
|
||
"status": 1,
|
||
},
|
||
"diagnosis": {"id": 2001, "patient_id": 1001},
|
||
}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.refresh()
|
||
assert page.queue_loading_indicator.isVisibleTo(page)
|
||
assert page.queue_loading_indicator.label.text() == "正在加载…"
|
||
assert page.queue_loading_indicator.spinner._timer.isActive()
|
||
|
||
queued_async[0]["on_success"](
|
||
{
|
||
"lists": [{"id": 1, "patient_name": "加载患者", "status": 1}],
|
||
"count": 16,
|
||
}
|
||
)
|
||
queued_async[0]["on_finished"]()
|
||
assert not page.queue_loading_indicator.isVisibleTo(page)
|
||
assert not page.queue_loading_indicator.spinner._timer.isActive()
|
||
|
||
page._load_more()
|
||
assert page.queue_loading_indicator.isVisibleTo(page)
|
||
load_more_job = queued_async[-1]
|
||
load_more_job["on_success"](
|
||
{
|
||
"lists": [{"id": 2, "patient_name": "第二页患者", "status": 1}],
|
||
"count": 16,
|
||
}
|
||
)
|
||
load_more_job["on_finished"]()
|
||
assert not page.queue_loading_indicator.isVisibleTo(page)
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_scrolls_to_bottom_loads_next_page(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
all_rows = [
|
||
{
|
||
"id": index,
|
||
"patient_id": 1000 + index,
|
||
"diagnosis_id": 2000 + index,
|
||
"patient_name": f"患者{index:02d}",
|
||
"status": 1,
|
||
}
|
||
for index in range(1, 23)
|
||
]
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
start = (kwargs["page_no"] - 1) * kwargs["page_size"]
|
||
return {
|
||
"lists": all_rows[start : start + kwargs["page_size"]],
|
||
"count": len(all_rows),
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
row = all_rows[appointment_id - 1]
|
||
return {
|
||
"appointment": row,
|
||
"diagnosis": {"id": row["diagnosis_id"], "patient_id": row["patient_id"]},
|
||
}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.resize(1280, 640)
|
||
page.show()
|
||
application.processEvents()
|
||
assert page.queue_list.count() == 15
|
||
scrollbar = page.queue_list.verticalScrollBar()
|
||
assert scrollbar.maximum() > 0
|
||
scrollbar.setValue(scrollbar.maximum())
|
||
application.processEvents()
|
||
assert [call["page_no"] for call in calls] == [1, 2]
|
||
assert page.queue_list.count() == 22
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_search_and_tab_changes_reset_accumulated_pages(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
if kwargs["status"] == 4:
|
||
rows = [{"id": 401, "patient_name": "过号患者", "status": 4}]
|
||
elif kwargs["patient_name"]:
|
||
rows = [{"id": 201, "patient_name": "搜索患者", "status": 1}]
|
||
else:
|
||
rows = [
|
||
{"id": index, "patient_name": f"患者{index}", "status": 1}
|
||
for index in range(1, 19)
|
||
]
|
||
start = (kwargs["page_no"] - 1) * kwargs["page_size"]
|
||
return {"lists": rows[start : start + kwargs["page_size"]], "count": len(rows)}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
return {
|
||
"appointment": {
|
||
"id": appointment_id,
|
||
"patient_id": appointment_id + 1000,
|
||
"status": 1,
|
||
},
|
||
"diagnosis": {
|
||
"id": appointment_id + 2000,
|
||
"patient_id": appointment_id + 1000,
|
||
},
|
||
}
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.refresh()
|
||
page._load_more()
|
||
assert page.queue_list.count() == 18
|
||
|
||
page.search_edit.setText("搜索")
|
||
page.refresh()
|
||
assert page.queue_list.count() == 1
|
||
assert page._queue_page == 1
|
||
assert calls[-1]["patient_name"] == "搜索"
|
||
|
||
page.queue_tabs.setCurrentIndex(1)
|
||
assert page.queue_list.count() == 1
|
||
assert page._queue_page == 1
|
||
assert calls[-1]["status"] == 4
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_queue_worker_uses_frozen_widget_snapshot(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[Any] = []
|
||
calls: list[dict[str, Any]] = []
|
||
|
||
def queue_async(function: Any, **_options: Any) -> object:
|
||
jobs.append(function)
|
||
return object()
|
||
|
||
class Repository:
|
||
def list_appointments(self, **kwargs: Any) -> dict[str, Any]:
|
||
calls.append(kwargs)
|
||
return {"lists": [], "count": 0}
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page.search_edit.setText("甲患者")
|
||
page.refresh()
|
||
page.search_edit.blockSignals(True)
|
||
page.search_edit.setText("乙患者")
|
||
page.search_edit.blockSignals(False)
|
||
page.queue_tabs.blockSignals(True)
|
||
page.queue_tabs.setCurrentIndex(1)
|
||
page.queue_tabs.blockSignals(False)
|
||
|
||
jobs[0]()
|
||
assert calls == [
|
||
{
|
||
"status": 1,
|
||
"start_date": date.today().isoformat(),
|
||
"end_date": date.today().isoformat(),
|
||
"page_no": 1,
|
||
"page_size": 15,
|
||
"patient_name": "甲患者",
|
||
}
|
||
]
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_phone_permission_and_ungated_notify_video_actions(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(31, name="脱敏患者")
|
||
|
||
class Repository:
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 31
|
||
return detail
|
||
|
||
masked_page = ReceptionPage(Repository(), PermissionSet([]))
|
||
masked_page._select_record(detail["appointment"])
|
||
assert masked_page.patient_labels["phone"].text() == "138****8000"
|
||
assert not masked_page.notify_button.isHidden()
|
||
assert not masked_page.video_button.isHidden()
|
||
|
||
plain_page = ReceptionPage(Repository(), PermissionSet(["tcm.diagnosis/phonePlain"]))
|
||
plain_page._select_record(detail["appointment"])
|
||
assert plain_page.patient_labels["phone"].text() == "13800138000"
|
||
assert plain_page.appointment_labels["doctor"].text() == "张医生"
|
||
assert plain_page.appointment_labels["assistant"].text() == "李医助"
|
||
assert plain_page.patient_labels["region"].text() == "浙江省杭州市"
|
||
|
||
masked_page.close()
|
||
plain_page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_ai_report_button_follows_permission(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
hidden = ReceptionPage(DemoDoctorRepository(), PermissionSet([]))
|
||
assert hidden.ai_button.isHidden()
|
||
hidden.close()
|
||
|
||
page = ReceptionPage(
|
||
DemoDoctorRepository(),
|
||
PermissionSet(["doctor.appointment/reception"]),
|
||
)
|
||
assert not page.ai_button.isHidden()
|
||
assert not page.ai_button.isEnabled()
|
||
assert not page.ai_question_edit.isEnabled()
|
||
assert not page.ai_send_button.isEnabled()
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_assistant_maps_prompt_to_report_model(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
detail = _detail(32, name="AI 患者")
|
||
opened: list[dict[str, Any]] = []
|
||
|
||
class Repository:
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 32
|
||
return detail
|
||
|
||
monkeypatch.setattr(
|
||
reception_module,
|
||
"present_diagnosis_ai_assistant",
|
||
lambda repository, parent, **options: opened.append(options),
|
||
)
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["doctor.appointment/reception", "tcm.diagnosis/aiAssistant"]),
|
||
)
|
||
page._select_record(detail["appointment"])
|
||
|
||
assert page.ai_question_edit.isEnabled()
|
||
assert page.ai_question_edit.maxLength() == 500
|
||
assert all(button.isEnabled() for button in page.ai_prompt_buttons)
|
||
page._open_ai_report_for_prompt("并发症筛查")
|
||
page.ai_question_edit.setText("请给出中药用药调整建议")
|
||
page._submit_ai_question()
|
||
|
||
assert [item["diagnosis_id"] for item in opened] == [232, 232]
|
||
assert [item["task"] for item in opened] == [
|
||
"complication_risk",
|
||
"medication_review",
|
||
]
|
||
assert opened[1]["prompt"] == "请给出中药用药调整建议"
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_auto_loads_structured_ai_analysis_and_matches_reference_geometry(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(33, name="智能分析患者")
|
||
analysis_calls: list[tuple[int, str]] = []
|
||
|
||
class Repository:
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 33
|
||
return detail
|
||
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
analysis_calls.append((diagnosis_id, model))
|
||
return _analysis_payload(model=model)
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(
|
||
[
|
||
"tcm.diagnosis/aiAnalysis",
|
||
"tcm.diagnosis/aiAssistant",
|
||
]
|
||
),
|
||
)
|
||
page.resize(1494, 832)
|
||
page.show()
|
||
page._select_record(detail["appointment"])
|
||
application.processEvents()
|
||
|
||
assert analysis_calls == [(233, "qwen"), (233, "openai")]
|
||
assert [
|
||
(
|
||
page.ai_analysis_model_selector.itemText(index),
|
||
page.ai_analysis_model_selector.itemData(index),
|
||
)
|
||
for index in range(page.ai_analysis_model_selector.count())
|
||
] == [("千问", "qwen"), ("OpenAI", "openai")]
|
||
assert page.ai_analysis_model_selector.currentData() == "qwen"
|
||
assert page._ai_analysis_state == "success"
|
||
assert page.ai_analysis_stack.currentWidget() is page.ai_analysis_content_page
|
||
assert page.ai_summary_label.isVisible()
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
assert page.ai_treatment_label.text().startswith("建议调整降糖方案")
|
||
assert page.ai_summary_label.textFormat() == Qt.TextFormat.PlainText
|
||
assert page.ai_treatment_label.textFormat() == Qt.TextFormat.PlainText
|
||
chips = [
|
||
label
|
||
for label in page.ai_risk_chip_host.findChildren(QLabel)
|
||
if label.property("receptionRiskChip")
|
||
]
|
||
assert [label.text() for label in chips] == ["高血糖风险", "心血管风险", "肾脏风险"]
|
||
assert all(label.textFormat() == Qt.TextFormat.PlainText for label in chips)
|
||
assert [label.property("riskLevel") for label in chips] == ["high", "medium", "low"]
|
||
assert "千问" in page.ai_analysis_title.toolTip()
|
||
assert [button.text() for button in page.ai_prompt_buttons] == [
|
||
"基于当前病史，下一步检查建议？",
|
||
"该患者的用药调整建议？",
|
||
"糖尿病教育要点有哪些？",
|
||
"并发症筛查建议？",
|
||
]
|
||
assert page.ai_send_button.text() == ""
|
||
assert page.ai_send_button.accessibleName() == "发送 AI 问诊问题"
|
||
expand_button = page.ai_analysis_expand_button
|
||
assert expand_button.objectName() == "ReceptionAiAnalysisExpandButton"
|
||
assert expand_button.property("iconKind") == "expand-corners"
|
||
assert expand_button.text() == ""
|
||
assert expand_button.accessibleName() == "查看完整 AI 智能分析"
|
||
assert expand_button.toolTip() == "放大查看 AI 智能分析"
|
||
assert expand_button.isEnabled()
|
||
assert expand_button.focusPolicy() != Qt.FocusPolicy.NoFocus
|
||
assert expand_button.size().width() == expand_button.size().height() == 28
|
||
title_right = page.ai_analysis_title.mapTo(
|
||
page.ai_analysis_card,
|
||
QPoint(page.ai_analysis_title.width(), 0),
|
||
).x()
|
||
button_left = expand_button.mapTo(page.ai_analysis_card, QPoint(0, 0)).x()
|
||
button_right = expand_button.mapTo(
|
||
page.ai_analysis_card,
|
||
QPoint(expand_button.width(), 0),
|
||
).x()
|
||
assert button_left > title_right
|
||
assert button_right <= page.ai_analysis_card.contentsRect().right() + 1
|
||
|
||
left = page.ai_analysis_card.geometry()
|
||
right = page.ai_assistant_card.geometry()
|
||
assert 470 <= left.height() <= 520
|
||
assert left.height() == right.height()
|
||
assert 0 <= right.left() - left.right() - 1 <= 2
|
||
assert abs(left.width() * 5 - right.width() * 4) <= 10
|
||
assert page.ai_summary_label.width() <= page.ai_analysis_card.contentsRect().width()
|
||
assert page.ai_summary_label.minimumSizeHint().width() <= 48
|
||
|
||
calls_before_switch = list(analysis_calls)
|
||
page.ai_analysis_model_selector.setCurrentIndex(
|
||
page.ai_analysis_model_selector.findData("openai")
|
||
)
|
||
assert page.ai_analysis_model_selector.currentData() == "openai"
|
||
assert page.ai_summary_label.text().startswith("OpenAI：")
|
||
page.ai_analysis_model_selector.setCurrentIndex(
|
||
page.ai_analysis_model_selector.findData("qwen")
|
||
)
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
assert analysis_calls == calls_before_switch
|
||
assert page.detail_scroll.objectName() == "ReceptionDetailScroll"
|
||
assert (
|
||
page.detail_scroll.verticalScrollBarPolicy()
|
||
== Qt.ScrollBarPolicy.ScrollBarAsNeeded
|
||
)
|
||
page.resize(1494, 620)
|
||
application.processEvents()
|
||
assert page.detail_scroll.verticalScrollBar().maximum() > 0
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_dual_ai_queues_openai_only_after_qwen_is_visible(
|
||
application: QApplication,
|
||
queued_async: list[dict[str, Any]],
|
||
) -> None:
|
||
detail = _detail(83, name="双模型排队患者")
|
||
calls: list[tuple[int, str]] = []
|
||
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
calls.append((diagnosis_id, model))
|
||
return _analysis_payload(f"{model} 排队风险", model=model)
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page._select_record(detail["appointment"])
|
||
assert len(queued_async) == 1
|
||
|
||
queued_async[0]["on_success"]({"detail": detail, "warnings": []})
|
||
queued_async[0]["on_finished"]()
|
||
assert len(queued_async) == 2
|
||
assert calls == []
|
||
assert page.ai_analysis_model_selector.currentData() == "qwen"
|
||
assert page._ai_analysis_model_states == {"qwen": "loading", "openai": "waiting"}
|
||
|
||
qwen_result = queued_async[1]["function"](*queued_async[1]["args"])
|
||
assert calls == [(283, "qwen")]
|
||
queued_async[1]["on_success"](qwen_result)
|
||
|
||
assert len(queued_async) == 3
|
||
assert calls == [(283, "qwen")]
|
||
assert page.ai_analysis_model_selector.currentData() == "qwen"
|
||
assert page._ai_analysis_state == "success"
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
assert page._ai_analysis_payloads["qwen"] == qwen_result
|
||
assert page._ai_analysis_model_states["openai"] == "loading"
|
||
assert page.ai_analysis_secondary_status.property("secondaryState") == "loading"
|
||
assert "OpenAI 正在生成" in page.ai_analysis_secondary_status.text()
|
||
assert page.ai_analysis_expand_button.isEnabled()
|
||
|
||
openai_result = queued_async[2]["function"](*queued_async[2]["args"])
|
||
assert calls == [(283, "qwen"), (283, "openai")]
|
||
queued_async[2]["on_success"](openai_result)
|
||
assert page.ai_analysis_model_selector.currentData() == "qwen"
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
|
||
jobs_before_switch = len(queued_async)
|
||
calls_before_switch = list(calls)
|
||
page.ai_analysis_model_selector.setCurrentIndex(
|
||
page.ai_analysis_model_selector.findData("openai")
|
||
)
|
||
assert page.ai_summary_label.text().startswith("OpenAI：")
|
||
assert "openai 排队风险" in page.ai_risk_label.text()
|
||
page.ai_analysis_model_selector.setCurrentIndex(
|
||
page.ai_analysis_model_selector.findData("qwen")
|
||
)
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
assert len(queued_async) == jobs_before_switch
|
||
assert calls == calls_before_switch
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_openai_failure_keeps_qwen_success_visible(
|
||
application: QApplication,
|
||
queued_async: list[dict[str, Any]],
|
||
) -> None:
|
||
detail = _detail(84, name="第二模型失败患者")
|
||
calls: list[tuple[int, str]] = []
|
||
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
calls.append((diagnosis_id, model))
|
||
if model == "openai":
|
||
raise RuntimeError("openai upstream unavailable")
|
||
return _analysis_payload("保留的千问风险", model=model)
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page._select_record(detail["appointment"])
|
||
queued_async[0]["on_success"]({"detail": detail, "warnings": []})
|
||
qwen_result = queued_async[1]["function"](*queued_async[1]["args"])
|
||
queued_async[1]["on_success"](qwen_result)
|
||
assert len(queued_async) == 3
|
||
|
||
with pytest.raises(RuntimeError) as failure:
|
||
queued_async[2]["function"](*queued_async[2]["args"])
|
||
queued_async[2]["on_error"](failure.value)
|
||
|
||
assert calls == [(284, "qwen"), (284, "openai")]
|
||
assert page.ai_analysis_model_selector.currentData() == "qwen"
|
||
assert page._ai_analysis_state == "success"
|
||
assert page._ai_analysis_payload == qwen_result
|
||
assert page._ai_analysis_payloads == {"qwen": qwen_result}
|
||
assert page.ai_summary_label.text() == "2 型糖尿病，血糖控制不佳"
|
||
assert "保留的千问风险" in page.ai_risk_label.text()
|
||
assert page.ai_analysis_expand_button.isEnabled()
|
||
assert page.ai_analysis_secondary_status.property("secondaryState") == "error"
|
||
assert "千问结果已保留" in page.ai_analysis_secondary_status.text()
|
||
assert "OpenAI分析加载失败" in page.ai_analysis_secondary_status.text()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_ai_analysis_permission_error_retry_states(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(34, name="状态患者")
|
||
|
||
class Repository:
|
||
def __init__(self) -> None:
|
||
self.fail = True
|
||
self.calls: list[tuple[int, str]] = []
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
return detail
|
||
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
self.calls.append((diagnosis_id, model))
|
||
if self.fail:
|
||
raise RuntimeError("analysis unavailable")
|
||
return _analysis_payload("重试后风险", model=model)
|
||
|
||
denied_repository = Repository()
|
||
denied = ReceptionPage(denied_repository, PermissionSet([]))
|
||
denied._select_record(detail["appointment"])
|
||
assert denied_repository.calls == []
|
||
assert denied._ai_analysis_state == "permission"
|
||
assert denied._ai_analysis_payload is None
|
||
assert denied._ai_analysis_payloads == {}
|
||
assert not denied.ai_analysis_model_selector.isEnabled()
|
||
assert not denied.ai_analysis_expand_button.isEnabled()
|
||
assert "没有查看" in denied.ai_analysis_state_label.text()
|
||
assert denied.ai_analysis_retry_button.isHidden()
|
||
denied.close()
|
||
|
||
repository = Repository()
|
||
page = ReceptionPage(
|
||
repository,
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page.show()
|
||
page._select_record(detail["appointment"])
|
||
application.processEvents()
|
||
assert repository.calls == [(234, "qwen")]
|
||
assert page._ai_analysis_state == "error"
|
||
assert page._ai_analysis_payload is None
|
||
assert not page.ai_analysis_expand_button.isEnabled()
|
||
assert "加载失败" in page.ai_analysis_state_label.text()
|
||
assert not page.ai_analysis_retry_button.isHidden()
|
||
|
||
page._load_detail(detail["appointment"], silent=True, clear=False)
|
||
assert repository.calls == [(234, "qwen")]
|
||
assert page._ai_analysis_state == "error"
|
||
|
||
repository.fail = False
|
||
page.ai_analysis_retry_button.click()
|
||
application.processEvents()
|
||
assert repository.calls == [
|
||
(234, "qwen"),
|
||
(234, "qwen"),
|
||
(234, "openai"),
|
||
]
|
||
assert page._ai_analysis_state == "success"
|
||
assert page.ai_analysis_expand_button.isEnabled()
|
||
assert "重试后风险" in page.ai_risk_label.text()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_saved_diagnosis_forces_a_fresh_structured_ai_analysis(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(38, name="病历更新患者")
|
||
|
||
class Repository:
|
||
def __init__(self) -> None:
|
||
self.analysis_calls: list[tuple[int, str]] = []
|
||
self.model_counts = {"qwen": 0, "openai": 0}
|
||
|
||
def list_appointments(self, **_kwargs: Any) -> dict[str, Any]:
|
||
return {
|
||
"lists": [
|
||
{
|
||
**detail["appointment"],
|
||
"diagnosis_id": detail["diagnosis"]["id"],
|
||
}
|
||
],
|
||
"count": 1,
|
||
}
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 38
|
||
return detail
|
||
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
assert diagnosis_id == 238
|
||
self.analysis_calls.append((diagnosis_id, model))
|
||
self.model_counts[model] += 1
|
||
return _analysis_payload(
|
||
f"第 {self.model_counts[model]} 次{model}分析",
|
||
model=model,
|
||
)
|
||
|
||
repository = Repository()
|
||
page = ReceptionPage(
|
||
repository,
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page.refresh()
|
||
assert repository.analysis_calls == [(238, "qwen"), (238, "openai")]
|
||
|
||
page._diagnosis_saved()
|
||
assert repository.analysis_calls == [
|
||
(238, "qwen"),
|
||
(238, "openai"),
|
||
(238, "qwen"),
|
||
(238, "openai"),
|
||
]
|
||
assert "第 2 次qwen分析" in page.ai_risk_label.text()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_ai_analysis_dialog_switches_complete_cached_payloads_without_requests(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
detail = _detail(39, name="完整分析患者")
|
||
diagnosis_texts = {
|
||
"qwen": ("千问诊断建议需要完整展示，不能在卡片中被截断。" * 80)
|
||
+ "[千问诊断末尾]",
|
||
"openai": ("OpenAI 诊断建议需要独立完整展示并支持模型对照。" * 80)
|
||
+ "[OpenAI诊断末尾]",
|
||
}
|
||
treatment_texts = {
|
||
"qwen": ("千问治疗建议应支持长文本滚动阅读，并允许复制。" * 80)
|
||
+ "[千问治疗末尾]",
|
||
"openai": ("OpenAI 治疗建议应保留全部上下文与检查复核事项。" * 80)
|
||
+ "[OpenAI治疗末尾]",
|
||
}
|
||
payloads = {
|
||
model: {
|
||
"diagnosis_advice": diagnosis_texts[model],
|
||
"risk_assessment": [
|
||
{
|
||
"label": f"{model} 风险项目 {index}",
|
||
"level": ("high", "medium", "low")[(index - 1) % 3],
|
||
}
|
||
for index in range(1, 7)
|
||
],
|
||
"treatment_advice": treatment_texts[model],
|
||
"model_key": model,
|
||
"model_label": "千问" if model == "qwen" else "OpenAI",
|
||
"model_name": "qwen3.6-35b" if model == "qwen" else "gpt-5.2",
|
||
"generated_at": (
|
||
"2026-08-14 11:05:14"
|
||
if model == "qwen"
|
||
else "2026-08-14 11:06:20"
|
||
),
|
||
"raw_marker": {
|
||
"model": model,
|
||
"tail": f"{model}-payload-complete",
|
||
},
|
||
}
|
||
for model in ("qwen", "openai")
|
||
}
|
||
|
||
class Repository:
|
||
def __init__(self) -> None:
|
||
self.analysis_calls: list[tuple[int, str]] = []
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 39
|
||
return detail
|
||
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
assert diagnosis_id == 239
|
||
self.analysis_calls.append((diagnosis_id, model))
|
||
return payloads[model]
|
||
|
||
dialogs: list[Any] = []
|
||
|
||
def capture_dialog(dialog: Any) -> int:
|
||
dialogs.append(dialog)
|
||
dialog.show()
|
||
application.processEvents()
|
||
return 0
|
||
|
||
monkeypatch.setattr(reception_module._ReceptionAiAnalysisDialog, "exec", capture_dialog)
|
||
repository = Repository()
|
||
page = ReceptionPage(
|
||
repository,
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
page.resize(1494, 832)
|
||
page.show()
|
||
page._select_record(detail["appointment"])
|
||
application.processEvents()
|
||
|
||
card_risks = [
|
||
label
|
||
for label in page.ai_risk_chip_host.findChildren(QLabel)
|
||
if label.property("receptionRiskChip")
|
||
]
|
||
assert [label.text() for label in card_risks] == [
|
||
"qwen 风险项目 1",
|
||
"qwen 风险项目 2",
|
||
"qwen 风险项目 3",
|
||
"qwen 风险项目 4",
|
||
]
|
||
calls_before_dialog = list(repository.analysis_calls)
|
||
page.ai_analysis_expand_button.click()
|
||
|
||
assert calls_before_dialog == [(239, "qwen"), (239, "openai")]
|
||
assert repository.analysis_calls == calls_before_dialog
|
||
assert len(dialogs) == 1
|
||
dialog = dialogs[0]
|
||
assert dialog.objectName() == "ReceptionAiAnalysisDialog"
|
||
assert dialog.parent() is page
|
||
assert dialog.isModal()
|
||
assert dialog.property("businessDialog") is True
|
||
scroll = dialog.findChild(QScrollArea, "ReceptionAiAnalysisDialogScroll")
|
||
assert scroll is not None
|
||
assert scroll.widgetResizable()
|
||
assert (
|
||
scroll.horizontalScrollBarPolicy()
|
||
== Qt.ScrollBarPolicy.ScrollBarAlwaysOff
|
||
)
|
||
diagnosis_label = dialog.findChild(
|
||
QLabel,
|
||
"ReceptionAiAnalysisDialogDiagnosisText",
|
||
)
|
||
treatment_label = dialog.findChild(
|
||
QLabel,
|
||
"ReceptionAiAnalysisDialogTreatmentText",
|
||
)
|
||
assert diagnosis_label is not None
|
||
assert treatment_label is not None
|
||
assert [
|
||
(dialog.model_selector.itemText(index), dialog.model_selector.itemData(index))
|
||
for index in range(dialog.model_selector.count())
|
||
] == [("千问", "qwen"), ("OpenAI", "openai")]
|
||
assert dialog.model_selector.currentData() == "qwen"
|
||
assert diagnosis_label.text() == diagnosis_texts["qwen"]
|
||
assert treatment_label.text() == treatment_texts["qwen"]
|
||
assert diagnosis_label.text().endswith("[千问诊断末尾]")
|
||
assert treatment_label.text().endswith("[千问治疗末尾]")
|
||
assert diagnosis_label.textFormat() == Qt.TextFormat.PlainText
|
||
assert treatment_label.textFormat() == Qt.TextFormat.PlainText
|
||
dialog_risks = [
|
||
label
|
||
for label in dialog.findChildren(QLabel)
|
||
if label.property("dialogAiRisk")
|
||
]
|
||
assert [label.text() for label in dialog_risks] == [
|
||
f"qwen 风险项目 {index}" for index in range(1, 7)
|
||
]
|
||
assert [label.property("riskLevel") for label in dialog_risks] == [
|
||
"high",
|
||
"medium",
|
||
"low",
|
||
"high",
|
||
"medium",
|
||
"low",
|
||
]
|
||
assert dialog.payloads == payloads
|
||
assert dialog.payloads["qwen"]["raw_marker"]["tail"] == "qwen-payload-complete"
|
||
|
||
dialog.model_selector.setCurrentIndex(dialog.model_selector.findData("openai"))
|
||
application.processEvents()
|
||
assert dialog.active_model == "openai"
|
||
assert diagnosis_label.text() == diagnosis_texts["openai"]
|
||
assert treatment_label.text() == treatment_texts["openai"]
|
||
assert diagnosis_label.text().endswith("[OpenAI诊断末尾]")
|
||
assert treatment_label.text().endswith("[OpenAI治疗末尾]")
|
||
assert "OpenAI" in dialog.meta_label.text()
|
||
dialog_risks = [
|
||
label
|
||
for label in dialog.findChildren(QLabel)
|
||
if label.property("dialogAiRisk")
|
||
]
|
||
assert [label.text() for label in dialog_risks] == [
|
||
f"openai 风险项目 {index}" for index in range(1, 7)
|
||
]
|
||
assert [label.property("riskLevel") for label in dialog_risks] == [
|
||
"high",
|
||
"medium",
|
||
"low",
|
||
"high",
|
||
"medium",
|
||
"low",
|
||
]
|
||
assert dialog.payloads["openai"]["raw_marker"]["tail"] == (
|
||
"openai-payload-complete"
|
||
)
|
||
assert repository.analysis_calls == calls_before_dialog
|
||
dialog.resize(640, 420)
|
||
application.processEvents()
|
||
assert scroll.verticalScrollBar().maximum() > 0
|
||
assert dialog.close_button.accessibleName() == "关闭 AI 智能分析详情"
|
||
dialog.close_button.click()
|
||
assert not dialog.isVisible()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_legacy_single_model_repository_survives_same_patient_detail_refresh(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
calls: list[int] = []
|
||
|
||
def queue_async(function: Any, **options: Any) -> object:
|
||
jobs.append({"function": function, **options})
|
||
return object()
|
||
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(self, diagnosis_id: int) -> dict[str, Any]:
|
||
calls.append(diagnosis_id)
|
||
result = _analysis_payload("刷新期间返回")
|
||
result["diagnosis_advice"] = "详情刷新后仍接收"
|
||
return result
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
detail = _detail(37, name="同一患者")
|
||
|
||
page._select_record(detail["appointment"])
|
||
jobs[0]["on_success"]({"detail": detail, "warnings": []})
|
||
assert len(jobs) == 2
|
||
assert page._ai_analysis_state == "loading"
|
||
|
||
page._load_detail(detail["appointment"], silent=True, clear=False)
|
||
assert len(jobs) == 3
|
||
jobs[2]["on_success"]({"detail": detail, "warnings": []})
|
||
assert len(jobs) == 3
|
||
|
||
result = jobs[1]["function"]()
|
||
jobs[1]["on_success"](result)
|
||
assert calls == [237]
|
||
assert len(jobs) == 3
|
||
assert page._ai_analysis_state == "success"
|
||
assert page.ai_summary_label.text() == "详情刷新后仍接收"
|
||
assert "刷新期间返回" in page.ai_risk_label.text()
|
||
assert page._ai_analysis_model_states["openai"] == "unsupported"
|
||
assert page.ai_analysis_secondary_status.property("secondaryState") == "unsupported"
|
||
assert "仅支持千问" in page.ai_analysis_secondary_status.text()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_detail_failure_stops_ai_loading_and_keeps_retry_available(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
jobs.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
raise AssertionError(
|
||
"queued async test must not execute AI worker inline: "
|
||
f"{diagnosis_id}/{model}"
|
||
)
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
record = {
|
||
"id": 81,
|
||
"patient_id": 181,
|
||
"diagnosis_id": 281,
|
||
"patient_name": "失败患者",
|
||
}
|
||
|
||
page._select_record(record)
|
||
assert page._ai_analysis_state == "loading"
|
||
jobs[0]["on_error"](RuntimeError("detail unavailable"))
|
||
|
||
assert page._ai_analysis_state == "error"
|
||
assert not page._ai_analysis_loading
|
||
assert "病历加载失败" in page.ai_analysis_state_label.text()
|
||
assert not page.ai_analysis_retry_button.isHidden()
|
||
|
||
jobs[0]["on_finished"]()
|
||
assert not page._detail_loading
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_ai_success_clears_loading_before_worker_finished(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[dict[str, Any]] = []
|
||
|
||
def queue_async(_function: Any, **options: Any) -> object:
|
||
jobs.append(options)
|
||
return object()
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
raise AssertionError(
|
||
"queued async test must not execute AI worker inline: "
|
||
f"{diagnosis_id}/{model}"
|
||
)
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
detail = _detail(82, name="成功患者")
|
||
|
||
page._select_record(detail["appointment"])
|
||
jobs[0]["on_success"]({"detail": detail, "warnings": []})
|
||
assert len(jobs) == 2
|
||
assert page._ai_analysis_loading
|
||
|
||
jobs[1]["on_success"](_analysis_payload())
|
||
assert len(jobs) == 3
|
||
assert page._ai_analysis_state == "success"
|
||
assert not page._ai_analysis_loading
|
||
assert page._ai_analysis_model_states["openai"] == "loading"
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_reception_ai_analysis_discards_late_qwen_and_openai_results(
|
||
application: QApplication,
|
||
queued_async: list[dict[str, Any]],
|
||
) -> None:
|
||
class Repository:
|
||
def get_diagnosis_ai_analysis(
|
||
self,
|
||
diagnosis_id: int,
|
||
model: str,
|
||
) -> dict[str, Any]:
|
||
raise AssertionError("queued async test must not execute worker inline")
|
||
|
||
page = ReceptionPage(
|
||
Repository(),
|
||
PermissionSet(["tcm.diagnosis/aiAnalysis"]),
|
||
)
|
||
first = _detail(35, name="第一位患者")
|
||
second = _detail(36, name="第二位患者")
|
||
third = _detail(37, name="第三位患者")
|
||
|
||
page._select_record(first["appointment"])
|
||
queued_async[0]["on_success"]({"detail": first, "warnings": []})
|
||
assert len(queued_async) == 2
|
||
|
||
page._select_record(second["appointment"])
|
||
queued_async[2]["on_success"]({"detail": second, "warnings": []})
|
||
assert len(queued_async) == 4
|
||
second_qwen = _analysis_payload("第二位千问风险")
|
||
second_qwen["diagnosis_advice"] = "第二位患者千问分析"
|
||
queued_async[3]["on_success"](second_qwen)
|
||
assert len(queued_async) == 5
|
||
|
||
page._select_record(third["appointment"])
|
||
queued_async[5]["on_success"]({"detail": third, "warnings": []})
|
||
assert len(queued_async) == 7
|
||
third_qwen = _analysis_payload("第三位千问风险")
|
||
third_qwen["diagnosis_advice"] = "第三位患者千问分析"
|
||
queued_async[6]["on_success"](third_qwen)
|
||
assert len(queued_async) == 8
|
||
third_openai = _analysis_payload("第三位 OpenAI 风险", model="openai")
|
||
third_openai["diagnosis_advice"] = "第三位患者 OpenAI 分析"
|
||
queued_async[7]["on_success"](third_openai)
|
||
|
||
assert page._selected_appointment_id == 37
|
||
assert page._ai_analysis_payloads == {
|
||
"qwen": third_qwen,
|
||
"openai": third_openai,
|
||
}
|
||
assert page.ai_summary_label.text() == "第三位患者千问分析"
|
||
|
||
jobs_before_stale_results = len(queued_async)
|
||
stale_qwen = _analysis_payload("第一位迟到千问风险")
|
||
stale_qwen["diagnosis_advice"] = "第一位患者迟到千问结果"
|
||
queued_async[1]["on_success"](stale_qwen)
|
||
stale_openai = _analysis_payload("第二位迟到 OpenAI 风险", model="openai")
|
||
stale_openai["diagnosis_advice"] = "第二位患者迟到 OpenAI 结果"
|
||
queued_async[4]["on_success"](stale_openai)
|
||
|
||
assert len(queued_async) == jobs_before_stale_results
|
||
assert page._selected_appointment_id == 37
|
||
assert page._ai_analysis_payloads == {
|
||
"qwen": third_qwen,
|
||
"openai": third_openai,
|
||
}
|
||
assert page.ai_summary_label.text() == "第三位患者千问分析"
|
||
assert "第三位千问风险" in page.ai_risk_label.text()
|
||
page.ai_analysis_model_selector.setCurrentIndex(
|
||
page.ai_analysis_model_selector.findData("openai")
|
||
)
|
||
assert page.ai_summary_label.text() == "第三位患者 OpenAI 分析"
|
||
assert "第三位 OpenAI 风险" in page.ai_risk_label.text()
|
||
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_video_payload_keeps_three_identifiers_distinct(
|
||
application: QApplication,
|
||
immediate_async: None,
|
||
) -> None:
|
||
detail = _detail(41, name="视频患者")
|
||
|
||
class Repository:
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
assert appointment_id == 41
|
||
return detail
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
page._select_record(detail["appointment"])
|
||
emitted: list[dict[str, Any]] = []
|
||
page.video_requested.connect(emitted.append)
|
||
page._request_video()
|
||
|
||
assert emitted == [
|
||
{
|
||
"source": "reception",
|
||
"appointment_id": 41,
|
||
"patient_id": 141,
|
||
"diagnosis_id": 241,
|
||
"patient_name": "视频患者",
|
||
"mode": "im",
|
||
"record": detail["appointment"],
|
||
}
|
||
]
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_completion_revalidates_server_status_before_write(
|
||
application: QApplication,
|
||
) -> None:
|
||
completed: list[int] = []
|
||
|
||
class Repository:
|
||
status = 3
|
||
|
||
def get_reception(self, appointment_id: int) -> dict[str, Any]:
|
||
return {
|
||
"appointment": {
|
||
"id": appointment_id,
|
||
"patient_id": 151,
|
||
"status": self.status,
|
||
}
|
||
}
|
||
|
||
def complete_appointment(self, appointment_id: int) -> dict[str, bool]:
|
||
completed.append(appointment_id)
|
||
return {"ok": True}
|
||
|
||
repository = Repository()
|
||
page = ReceptionPage(repository, PermissionSet(["doctor.appointment/complete"]))
|
||
|
||
with pytest.raises(ValueError, match="状态已变化"):
|
||
page._complete_after_revalidation(51)
|
||
assert completed == []
|
||
|
||
repository.status = 4
|
||
assert page._complete_after_revalidation(51) == {"ok": True}
|
||
assert completed == [51]
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_note_limit_and_attachment_payload_contract(
|
||
application: QApplication,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
jobs: list[tuple[Any, dict[str, Any]]] = []
|
||
received: list[dict[str, Any]] = []
|
||
uploads: list[dict[str, Any]] = []
|
||
|
||
def queue_async(function: Any, **options: Any) -> object:
|
||
jobs.append((function, options))
|
||
return object()
|
||
|
||
class Repository:
|
||
def upload_material(self, **kwargs: Any) -> str:
|
||
uploads.append(kwargs)
|
||
suffix = "tongue.jpg" if kwargs["material_type"] == "image" else "report.pdf"
|
||
return f"/uploads/{kwargs['material_type']}/{suffix}"
|
||
|
||
def add_doctor_note(self, diagnosis_id: int, content: str, **kwargs: Any) -> None:
|
||
received.append({"diagnosis_id": diagnosis_id, "content": content, **kwargs})
|
||
|
||
monkeypatch.setattr(reception_module, "run_async", queue_async)
|
||
page = ReceptionPage(Repository(), PermissionSet(["doctor.appointment/addDoctorNote"]))
|
||
detail = _detail(61, name="备注患者")
|
||
page._selected_record = detail["appointment"]
|
||
page._selected_appointment_id = 61
|
||
page._selected_detail = detail
|
||
page._detail_generation = 7
|
||
page.note_edit.setPlainText("字" * (NOTE_LIMIT + 20))
|
||
page._pending_tongue_images = [r"C:\records\tongue.jpg"]
|
||
page._pending_report_files = [r"C:\records\report.pdf"]
|
||
|
||
assert len(page.note_edit.toPlainText()) == NOTE_LIMIT
|
||
assert page.note_counter.text() == f"{NOTE_LIMIT} / {NOTE_LIMIT}"
|
||
page._save_note()
|
||
assert len(jobs) == 1
|
||
jobs[0][0]()
|
||
|
||
assert uploads == [
|
||
{"path": r"C:\records\tongue.jpg", "material_type": "image", "cid": 0},
|
||
{"path": r"C:\records\report.pdf", "material_type": "file", "cid": 0},
|
||
]
|
||
assert received == [
|
||
{
|
||
"diagnosis_id": 261,
|
||
"content": "字" * NOTE_LIMIT,
|
||
"tongue_images": ["/uploads/image/tongue.jpg"],
|
||
"report_files": ["/uploads/file/report.pdf"],
|
||
}
|
||
]
|
||
jobs[0][1]["on_finished"]()
|
||
assert not page._note_busy
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_remote_note_uses_multipart_then_server_urls_only(
|
||
application: QApplication,
|
||
tmp_path: Path,
|
||
) -> None:
|
||
requests: list[httpx.Request] = []
|
||
tongue = tmp_path / "tongue.jpg"
|
||
report = tmp_path / "report.pdf"
|
||
tongue.write_bytes(b"tongue-image")
|
||
report.write_bytes(b"report-file")
|
||
|
||
def handler(request: httpx.Request) -> httpx.Response:
|
||
requests.append(request)
|
||
if request.url.path.endswith("/upload/image"):
|
||
return httpx.Response(
|
||
200,
|
||
json={"code": 1, "data": {"uri": "/materials/tongue.jpg"}},
|
||
)
|
||
if request.url.path.endswith("/upload/file"):
|
||
return httpx.Response(
|
||
200,
|
||
json={"code": 1, "data": {"url": "https://cdn.test/report.pdf"}},
|
||
)
|
||
return httpx.Response(200, json={"code": 1, "data": {"id": 9}})
|
||
|
||
with ApiClient(
|
||
"https://example.test",
|
||
transport=httpx.MockTransport(handler),
|
||
) as client:
|
||
repository = RemoteDoctorRepository(client)
|
||
page = ReceptionPage(repository, PermissionSet([]))
|
||
result = page._upload_and_add_note(
|
||
501,
|
||
"两阶段备注",
|
||
[str(tongue)],
|
||
[str(report)],
|
||
)
|
||
|
||
assert result == {"id": 9}
|
||
assert [request.url.path.rsplit("/", 2)[-2:] for request in requests] == [
|
||
["upload", "image"],
|
||
["upload", "file"],
|
||
["doctor.appointment", "addDoctorNote"],
|
||
]
|
||
for upload_request in requests[:2]:
|
||
assert upload_request.headers["content-type"].startswith("multipart/form-data; boundary=")
|
||
assert b'name="cid"' in upload_request.content
|
||
assert b"\r\n0\r\n" in upload_request.content
|
||
note_payload = json.loads(requests[-1].content)
|
||
assert note_payload == {
|
||
"diagnosis_id": 501,
|
||
"content": "两阶段备注",
|
||
"tongue_images": ["/materials/tongue.jpg"],
|
||
"report_files": ["https://cdn.test/report.pdf"],
|
||
}
|
||
assert str(tmp_path) not in requests[-1].content.decode("utf-8")
|
||
page.close()
|
||
application.processEvents()
|
||
|
||
|
||
def test_partial_upload_failure_never_submits_note(
|
||
application: QApplication,
|
||
) -> None:
|
||
submitted: list[dict[str, Any]] = []
|
||
|
||
class Repository:
|
||
def upload_material(self, path: str, material_type: str, cid: int = 0) -> str:
|
||
del material_type, cid
|
||
if path.endswith("bad.pdf"):
|
||
raise OSError("磁盘读取失败")
|
||
return "/materials/good.jpg"
|
||
|
||
def add_doctor_note(self, **kwargs: Any) -> None:
|
||
submitted.append(kwargs)
|
||
|
||
page = ReceptionPage(Repository(), PermissionSet([]))
|
||
with pytest.raises(RuntimeError, match="bad.pdf.*上传失败"):
|
||
page._upload_and_add_note(
|
||
501,
|
||
"不会提交",
|
||
[r"C:\records\good.jpg"],
|
||
[r"C:\records\bad.pdf"],
|
||
)
|
||
assert submitted == []
|
||
page.close()
|
||
application.processEvents()
|