This commit is contained in:
Your Name
2026-08-28 18:24:37 +08:00
parent 43ad07208f
commit ed48f8be31
383 changed files with 8673 additions and 2222 deletions
+136 -1
View File
@@ -12,9 +12,11 @@ from PySide6.QtCore import Qt
from PySide6.QtTest import QTest
from PySide6.QtWidgets import (
QApplication,
QDialog,
QLabel,
QPushButton,
QTextBrowser,
QVBoxLayout,
QWidget,
)
@@ -1506,9 +1508,17 @@ def test_sectioned_reply_becomes_a_structured_report(application: QApplication)
titles = [
label.text()
for label in panel.findChildren(QLabel)
if label.objectName() == "AiConsultClinicalSectionTitle"
if label.objectName() == "AiConsultReportSectionTitle"
]
assert titles == ["症状演变与疗效评估", "血糖控制与监测细节", "用药依从性与生活方式"]
# 要点不再每句一个方框,而是「小标题 + 正文」两级文字。
assert panel.findChildren(ai_consult_module.QFrame, "AiConsultGapRow") == []
leads = [
label.text()
for label in panel.findChildren(QLabel)
if label.objectName() == "AiConsultReportLead"
]
assert leads == ["麻木症状", "皮肤瘙痒", "空腹血糖波动", "西药服用情况"]
bubble.deleteLater()
@@ -1522,3 +1532,128 @@ def test_sectioned_reply_becomes_a_structured_report(application: QApplication)
)
def test_unsectioned_replies_stay_plain_text(reply: str, application: QApplication) -> None:
assert ai_consult_module.parse_structured_report(reply) is None
def test_report_points_split_into_a_scannable_label_and_body() -> None:
split = ai_consult_module.split_report_lead
assert split("糖尿病管理缺失: 患者确诊糖尿病3年,空腹血糖为 8.5 mmol/L。") == (
"糖尿病管理缺失",
"患者确诊糖尿病3年,空腹血糖为 8.5 mmol/L。",
)
assert split("结论:由于患者当前未使用任何药物,无需复核。")[0] == "结论"
# 冒号前是一整句话,或正文太短,都按普通要点整段显示。
assert split("尽管无需复核用药,但基于患者病史,以下临床风险点需重点关注。以下为要点:细节")[0] == ""
assert split("空腹血糖: 8.5") == ("", "空腹血糖: 8.5")
assert split("没有冒号的一条要点") == ("", "没有冒号的一条要点")
def test_report_section_titles_drop_the_number_the_chip_already_shows() -> None:
strip = ai_consult_module._strip_leading_ordinal
assert strip("1. 当前用药状态评估") == "当前用药状态评估"
assert strip("二、临床风险与干预提示") == "临床风险与干预提示"
assert strip("建议下一步行动") == "建议下一步行动"
def test_report_body_escapes_markup_and_carries_reading_rhythm() -> None:
html = ai_consult_module._reading_html('血糖 <7.0 mmol/L 且 "达标" & 稳定')
assert "line-height" in html
assert "&lt;7.0" in html
assert "&amp;" in html
assert "<7.0" not in html
def test_structured_report_uses_a_readable_column_width(application: QApplication) -> None:
reply = (
"针对该患者的用药复核评估如下:\n\n"
"### 1. 当前用药状态评估\n"
"- 无当前处方药物: 病例数据中明确记录患者目前没有服药,系统内也没有有效处方记录。\n"
"- 结论: 患者当前未使用任何药物,不存在药物相互作用或配伍禁忌风险。\n"
"### 2. 临床风险与干预提示\n"
"- 糖尿病管理缺失: 患者确诊糖尿病3年,空腹血糖高于一般控制目标且未接受药物治疗。\n"
"重要提示: 本分析不能替代执业医师的面诊与完整病历评估。\n"
)
bubble = ai_consult_module._ChatBubble(role="ai", text=reply, time_text="千问 · 16:32")
assert bubble.finalize_clinical_analysis() is True
panel = bubble.findChild(ai_consult_module._StructuredReportPanel)
assert panel is not None
# 报告收窄到易读行宽,而不是继续用多栏面板的 1080。
assert panel.PREFERRED_MAX_WIDTH == 880
assert bubble._bubble_frame.maximumWidth() == 880
sections = panel.findChildren(ai_consult_module.QFrame, "AiConsultReportSection")
assert len(sections) == 2
assert panel.findChildren(ai_consult_module.QFrame, "AiConsultGapRow") == []
bubble.deleteLater()
def _fitted_bubble(reply: str, width: int = 900) -> Any:
host = QDialog()
host.setObjectName("AiConsultDialog")
host.setStyleSheet(ai_consult_module.AI_CONSULT_QSS)
layout = QVBoxLayout(host)
layout.setContentsMargins(12, 12, 12, 12)
bubble = ai_consult_module._ChatBubble(role="ai", text=reply, time_text="千问 · 16:32")
assert bubble.finalize_clinical_analysis() is True
layout.addWidget(bubble)
layout.addStretch(1)
host.setFixedWidth(width)
host.show()
for _ in range(12):
QApplication.processEvents()
host.adjustSize()
for _ in range(6):
QApplication.processEvents()
return host, bubble
@pytest.mark.parametrize(
"reply",
[
LONG_CLINICAL_REPLY,
(
"针对该患者的用药复核评估如下:\n\n"
"### 1. 当前用药状态评估\n"
"- 无当前处方药物: 病例数据中明确记录患者目前没有服药,系统内也没有有效处方记录,"
"因此不存在药物相互作用或配伍禁忌风险,无需再做安全性复核。\n"
"### 2. 临床风险与干预提示\n"
"- 糖尿病管理缺失: 患者确诊糖尿病3年,空腹血糖高于一般控制目标且未接受任何药物治疗,"
"存在长期高血糖导致微血管及大血管并发症的风险,需要尽快评估。\n"
),
],
ids=["clinical", "structured"],
)
def test_report_bubbles_report_the_height_they_actually_paint(
reply: str,
application: QApplication,
) -> None:
"""否则聊天区会按高估的高度撑出滚动空白,打开就是一片空白要往上滑。"""
host, bubble = _fitted_bubble(reply)
assert bubble.height() > 0
assert abs(bubble.sizeHint().height() - bubble.height()) <= 2
host.close()
host.deleteLater()
def test_risk_block_uses_the_red_alert_palette() -> None:
qss = ai_consult_module.AI_CONSULT_QSS
risk_card = qss.split("QFrame#AiConsultRiskCard {", 1)[1].split("}", 1)[0]
assert "#FEF3F2" in risk_card
assert "#F1B35C" not in risk_card # 旧的橙色描边
marker = qss.split("QLabel#AiConsultRiskMarker {", 1)[1].split("}", 1)[0]
assert "#C0392B" in marker
def test_clinical_bodies_are_no_longer_rendered_at_eleven_pixels() -> None:
qss = ai_consult_module.AI_CONSULT_QSS
body = qss.split("QLabel#AiConsultRiskBody {\n color: #46557A;", 1)
assert len(body) == 2 or "font-size: 13px" in qss
block = qss.split("QLabel#AiConsultClinicalBody,", 1)[1].split("}", 1)[0]
assert "font-size: 13px" in block
assert "font-size: 11px" not in block
+217 -22
View File
@@ -29,6 +29,19 @@ def application() -> QApplication:
return QApplication.instance() or QApplication([])
@pytest.fixture(autouse=True)
def offline_thumbnails(monkeypatch: pytest.MonkeyPatch) -> None:
"""离线传输:缩略图停在 loading 状态,工作站用例永不真正联网。"""
def hold(self: Any, source: str) -> None:
self._source = str(source).strip()
self._invalidate_request()
self.setToolTip(self._source)
self._show_loading()
monkeypatch.setattr(ai_consult_module._RemoteImageButton, "load_url", hold)
@pytest.fixture
def immediate_async(monkeypatch: pytest.MonkeyPatch) -> None:
def run_immediately(
@@ -468,17 +481,33 @@ def test_patient_report_response_owner_must_match_exactly(
dialog.close()
def test_exam_tab_filters_foreign_attachments_and_blocks_file_urls(
def test_exam_tab_previews_images_in_app_and_blocks_file_urls(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
opened: list[str] = []
previews: list[tuple[tuple[str, ...], int]] = []
monkeypatch.setattr(
ai_consult_module,
"open_safe_http_url",
lambda target: opened.append(target) or True,
)
class RecordingPreviewDialog(QWidget):
def __init__(
self,
sources: Any,
*,
index: int = 0,
names: Any = None,
title: str = "",
parent: QWidget | None = None,
) -> None:
super().__init__(parent)
previews.append((tuple(sources), int(index)))
monkeypatch.setattr(ai_consult_module, "ImagePreviewDialog", RecordingPreviewDialog)
dialog = _open_dialog(application, WorkspaceRepository())
pane = dialog.records["检查检验"]
dialog.tabs.setCurrentIndex(2)
@@ -498,20 +527,187 @@ def test_exam_tab_filters_foreign_attachments_and_blocks_file_urls(
assert len(thumbnails) == 1
assert thumbnails[0].isEnabled()
assert thumbnails[0].accessibleName() == "舌苔图片点击查看"
assert thumbnails[0].property("loadState") == "blocked"
assert thumbnails[0].property("loadState") == "loading"
assert pane.state_label.property("state") == "warning" # type: ignore[attr-defined]
assert pane.retry_button.isHidden() # type: ignore[attr-defined]
image_button = next(button for button in buttons if "甲舌苔照片.jpg" in button.text())
assert image_button.text().endswith("· 预览")
report_button = next(button for button in buttons if "甲血糖报告.pdf" in button.text())
assert report_button.text().endswith("· 打开")
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)
thumbnails[0].click()
# 图片留在工作站内预览,只有非图片附件才交给系统打开。
assert opened == ["https://media.example.invalid/甲/report.pdf"]
assert previews == [(("https://media.example.invalid/甲/tongue.jpg",), 0)] * 2
dialog.close()
def test_tongue_thumbnail_auto_get_requires_configured_https_origin(
class ProductionShapeRepository(WorkspaceRepository):
"""按线上 readonlyDetail 的真实返回构造:既有 code,也有后端补的 *_text。"""
def get_diagnosis_detail(
self,
diagnosis_id: int,
*,
readonly: bool = False,
) -> dict[str, Any]:
detail = dict(super().get_diagnosis_detail(diagnosis_id, readonly=readonly))
diagnosis = dict(detail["diagnosis"])
diagnosis.update(
{
"gender": 1,
"gender_text": "",
"diagnosis_type": "follow_up",
"diagnosis_type_text": "复诊",
"eye_condition": "blurred,dry",
"eye_condition_text": "模糊、干涩",
"skin_condition": "dry,itching",
"skin_condition_text": "干燥、瘙痒",
"urine_condition": "yellow_urine",
"urine_condition_text": "尿黄",
"fatty_liver_degree": "mild",
"fatty_liver_degree_text": "轻度",
"past_history": "hypertension",
"past_history_text": "高血压",
"trauma_history": 0,
"trauma_history_text": "",
# 诊单表里的技术列:医生页面不应出现这些英文列名。
"status": 1,
"show_card": 1,
"revisit_slot_start_offset": 0,
"delete_time": None,
"assistant_id": 131,
"assign_read_at": 1787882294,
"shipped_non_er_assistant_cleared_at": 0,
"external_userid": "",
"is_view": 0,
"create_time": 1787882294,
"update_time": 1787882303,
"tongue_images": [
"https://media.example.invalid/11702/a.jpg",
"https://media.example.invalid/11702/b.jpg",
],
}
)
detail["diagnosis"] = diagnosis
return detail
def test_case_tab_hides_raw_columns_and_never_shows_text_mirror_fields(
application: QApplication,
immediate_async: None,
) -> None:
dialog = _open_dialog(application, ProductionShapeRepository())
dialog.tabs.setCurrentIndex(1)
application.processEvents()
text = _pane_text(dialog.records["病历资料"])
# 后端补的 *_text 只用来取值,不再作为独立英文字段列出来。
for mirror in (
"diagnosis type text",
"eye condition text",
"gender text",
"past history text",
):
assert mirror not in text
# 技术列不再泄漏英文列名。
for internal in (
"assign read at",
"shipped non er assistant cleared at",
"external userid",
"is view",
"assistant id",
"show card",
"revisit slot start offset",
"delete time",
):
assert internal not in text
# 有中文名的字段照常显示,取的是后端翻译过的值。
assert "复诊" in text
assert "模糊、干涩" in text
assert "1787882294" not in text
assert "https://media.example.invalid/11702/a.jpg" not in text
assert len(dialog.records["病历资料"].findChildren(QPushButton, "AiConsultTongueThumb")) == 2
dialog.close()
class RawCodeRepository(WorkspaceRepository):
"""只读接口偶尔缺少 `*_text`(历史数据 / 快照),此时必须自己翻译字典 code。"""
def get_diagnosis_detail(
self,
diagnosis_id: int,
*,
readonly: bool = False,
) -> dict[str, Any]:
detail = dict(super().get_diagnosis_detail(diagnosis_id, readonly=readonly))
diagnosis = dict(detail["diagnosis"])
diagnosis.update(
{
"gender": 1,
"diagnosis_type": "follow_up",
"appetite": "dry,bitter",
"weight_change": "lose_10_jin",
"fatty_liver_degree": "mild",
"allergy_history": 0,
"status": 1,
"show_card": 1,
"revisit_slot_start_offset": 0,
"delete_time": None,
"create_source": "admin",
"create_time": 1783838927,
"tongue_images": ["https://media.example.invalid/501/tongue-raw.jpg"],
}
)
detail["diagnosis"] = diagnosis
patient = dict(detail.get("patient") or {})
patient.update({"gender": 1, "marital_status": 1})
detail["patient"] = patient
return detail
def test_case_and_health_tabs_translate_codes_and_hide_internal_columns(
application: QApplication,
immediate_async: None,
) -> None:
dialog = _open_dialog(application, RawCodeRepository())
dialog.tabs.setCurrentIndex(1)
application.processEvents()
case_text = _pane_text(dialog.records["病历资料"])
# 字典 code、性别与是否类枚举、时间戳都按后台只读页的口径显示。
assert "干、苦" in case_text
assert "瘦10斤" in case_text
assert "轻度" in case_text
assert "复诊" in case_text
assert "后台创建" in case_text
assert "dry,bitter" not in case_text
assert "lose_10_jin" not in case_text
assert "follow_up" not in case_text
assert "1783838927" not in case_text
# 纯内部列不再泄漏给医生。
for internal in ("show card", "revisit slot start offset", "delete time"):
assert internal not in case_text
# 舌象附件渲染成缩略图,不再是一长串 URL 文本。
assert "https://media.example.invalid/501/tongue-raw.jpg" not in case_text
thumbnails = dialog.records["病历资料"].findChildren(QPushButton, "AiConsultTongueThumb")
assert len(thumbnails) == 1
dialog.tabs.setCurrentIndex(4)
application.processEvents()
health_text = _pane_text(dialog.records["健康档案"])
assert "性别\n" in health_text
assert "过敏史\n" in health_text
assert "162 cm" in health_text
assert dialog.records["健康档案"].findChildren(QPushButton, "AiConsultTongueThumb")
dialog.close()
def test_tongue_thumbnails_load_safe_http_sources_and_skip_unsafe_schemes(
application: QApplication,
immediate_async: None,
monkeypatch: pytest.MonkeyPatch,
@@ -533,23 +729,22 @@ def test_tongue_thumbnail_auto_get_requires_configured_https_origin(
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)
dialog = _open_dialog(application, WorkspaceRepository())
# 舌象照片按字段直接渲染,PDF 报告与 file:// 附件不会发起任何图片请求。
assert requested == ["https://media.example.invalid/甲/tongue.jpg"]
trusted.close()
assert not ai_consult_module._previewable_attachment(
"tongue_images",
"file:///C:/private/tongue.jpg",
)
assert not ai_consult_module._previewable_attachment(
"report_files",
"https://media.example.invalid/甲/report.pdf",
)
assert ai_consult_module._previewable_attachment(
"report_files",
"https://media.example.invalid/甲/report.PNG",
)
dialog.close()
def test_three_prescription_cards_open_exact_details_and_reject_wrong_or_late_ids(
+536 -357
View File
@@ -1,94 +1,124 @@
"""Desktop auto-update check, download and payload discovery."""
from __future__ import annotations
import hashlib
import zipfile
from pathlib import Path
import httpx
import pytest
from doctor_workstation.services import app_update
from doctor_workstation.services.api_client import ApiClient
from doctor_workstation.services.app_update import (
PACKAGE_TYPE_ARCHIVE,
PACKAGE_TYPE_INNO_SETUP,
AppUpdateError,
UpdatePackage,
apply_extracted_update,
apply_inno_setup_update,
compare_version,
discover_payload,
download_package,
fetch_update_offer,
normalize_version,
package_filename,
parse_update_offer,
safe_extract_zip,
validate_installer_download_policy,
validate_windows_installer,
)
def test_normalize_and_compare_versions() -> None:
assert normalize_version("0.2") == "0.2.0"
assert normalize_version("1.2.3.4") == "1.2.3"
assert normalize_version("nope") == ""
assert compare_version("0.1.0", "0.2.0") < 0
assert compare_version("0.2.0", "0.2.0") == 0
assert compare_version("1.0.0", "0.9.9") > 0
def test_parse_offer_requires_hash_before_install() -> None:
offer = parse_update_offer(
{
"has_update": True,
"force": True,
"enabled": True,
"latest_version": "0.2.0",
"package": {
"url": "https://cdn.example.com/app.zip",
"sha256": "",
"size": 12,
"filename": "app.zip",
},
"can_install": True,
},
current_version="0.1.0",
)
assert offer.has_update is True
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
"""Desktop auto-update check, download and payload discovery."""
from __future__ import annotations
import hashlib
import zipfile
from pathlib import Path
import httpx
import pytest
from doctor_workstation.services import app_update
from doctor_workstation.services.api_client import ApiClient
from doctor_workstation.services.app_update import (
PACKAGE_TYPE_ARCHIVE,
PACKAGE_TYPE_INNO_SETUP,
AppUpdateError,
UpdatePackage,
apply_extracted_update,
apply_inno_setup_update,
compare_version,
discover_payload,
download_package,
fetch_update_offer,
normalize_version,
package_filename,
parse_update_offer,
safe_extract_zip,
validate_installer_download_policy,
validate_windows_installer,
)
def test_normalize_and_compare_versions() -> None:
assert normalize_version("0.2") == "0.2.0"
assert normalize_version("1.2.3.4") == "1.2.3"
assert normalize_version("nope") == ""
assert compare_version("0.1.0", "0.2.0") < 0
assert compare_version("0.2.0", "0.2.0") == 0
assert compare_version("1.0.0", "0.9.9") > 0
def test_parse_offer_requires_hash_before_install() -> None:
offer = parse_update_offer(
{
"has_update": True,
"force": True,
"enabled": True,
"latest_version": "0.2.0",
"package": {
"url": "https://cdn.example.com/app.zip",
"sha256": "",
"size": 12,
"filename": "app.zip",
},
"can_install": True,
},
current_version="0.1.0",
)
assert offer.has_update is True
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
def test_parse_offer_accepts_explicit_inno_setup_type() -> None:
offer = parse_update_offer(
{
"has_update": True,
"enabled": True,
"latest_version": "0.2.0",
"platform": "windows",
"arch": "x64",
"package": {
"url": "https://cdn.example.com/DoctorWorkstation-Setup.exe",
"sha256": "a" * 64,
"size": 123,
"filename": "DoctorWorkstation-Setup.exe",
"type": "inno_setup",
},
"can_install": True,
},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.can_install is True
assert offer.package is not None
offer = parse_update_offer(
{
"has_update": True,
"enabled": True,
"latest_version": "0.2.0",
"platform": "windows",
"arch": "x64",
"package": {
"url": "https://cdn.example.com/DoctorWorkstation-Setup.exe",
"sha256": "a" * 64,
"size": 123,
"filename": "DoctorWorkstation-Setup.exe",
"type": "inno_setup",
},
"can_install": True,
},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.can_install is True
assert offer.package is not None
assert offer.package.type == PACKAGE_TYPE_INNO_SETUP
def test_parse_offer_rejects_package_version_mismatch() -> None:
offer = parse_update_offer(
{
"enabled": True,
"has_update": True,
"force": True,
"can_install": True,
"latest_version": "1.3.0",
"platform": "windows",
"arch": "x64",
"package": {
"type": PACKAGE_TYPE_INNO_SETUP,
"url": "https://cdn.example.com/DoctorWorkstation-Setup-1.1.0.exe",
"filename": "DoctorWorkstation-Setup-1.1.0.exe",
"sha256": "a" * 64,
"size": 1024,
},
},
current_version="1.2.0",
platform_name="windows",
arch="x64",
)
assert offer.has_update is True
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
assert "安装包版本 1.1.0 与发布版本 1.3.0 不一致" in offer.install_unavailable_reason
def test_parse_offer_disables_insecure_inno_setup_transport() -> None:
offer = parse_update_offer(
{
@@ -113,235 +143,236 @@ def test_parse_offer_disables_insecure_inno_setup_transport() -> None:
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
@pytest.mark.parametrize("package_type", ["msi", "script", "unknown"])
def test_parse_offer_rejects_unknown_package_type(package_type: str) -> None:
offer = parse_update_offer(
{
"has_update": True,
"latest_version": "0.2.0",
"platform": "windows",
"arch": "x64",
"package": {
"url": "https://cdn.example.com/update.bin",
"sha256": "a" * 64,
"type": package_type,
},
"can_install": True,
},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
def test_parse_offer_rejects_stale_or_wrong_platform_response() -> None:
base = {
"has_update": True,
"latest_version": "0.1.0",
"platform": "windows",
"arch": "x64",
"can_install": False,
}
stale = parse_update_offer(
base,
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
wrong_platform = parse_update_offer(
{**base, "latest_version": "0.2.0", "platform": "macos"},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert stale.has_update is False
assert wrong_platform.has_update is False
def test_fetch_update_offer_uses_check_endpoint() -> None:
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(
200,
json={
"code": 1,
"data": {
"has_update": True,
"force": True,
"enabled": True,
"latest_version": "0.2.0",
"title": "医生工作站 0.2.0",
"notes": "修复登录",
"package": {
"url": "https://cdn.example.com/DoctorWorkstation.zip",
"sha256": "a" * 64,
"size": 2048,
"filename": "DoctorWorkstation.zip",
},
"can_install": True,
},
},
)
with ApiClient("https://example.test", transport=httpx.MockTransport(handler)) as client:
offer = fetch_update_offer(
client,
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.has_update is True
assert offer.force is True
assert offer.can_install is True
assert offer.package is not None
assert "setting.desktop_workstation/check" in str(requests[0].url)
assert "current_version=0.1.0" in str(requests[0].url)
assert "platform=windows" in str(requests[0].url)
def test_safe_extract_rejects_zip_slip(tmp_path: Path) -> None:
archive = tmp_path / "evil.zip"
with zipfile.ZipFile(archive, "w") as bundle:
bundle.writestr("../outside.txt", "nope")
with pytest.raises(AppUpdateError, match="非法路径"):
safe_extract_zip(archive, tmp_path / "out")
def test_discover_windows_payload_prefers_internal_onedir(tmp_path: Path) -> None:
wrapped = tmp_path / "DoctorWorkstation"
wrapped.mkdir()
(wrapped / "_internal").mkdir()
(wrapped / "DoctorWorkstation.exe").write_bytes(b"mz")
(tmp_path / "Start_DoctorWorkstation.bat").write_text("start", encoding="utf-8")
assert discover_payload(tmp_path, platform_name="windows") == wrapped
def test_discover_macos_payload_finds_app_bundle(tmp_path: Path) -> None:
app = tmp_path / "DoctorWorkstation.app"
macos = app / "Contents" / "MacOS"
macos.mkdir(parents=True)
(macos / "DoctorWorkstation").write_text("bin", encoding="utf-8")
assert discover_payload(tmp_path, platform_name="macos") == app
def test_download_package_verifies_sha256_and_reports_progress(tmp_path: Path) -> None:
payload = b"doctor-workstation-zip"
digest = hashlib.sha256(payload).hexdigest()
progress: list[tuple[int, int]] = []
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(
200,
content=payload,
headers={"content-length": str(len(payload))},
)
destination = tmp_path / "pkg.zip"
download_package(
"https://cdn.example.com/pkg.zip",
destination,
sha256=digest,
progress=lambda received, total: progress.append((received, total)),
transport=httpx.MockTransport(handler),
)
assert destination.read_bytes() == payload
assert progress[-1][0] == len(payload)
def test_download_package_rejects_hash_mismatch(tmp_path: Path) -> None:
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(200, content=b"tampered")
destination = tmp_path / "pkg.zip"
with pytest.raises(AppUpdateError, match="校验失败"):
download_package(
"https://cdn.example.com/pkg.zip",
destination,
sha256="b" * 64,
transport=httpx.MockTransport(handler),
)
assert not destination.exists()
def test_download_package_rejects_declared_size_mismatch(tmp_path: Path) -> None:
payload = b"short"
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(200, content=payload)
destination = tmp_path / "pkg.exe"
with pytest.raises(AppUpdateError, match="文件大小"):
download_package(
"https://cdn.example.com/pkg.exe",
destination,
sha256=hashlib.sha256(payload).hexdigest(),
expected_size=len(payload) + 1,
transport=httpx.MockTransport(handler),
)
assert not destination.exists()
assert not (tmp_path / "pkg.exe.part").exists()
def test_windows_installer_download_policy_requires_verified_https() -> None:
with pytest.raises(AppUpdateError, match="HTTPS"):
validate_installer_download_policy(
"http://cdn.example.com/setup.exe",
verify_ssl=True,
)
with pytest.raises(AppUpdateError, match="证书校验"):
validate_installer_download_policy(
"https://cdn.example.com/setup.exe",
verify_ssl=False,
)
validate_installer_download_policy(
"http://127.0.0.1/setup.exe",
verify_ssl=True,
)
def test_validate_windows_installer_requires_exe_and_pe_header(tmp_path: Path) -> None:
installer = tmp_path / "Setup.exe"
installer.write_bytes(b"MZ" + b"\0" * 32)
assert validate_windows_installer(installer) == installer.resolve()
invalid = tmp_path / "invalid.exe"
invalid.write_bytes(b"PK")
with pytest.raises(AppUpdateError, match="PE"):
validate_windows_installer(invalid)
def test_inno_setup_applier_waits_installs_and_restarts_installed_exe(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
install_root = tmp_path / "installed"
install_root.mkdir()
installed_exe = install_root / "DoctorWorkstation.exe"
installed_exe.write_bytes(b"MZ")
installer = tmp_path / "DoctorWorkstation-Setup.exe"
installer.write_bytes(b"MZ" + b"\0" * 32)
spawned: dict[str, Path] = {}
monkeypatch.setattr(app_update.sys, "platform", "win32")
def capture_spawn(
script: Path,
@pytest.mark.parametrize("package_type", ["msi", "script", "unknown"])
def test_parse_offer_rejects_unknown_package_type(package_type: str) -> None:
offer = parse_update_offer(
{
"has_update": True,
"latest_version": "0.2.0",
"platform": "windows",
"arch": "x64",
"package": {
"url": "https://cdn.example.com/update.bin",
"sha256": "a" * 64,
"type": package_type,
},
"can_install": True,
},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.can_install is False
assert offer.force is False
assert offer.package is None
def test_parse_offer_rejects_stale_or_wrong_platform_response() -> None:
base = {
"has_update": True,
"latest_version": "0.1.0",
"platform": "windows",
"arch": "x64",
"can_install": False,
}
stale = parse_update_offer(
base,
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
wrong_platform = parse_update_offer(
{**base, "latest_version": "0.2.0", "platform": "macos"},
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert stale.has_update is False
assert wrong_platform.has_update is False
def test_fetch_update_offer_uses_check_endpoint() -> None:
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(
200,
json={
"code": 1,
"data": {
"has_update": True,
"force": True,
"enabled": True,
"latest_version": "0.2.0",
"title": "医生工作站 0.2.0",
"notes": "修复登录",
"package": {
"url": "https://cdn.example.com/DoctorWorkstation.zip",
"sha256": "a" * 64,
"size": 2048,
"filename": "DoctorWorkstation.zip",
},
"can_install": True,
},
},
)
with ApiClient("https://example.test", transport=httpx.MockTransport(handler)) as client:
offer = fetch_update_offer(
client,
current_version="0.1.0",
platform_name="windows",
arch="x64",
)
assert offer.has_update is True
assert offer.force is True
assert offer.can_install is True
assert offer.package is not None
assert "setting.desktop_workstation/check" in str(requests[0].url)
assert "current_version=0.1.0" in str(requests[0].url)
assert "platform=windows" in str(requests[0].url)
def test_safe_extract_rejects_zip_slip(tmp_path: Path) -> None:
archive = tmp_path / "evil.zip"
with zipfile.ZipFile(archive, "w") as bundle:
bundle.writestr("../outside.txt", "nope")
with pytest.raises(AppUpdateError, match="非法路径"):
safe_extract_zip(archive, tmp_path / "out")
def test_discover_windows_payload_prefers_internal_onedir(tmp_path: Path) -> None:
wrapped = tmp_path / "DoctorWorkstation"
wrapped.mkdir()
(wrapped / "_internal").mkdir()
(wrapped / "DoctorWorkstation.exe").write_bytes(b"mz")
(tmp_path / "Start_DoctorWorkstation.bat").write_text("start", encoding="utf-8")
assert discover_payload(tmp_path, platform_name="windows") == wrapped
def test_discover_macos_payload_finds_app_bundle(tmp_path: Path) -> None:
app = tmp_path / "DoctorWorkstation.app"
macos = app / "Contents" / "MacOS"
macos.mkdir(parents=True)
(macos / "DoctorWorkstation").write_text("bin", encoding="utf-8")
assert discover_payload(tmp_path, platform_name="macos") == app
def test_download_package_verifies_sha256_and_reports_progress(tmp_path: Path) -> None:
payload = b"doctor-workstation-zip"
digest = hashlib.sha256(payload).hexdigest()
progress: list[tuple[int, int]] = []
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(
200,
content=payload,
headers={"content-length": str(len(payload))},
)
destination = tmp_path / "pkg.zip"
download_package(
"https://cdn.example.com/pkg.zip",
destination,
sha256=digest,
progress=lambda received, total: progress.append((received, total)),
transport=httpx.MockTransport(handler),
)
assert destination.read_bytes() == payload
assert progress[-1][0] == len(payload)
def test_download_package_rejects_hash_mismatch(tmp_path: Path) -> None:
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(200, content=b"tampered")
destination = tmp_path / "pkg.zip"
with pytest.raises(AppUpdateError, match="校验失败"):
download_package(
"https://cdn.example.com/pkg.zip",
destination,
sha256="b" * 64,
transport=httpx.MockTransport(handler),
)
assert not destination.exists()
def test_download_package_rejects_declared_size_mismatch(tmp_path: Path) -> None:
payload = b"short"
def handler(request: httpx.Request) -> httpx.Response:
del request
return httpx.Response(200, content=payload)
destination = tmp_path / "pkg.exe"
with pytest.raises(AppUpdateError, match="文件大小"):
download_package(
"https://cdn.example.com/pkg.exe",
destination,
sha256=hashlib.sha256(payload).hexdigest(),
expected_size=len(payload) + 1,
transport=httpx.MockTransport(handler),
)
assert not destination.exists()
assert not (tmp_path / "pkg.exe.part").exists()
def test_windows_installer_download_policy_requires_verified_https() -> None:
with pytest.raises(AppUpdateError, match="HTTPS"):
validate_installer_download_policy(
"http://cdn.example.com/setup.exe",
verify_ssl=True,
)
with pytest.raises(AppUpdateError, match="证书校验"):
validate_installer_download_policy(
"https://cdn.example.com/setup.exe",
verify_ssl=False,
)
validate_installer_download_policy(
"http://127.0.0.1/setup.exe",
verify_ssl=True,
)
def test_validate_windows_installer_requires_exe_and_pe_header(tmp_path: Path) -> None:
installer = tmp_path / "Setup.exe"
installer.write_bytes(b"MZ" + b"\0" * 32)
assert validate_windows_installer(installer) == installer.resolve()
invalid = tmp_path / "invalid.exe"
invalid.write_bytes(b"PK")
with pytest.raises(AppUpdateError, match="PE"):
validate_windows_installer(invalid)
def test_inno_setup_applier_waits_installs_and_restarts_installed_exe(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
install_root = tmp_path / "installed"
install_root.mkdir()
installed_exe = install_root / "DoctorWorkstation.exe"
installed_exe.write_bytes(b"MZ")
installer = tmp_path / "DoctorWorkstation-Setup.exe"
installer.write_bytes(b"MZ" + b"\0" * 32)
spawned: dict[str, Path] = {}
monkeypatch.setattr(app_update.sys, "platform", "win32")
def capture_spawn(
script: Path,
*,
installer: Path,
restart_exe: Path,
helper_log_file: Path,
installer_log_file: Path,
ready_file: Path,
) -> None:
spawned.update(
script=script,
@@ -349,44 +380,192 @@ def test_inno_setup_applier_waits_installs_and_restarts_installed_exe(
restart_exe=restart_exe,
helper_log_file=helper_log_file,
installer_log_file=installer_log_file,
ready_file=ready_file,
)
monkeypatch.setattr(app_update, "_spawn_inno_setup_applier", capture_spawn)
apply_inno_setup_update(installer, install_root=install_root)
script_text = spawned["script"].read_text(encoding="utf-8-sig")
assert spawned["installer"] == installer.resolve()
assert spawned["restart_exe"] == installed_exe
assert "/VERYSILENT" in script_text
assert "/RESTARTEXITCODE=3010" in script_text
monkeypatch.setattr(app_update, "_spawn_inno_setup_applier", capture_spawn)
apply_inno_setup_update(installer, install_root=install_root)
script_text = spawned["script"].read_text(encoding="utf-8-sig")
assert spawned["installer"] == installer.resolve()
assert spawned["restart_exe"] == installed_exe
assert "/VERYSILENT" in script_text
assert "/RESTARTEXITCODE=3010" in script_text
assert "/NOFORCECLOSEAPPLICATIONS" in script_text
assert "$HelperLogFile" in script_text
assert "$InstallerLogFile" in script_text
assert "$ReadyFile" in script_text
assert "helper ready" in script_text
assert "Restart-Application" in script_text
def test_inno_helper_uses_runnable_flags_and_waits_for_ready(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
script = tmp_path / "install_update.ps1"
script.write_text("", encoding="utf-8")
installer = tmp_path / "Setup.exe"
restart_exe = tmp_path / "DoctorWorkstation.exe"
helper_log = tmp_path / "helper.log"
installer_log = tmp_path / "inno.log"
ready_file = tmp_path / "helper.ready"
captured: dict[str, object] = {}
class FakeProcess:
def poll(self) -> None:
return None
def fake_popen(args: list[str], **kwargs: object) -> FakeProcess:
captured["args"] = args
captured.update(kwargs)
ready_file.write_text("ready", encoding="utf-8")
return FakeProcess()
monkeypatch.setattr(app_update.subprocess, "DETACHED_PROCESS", 8, raising=False)
monkeypatch.setattr(app_update.subprocess, "CREATE_NEW_PROCESS_GROUP", 512, raising=False)
monkeypatch.setattr(app_update.subprocess, "CREATE_NO_WINDOW", 134217728, raising=False)
monkeypatch.setattr(app_update.subprocess, "Popen", fake_popen)
app_update._spawn_inno_setup_applier(
script,
installer=installer,
restart_exe=restart_exe,
helper_log_file=helper_log,
installer_log_file=installer_log,
ready_file=ready_file,
)
flags = int(captured["creationflags"])
detached = int(getattr(app_update.subprocess, "DETACHED_PROCESS", 0))
assert not detached or flags & detached == 0
assert flags & int(getattr(app_update.subprocess, "CREATE_NEW_PROCESS_GROUP", 0))
assert flags & int(getattr(app_update.subprocess, "CREATE_NO_WINDOW", 0))
assert "-ReadyFile" in captured["args"]
def test_inno_helper_reports_exit_before_ready(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
script = tmp_path / "install_update.ps1"
script.write_text("", encoding="utf-8")
class ExitedProcess:
def poll(self) -> int:
return 23
monkeypatch.setattr(app_update.subprocess, "Popen", lambda *args, **kwargs: ExitedProcess())
with pytest.raises(OSError, match="提前退出(代码 23"):
app_update._spawn_inno_setup_applier(
script,
installer=tmp_path / "Setup.exe",
restart_exe=tmp_path / "DoctorWorkstation.exe",
helper_log_file=tmp_path / "helper.log",
installer_log_file=tmp_path / "inno.log",
ready_file=tmp_path / "helper.ready",
)
@pytest.mark.skipif(app_update.sys.platform != "win32", reason="Windows helper contract")
def test_inno_helper_executes_bootstrap_with_production_flags(tmp_path: Path) -> None:
script = tmp_path / "helper probe.ps1"
script.write_text(
"\n".join(
[
"param(",
"[int]$TargetPid, [string]$Installer, [string]$RestartExe,",
"[string]$HelperLogFile, [string]$InstallerLogFile, [string]$ReadyFile",
")",
'Set-Content -LiteralPath $ReadyFile -Value "ready" -Encoding UTF8',
]
),
encoding="utf-8-sig",
)
ready_file = tmp_path / "helper.ready"
app_update._spawn_inno_setup_applier(
script,
installer=tmp_path / "Setup.exe",
restart_exe=tmp_path / "DoctorWorkstation.exe",
helper_log_file=tmp_path / "helper.log",
installer_log_file=tmp_path / "inno.log",
ready_file=ready_file,
)
assert ready_file.read_text(encoding="utf-8-sig").strip() == "ready"
@pytest.mark.skipif(app_update.sys.platform != "win32", reason="Windows helper contract")
def test_inno_helper_survives_launcher_process_exit(tmp_path: Path) -> None:
script = tmp_path / "helper parent-exit probe.ps1"
script.write_text(
"\n".join(
[
"param(",
"[int]$TargetPid, [string]$Installer, [string]$RestartExe,",
"[string]$HelperLogFile, [string]$InstallerLogFile, [string]$ReadyFile",
")",
'Set-Content -LiteralPath $ReadyFile -Value "ready" -Encoding UTF8',
"while (Get-Process -Id $TargetPid -ErrorAction SilentlyContinue) {",
" Start-Sleep -Milliseconds 50",
"}",
'Set-Content -LiteralPath $HelperLogFile -Value "parent-exited" -Encoding UTF8',
]
),
encoding="utf-8-sig",
)
helper_log = tmp_path / "helper.log"
ready_file = tmp_path / "helper.ready"
launcher = (
"from pathlib import Path; import sys; "
"from doctor_workstation.services.app_update import _spawn_inno_setup_applier; "
"root=Path(sys.argv[1]); "
"_spawn_inno_setup_applier(root/'helper parent-exit probe.ps1', "
"installer=root/'Setup.exe', restart_exe=root/'DoctorWorkstation.exe', "
"helper_log_file=root/'helper.log', installer_log_file=root/'inno.log', "
"ready_file=root/'helper.ready')"
)
launched = app_update.subprocess.run(
[app_update.sys.executable, "-c", launcher, str(tmp_path)],
cwd=str(Path.cwd()),
capture_output=True,
text=True,
timeout=10,
)
assert launched.returncode == 0, launched.stderr
deadline = app_update.time.monotonic() + 5.0
while not helper_log.is_file() and app_update.time.monotonic() < deadline:
app_update.time.sleep(0.05)
assert ready_file.is_file()
assert helper_log.read_text(encoding="utf-8-sig").strip() == "parent-exited"
def test_archive_applier_restarts_from_install_root(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
payload = tmp_path / "payload"
payload.mkdir()
(payload / "DoctorWorkstation.exe").write_bytes(b"MZ")
install_root = tmp_path / "installed"
install_root.mkdir()
installed_exe = install_root / "DoctorWorkstation.exe"
installed_exe.write_bytes(b"MZ")
captured: dict[str, Path] = {}
script = tmp_path / "apply.ps1"
script.write_text("", encoding="utf-8")
def capture_script(**kwargs: Path) -> Path:
captured.update(kwargs)
return script
monkeypatch.setattr(app_update, "_write_apply_script", capture_script)
monkeypatch.setattr(app_update, "_spawn_applier", lambda *args, **kwargs: None)
apply_extracted_update(payload, install_root=install_root)
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
payload = tmp_path / "payload"
payload.mkdir()
(payload / "DoctorWorkstation.exe").write_bytes(b"MZ")
install_root = tmp_path / "installed"
install_root.mkdir()
installed_exe = install_root / "DoctorWorkstation.exe"
installed_exe.write_bytes(b"MZ")
captured: dict[str, Path] = {}
script = tmp_path / "apply.ps1"
script.write_text("", encoding="utf-8")
def capture_script(**kwargs: Path) -> Path:
captured.update(kwargs)
return script
monkeypatch.setattr(app_update, "_write_apply_script", capture_script)
monkeypatch.setattr(app_update, "_spawn_applier", lambda *args, **kwargs: None)
apply_extracted_update(payload, install_root=install_root)
assert captured["restart_exe"] == installed_exe
@@ -409,17 +588,17 @@ def test_inno_setup_applier_reports_helper_start_failure(
with pytest.raises(AppUpdateError, match="无法启动 Windows 更新助手"):
apply_inno_setup_update(installer, install_root=install_root)
def test_package_filename_defaults_match_package_type() -> None:
archive = UpdatePackage("https://cdn.example.com/", "a" * 64, 0, "")
installer = UpdatePackage(
"https://cdn.example.com/",
"a" * 64,
0,
"",
type=PACKAGE_TYPE_INNO_SETUP,
)
assert package_filename(archive, "0.2.0").endswith(".zip")
assert package_filename(installer, "0.2.0").endswith(".exe")
assert archive.type == PACKAGE_TYPE_ARCHIVE
def test_package_filename_defaults_match_package_type() -> None:
archive = UpdatePackage("https://cdn.example.com/", "a" * 64, 0, "")
installer = UpdatePackage(
"https://cdn.example.com/",
"a" * 64,
0,
"",
type=PACKAGE_TYPE_INNO_SETUP,
)
assert package_filename(archive, "0.2.0").endswith(".zip")
assert package_filename(installer, "0.2.0").endswith(".exe")
assert archive.type == PACKAGE_TYPE_ARCHIVE
+250 -70
View File
@@ -1,70 +1,250 @@
"""Update dialog contract for optional and forced desktop upgrades."""
from __future__ import annotations
import os
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtWidgets import QApplication
from doctor_workstation.services.app_update import UpdateOffer, UpdatePackage
from doctor_workstation.ui.dialogs.app_update import AppUpdateDialog
from doctor_workstation.ui.theme import apply_theme
def _offer(*, force: bool, can_install: bool = True) -> UpdateOffer:
package = (
UpdatePackage(
url="https://cdn.example.com/DoctorWorkstation.zip",
sha256="a" * 64,
size=1024,
filename="DoctorWorkstation.zip",
)
if can_install
else None
)
return UpdateOffer(
has_update=True,
force=force,
enabled=True,
current_version="0.1.0",
latest_version="0.2.0",
min_version="",
title="医生工作站 0.2.0",
notes="修复若干问题",
platform="windows",
arch="x64",
package=package,
can_install=can_install,
)
def test_optional_update_dialog_allows_later(application: QApplication | None = None) -> None:
app = application or QApplication.instance() or QApplication([])
apply_theme(app)
dialog = AppUpdateDialog(_offer(force=False))
dialog.show()
app.processEvents()
assert dialog.later_button.isVisible()
assert dialog.update_button.text() == "立即更新"
assert dialog.notes.toPlainText() == "修复若干问题"
dialog.close()
def test_forced_update_dialog_hides_defer_and_blocks_escape(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
apply_theme(app)
dialog = AppUpdateDialog(_offer(force=True))
dialog.show()
app.processEvents()
assert not dialog.later_button.isVisible()
assert "必须更新" in dialog.badge.text()
dialog.close()
app.processEvents()
assert dialog.isVisible()
dialog.offer = _offer(force=False)
dialog._busy = False
dialog.close()
"""Update dialog contract for optional and forced desktop upgrades."""
from __future__ import annotations
import os
from dataclasses import replace
from types import SimpleNamespace
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
from PySide6.QtCore import QObject, Qt
from PySide6.QtTest import QTest
from PySide6.QtWidgets import QApplication
from doctor_workstation.services.app_update import (
PACKAGE_TYPE_INNO_SETUP,
UpdateOffer,
UpdatePackage,
)
from doctor_workstation.ui.dialogs.app_update import AppUpdateDialog, AppUpdateSession
from doctor_workstation.ui.theme import apply_theme
def _offer(
*,
force: bool,
can_install: bool = True,
install_unavailable_reason: str = "",
) -> UpdateOffer:
package = (
UpdatePackage(
url="https://cdn.example.com/DoctorWorkstation.zip",
sha256="a" * 64,
size=1024,
filename="DoctorWorkstation.zip",
)
if can_install
else None
)
return UpdateOffer(
has_update=True,
force=force,
enabled=True,
current_version="0.1.0",
latest_version="0.2.0",
min_version="",
title="医生工作站 0.2.0",
notes="修复若干问题",
platform="windows",
arch="x64",
package=package,
can_install=can_install,
install_unavailable_reason=install_unavailable_reason,
)
def test_optional_update_dialog_allows_later(application: QApplication | None = None) -> None:
app = application or QApplication.instance() or QApplication([])
apply_theme(app)
dialog = AppUpdateDialog(_offer(force=False))
dialog.show()
app.processEvents()
deferred: list[bool] = []
dialog.update_deferred.connect(lambda: deferred.append(True))
assert dialog.later_button.isVisible()
assert dialog.later_button.isEnabled()
assert dialog.later_button.text() == "稍后提醒"
assert not dialog.exit_button.isVisible()
assert dialog.update_button.text() == "立即更新"
assert dialog.notes.toPlainText() == "修复若干问题"
dialog.later_button.click()
assert deferred == [True]
assert not dialog.isVisible()
dialog.deleteLater()
app.processEvents()
def test_forced_update_dialog_has_explicit_exit_and_blocks_implicit_close(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
apply_theme(app)
dialog = AppUpdateDialog(_offer(force=True))
dialog.show()
app.processEvents()
assert not dialog.later_button.isVisible()
assert dialog.exit_button.isVisible()
assert dialog.exit_button.isEnabled()
assert dialog.exit_button.text() == "退出软件"
assert "必须更新" in dialog.badge.text()
dialog.close()
app.processEvents()
assert dialog.isVisible()
QTest.keyClick(dialog, Qt.Key.Key_Escape)
app.processEvents()
assert dialog.isVisible()
dialog.set_busy(True)
dialog.show_download_progress(256, 1024)
app.processEvents()
assert dialog.exit_button.isVisible()
assert dialog.exit_button.isEnabled()
assert not dialog.update_button.isEnabled()
assert not dialog.cancel_button.isVisible()
dialog.close()
app.processEvents()
assert dialog.isVisible()
QTest.keyClick(dialog, Qt.Key.Key_Escape)
app.processEvents()
assert dialog.isVisible()
dialog.allow_application_exit()
dialog.close()
app.processEvents()
assert not dialog.isVisible()
dialog.deleteLater()
app.processEvents()
def test_forced_update_exit_button_emits_dedicated_request(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
dialog = AppUpdateDialog(_offer(force=True))
requests: list[bool] = []
dialog.exit_requested.connect(lambda: requests.append(True))
dialog.show()
dialog.set_busy(True)
app.processEvents()
dialog.exit_button.click()
assert requests == [True]
assert dialog.isVisible()
dialog.hide()
dialog.deleteLater()
app.processEvents()
def test_session_quits_immediately_when_update_has_not_started(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
host = QObject()
quit_requests: list[bool] = []
host.request_quit = lambda: quit_requests.append(True) # type: ignore[attr-defined]
session = AppUpdateSession(host)
dialog = AppUpdateDialog(_offer(force=True))
session.dialog = dialog
dialog.show()
app.processEvents()
session._request_exit(dialog)
assert session._cancel_event.is_set()
assert quit_requests == [True]
assert not dialog.exit_button.isEnabled()
assert not dialog.isVisible()
dialog.hide()
dialog.deleteLater()
app.processEvents()
def test_session_waits_for_update_worker_before_quitting(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
host = QObject()
quit_requests: list[bool] = []
host.request_quit = lambda: quit_requests.append(True) # type: ignore[attr-defined]
session = AppUpdateSession(host)
dialog = AppUpdateDialog(_offer(force=True))
session.dialog = dialog
active_signals = QObject()
session._signals = active_signals # type: ignore[assignment]
session._active_install_signals = active_signals # type: ignore[assignment]
dialog.show()
app.processEvents()
session._request_exit(dialog)
assert session._cancel_event.is_set()
assert quit_requests == []
assert not dialog.exit_button.isEnabled()
assert "退出软件" in dialog.status_label.text()
assert dialog.isVisible()
session._finish_install(active_signals, dialog, object()) # type: ignore[arg-type]
assert quit_requests == []
session._on_install_finished(active_signals) # type: ignore[arg-type]
assert quit_requests == [True]
assert session._active_install_signals is None
assert not dialog.isVisible()
dialog.hide()
dialog.deleteLater()
app.processEvents()
def test_unavailable_update_dialog_shows_policy_reason(
application: QApplication | None = None,
) -> None:
app = application or QApplication.instance() or QApplication([])
apply_theme(app)
reason = "无法自动安装:自动安装 Windows 更新必须开启 HTTPS 证书校验。"
dialog = AppUpdateDialog(
_offer(
force=False,
can_install=False,
install_unavailable_reason=reason,
)
)
dialog.show()
app.processEvents()
assert dialog.status_label.text() == reason
assert not dialog.update_button.isEnabled()
dialog.close()
def test_session_explains_disabled_certificate_verification() -> None:
host = QObject()
host.config = SimpleNamespace(verify_ssl=False) # type: ignore[attr-defined]
session = AppUpdateSession(host)
session._generation = 1
presented: list[UpdateOffer] = []
session._present = presented.append # type: ignore[method-assign]
offer = replace(
_offer(force=True),
package=UpdatePackage(
url="https://cdn.example.com/DoctorWorkstation-Setup.exe",
sha256="a" * 64,
size=1024,
filename="DoctorWorkstation-Setup.exe",
type=PACKAGE_TYPE_INNO_SETUP,
),
)
session._on_offer(offer, interactive=True, generation=1)
assert len(presented) == 1
assert presented[0].can_install is False
assert presented[0].force is False
assert presented[0].package is None
assert "开启 HTTPS 证书校验" in presented[0].install_unavailable_reason
assert "取消勾选“信任自签名证书" in presented[0].install_unavailable_reason
+207
View File
@@ -0,0 +1,207 @@
"""登录后聊天通知的契约:轮询、卡片、点击去向。"""
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, QWidget
from doctor_workstation.services.mock_repository import DemoDoctorRepository
from doctor_workstation.services.repository import RemoteDoctorRepository
from doctor_workstation.ui import chat_notifications as chat_module
from doctor_workstation.ui.chat_notifications import (
CONSULTATION_COMPLETE,
PATIENT_LEFT_CHAT,
PATIENT_OPENED_CHAT,
ChatNotificationCenter,
parse_notification,
relative_time,
)
@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(chat_module, "run_async", run_immediately)
class _NotifyRepository:
def __init__(self, *batches: list[dict[str, Any]]) -> None:
self.batches = list(batches)
self.calls = 0
def list_chat_notifications(self) -> list[dict[str, Any]]:
self.calls += 1
# 服务端取一次即消费,这里同样只发一次。
return self.batches.pop(0) if self.batches else []
def _row(identifier: str, kind: str = PATIENT_OPENED_CHAT, **extra: Any) -> dict[str, Any]:
row = {
"id": identifier,
"type": kind,
"doctor_id": 7,
"patient_id": "11676",
"patient_name": "甘先生",
"created_at": 1787882294,
}
row.update(extra)
return row
def test_rows_normalize_into_admin_equivalent_cards() -> None:
opened = parse_notification(_row("a1"))
assert opened is not None
assert opened.title == "患者打开会话"
assert opened.description == "甘先生 已打开与您的会话,请及时查看"
assert opened.action_text == "去接诊台"
left = parse_notification(_row("a2", PATIENT_LEFT_CHAT))
assert left is not None
assert left.description == "甘先生 已离开问诊会话页面"
complete = parse_notification(
_row("a3", CONSULTATION_COMPLETE, doctor_name="陈医生", diagnosis_id="8169")
)
assert complete is not None
assert complete.diagnosis_id == 8169
assert complete.description == "甘先生 的面诊已由 陈医生 完成,请及时跟进"
# 缺 id、未知 type、非映射行都不该变成卡片。
assert parse_notification(_row("", PATIENT_OPENED_CHAT)) is None
assert parse_notification(_row("a4", "unknown_business")) is None
assert parse_notification("not-a-row") is None
def test_relative_time_matches_the_admin_wording() -> None:
now = 1787882294 + 0.0
assert relative_time(1787882294, now=now) == "刚刚"
assert relative_time(1787882294 - 120, now=now) == "2 分钟前"
assert relative_time(1787882294 - 7200, now=now) == "2 小时前"
assert relative_time(0, now=now) == ""
def test_center_polls_once_per_tick_and_never_repeats_a_card(
application: QApplication,
immediate_async: None,
) -> None:
host = QWidget()
host.resize(1280, 800)
repository = _NotifyRepository([_row("a1"), _row("a1")], [_row("a2", PATIENT_LEFT_CHAT)])
center = ChatNotificationCenter(repository, host)
center.poll()
assert [item.id for item in center.pending] == ["a1"]
center.poll()
# 同一条通知重复下发也只留一张卡片,新的排在最前面。
assert [item.id for item in center.pending] == ["a2", "a1"]
assert repository.calls == 2
assert center.isVisible() is False or len(center.pending) == 2
center.dismiss("a1")
assert [item.id for item in center.pending] == ["a2"]
center.clear()
assert center.pending == []
host.deleteLater()
def test_center_keeps_only_the_newest_five_cards(
application: QApplication,
immediate_async: None,
) -> None:
host = QWidget()
repository = _NotifyRepository([_row(f"n{index}") for index in range(8)])
center = ChatNotificationCenter(repository, host)
center.poll()
assert [item.id for item in center.pending] == ["n7", "n6", "n5", "n4", "n3"]
host.deleteLater()
def test_activating_a_card_emits_it_once_and_removes_it(
application: QApplication,
immediate_async: None,
) -> None:
host = QWidget()
repository = _NotifyRepository([_row("a1", CONSULTATION_COMPLETE, diagnosis_id=8169)])
center = ChatNotificationCenter(repository, host)
activated: list[Any] = []
center.notification_activated.connect(activated.append)
center.poll()
card = next(iter(center._cards.values()))
card.open_button.click()
assert [item.diagnosis_id for item in activated] == [8169]
assert center.pending == []
host.deleteLater()
def test_center_stays_silent_when_the_source_cannot_answer(
application: QApplication,
immediate_async: None,
) -> None:
class _Failing:
def list_chat_notifications(self) -> list[dict[str, Any]]:
raise RuntimeError("服务暂时不可用")
host = QWidget()
center = ChatNotificationCenter(_Failing(), host)
center.poll()
assert center.pending == []
# 演示仓储与不支持该接口的数据源都不应该报错。
ChatNotificationCenter(DemoDoctorRepository(), host).poll()
ChatNotificationCenter(object(), host).poll()
host.deleteLater()
class _RecordingClient:
def __init__(self, payload: Any) -> None:
self.payload = payload
self.get_calls: list[tuple[str, dict[str, Any]]] = []
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
self.get_calls.append((endpoint, dict(params or {})))
return self.payload
def test_remote_consumes_the_same_admin_endpoint() -> None:
client = _RecordingClient([_row("a1")])
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
rows = repository.list_chat_notifications()
assert client.get_calls == [("chat/notifications", {})]
assert [row["id"] for row in rows] == ["a1"]
+82 -2
View File
@@ -1,9 +1,11 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from doctor_workstation import config as config_module
from doctor_workstation.config import AppConfig, normalize_api_base_url
@@ -36,8 +38,86 @@ def test_config_update_validates_video_mode() -> None:
def test_config_update_normalizes_ssl_boolean_strings() -> None:
assert AppConfig().with_updates(verify_ssl="false").verify_ssl is False
assert AppConfig(verify_ssl=False).with_updates(verify_ssl="true").verify_ssl is True
assert AppConfig(debug_mode=True).with_updates(verify_ssl="false").verify_ssl is False
assert (
AppConfig(debug_mode=True, verify_ssl=False)
.with_updates(verify_ssl="true")
.verify_ssl
is True
)
def test_production_config_locks_online_server_and_disables_demo(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
config_dir = tmp_path / "config"
config_dir.mkdir()
(config_dir / "preferences.json").write_text(
json.dumps(
{
"api_base_url": "https://stale.example.test/adminapi",
"demo_mode": True,
"verify_ssl": False,
}
),
encoding="utf-8",
)
monkeypatch.setattr(config_module, "DEBUG_MODE", False)
monkeypatch.setattr(
config_module,
"ONLINE_API_BASE_URL",
"https://prod.example.test",
)
monkeypatch.setenv("DOCTOR_CONFIG_DIR", str(config_dir))
monkeypatch.setenv("DOCTOR_API_BASE_URL", "https://env.example.test")
monkeypatch.setenv("DOCTOR_DEMO_MODE", "true")
monkeypatch.setenv("DOCTOR_VERIFY_SSL", "false")
config = AppConfig.load()
assert config.debug_mode is False
assert config.api_base_url == "https://prod.example.test/adminapi"
assert config.demo_mode is False
assert config.verify_ssl is True
updated = config.with_updates(
api_base_url="https://changed.example.test",
demo_mode=True,
verify_ssl=False,
)
assert updated.api_base_url == config.api_base_url
assert updated.demo_mode is False
assert updated.verify_ssl is True
def test_debug_config_keeps_environment_server_controls(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(config_module, "DEBUG_MODE", True)
monkeypatch.setenv("DOCTOR_CONFIG_DIR", str(tmp_path / "config"))
monkeypatch.setenv("DOCTOR_API_BASE_URL", "http://127.0.0.1:8080")
monkeypatch.setenv("DOCTOR_DEMO_MODE", "true")
monkeypatch.setenv("DOCTOR_VERIFY_SSL", "false")
config = AppConfig.load()
assert config.debug_mode is True
assert config.api_base_url == "http://127.0.0.1:8080/adminapi"
assert config.demo_mode is True
assert config.verify_ssl is False
def test_production_config_rejects_empty_online_domain(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
) -> None:
monkeypatch.setattr(config_module, "DEBUG_MODE", False)
monkeypatch.setattr(config_module, "ONLINE_API_BASE_URL", "")
monkeypatch.setenv("DOCTOR_CONFIG_DIR", str(tmp_path / "config"))
with pytest.raises(ValueError, match="ONLINE_API_BASE_URL 不能为空"):
AppConfig.load()
def test_runtime_directories_can_be_isolated_without_replacing_user_home(
@@ -16,6 +16,7 @@ from doctor_workstation.ui.diagnosis_drawer import (
NotesTimeline,
_RemoteImageButton,
)
from doctor_workstation.ui.diagnosis_media import ImagePreviewDialog
@pytest.fixture(scope="module")
@@ -112,7 +113,6 @@ def test_remote_image_request_is_thread_owned_and_rejects_stale_results(
object_name="DiagnosisTongueThumb",
parent=owner,
)
button._manager.deleteLater()
manager = _FakeManager(button)
button._manager = manager
manager.queue(_png_bytes(180, 90, "#DC2626"))
@@ -160,7 +160,6 @@ def test_remote_image_uses_text_only_after_request_or_decode_failure(
object_name="DiagnosisChatImage",
parent=owner,
)
button._manager.deleteLater()
manager = _FakeManager(button)
button._manager = manager
manager.queue(b"not-an-image")
@@ -193,7 +192,6 @@ def test_remote_image_enforces_same_origin_redirects_and_aborts_oversize_downloa
object_name="DiagnosisTongueThumb",
parent=owner,
)
button._manager.deleteLater()
manager = _FakeManager(button)
button._manager = manager
manager.queue(b"must-not-be-read")
@@ -218,6 +216,110 @@ def test_remote_image_enforces_same_origin_redirects_and_aborts_oversize_downloa
assert application.thread() == button.thread()
def _preview_with_offline_transport(
sources: list[str],
*,
index: int = 0,
names: list[str] | None = None,
) -> tuple[ImagePreviewDialog, _FakeManager]:
"""Build a preview window whose downloads are driven by the test, not the network."""
original_request = ImagePreviewDialog._request
ImagePreviewDialog._request = lambda self, target: None # type: ignore[method-assign]
try:
dialog = ImagePreviewDialog(sources, index=index, names=names)
finally:
ImagePreviewDialog._request = original_request # type: ignore[method-assign]
manager = _FakeManager(dialog)
dialog._manager = manager
return dialog, manager
def test_image_preview_pages_the_group_in_app_and_reuses_decoded_images(
application: QApplication,
) -> None:
dialog, manager = _preview_with_offline_transport(
[
"https://media.example.invalid/tongue-1.jpg",
"file:///C:/private/tongue.jpg",
"https://media.example.invalid/tongue-2.jpg",
],
index=2,
names=["舌象附件 1", "本地危险附件", "舌象附件 2"],
)
# file:// 附件既不进入分组,也不会发起任何请求。
assert dialog.sources == [
"https://media.example.invalid/tongue-1.jpg",
"https://media.example.invalid/tongue-2.jpg",
]
assert dialog.current_source == "https://media.example.invalid/tongue-2.jpg"
assert dialog.counter.text() == "第 2 / 2 张"
assert dialog.name_label.text() == "舌象附件 2"
manager.queue(_png_bytes(320, 200, "#DC2626"))
dialog.reload_current()
request = manager.request_objects[-1]
assert request.attribute(QNetworkRequest.Attribute.RedirectPolicyAttribute) == (
QNetworkRequest.RedirectPolicy.SameOriginRedirectPolicy
)
manager.replies[-1].finished.emit()
assert dialog.canvas.text() == ""
assert not dialog.canvas.pixmap().isNull()
manager.queue(_png_bytes(120, 90, "#16A34A"))
dialog.step(1)
assert dialog.current_source == "https://media.example.invalid/tongue-1.jpg"
manager.replies[-1].finished.emit()
assert not dialog.canvas.pixmap().isNull()
dialog.step(1)
assert dialog.current_source == "https://media.example.invalid/tongue-2.jpg"
assert manager.requests == [
"https://media.example.invalid/tongue-2.jpg",
"https://media.example.invalid/tongue-1.jpg",
]
assert not dialog.canvas.pixmap().isNull()
dialog.close()
def test_image_preview_aborts_oversize_and_falls_back_on_undecodable_payload(
application: QApplication,
) -> None:
dialog, manager = _preview_with_offline_transport(
["https://media.example.invalid/tongue.jpg"]
)
manager.queue(b"must-not-be-read")
dialog.reload_current()
reply = manager.replies[-1]
reply.downloadProgress.emit(dialog._MAX_IMAGE_BYTES, -1)
assert reply.aborted is False
reply.downloadProgress.emit(dialog._MAX_IMAGE_BYTES + 1, -1)
assert reply.aborted is True
reply.finished.emit()
assert reply.read_all_calls == 0
assert "12 MB" in dialog.canvas.text()
manager.queue(b"not-an-image")
dialog.reload_current()
manager.replies[-1].finished.emit()
assert dialog.canvas.pixmap().isNull()
assert "无法在工作站内预览" in dialog.canvas.text()
dialog.close()
def test_image_preview_refuses_a_group_without_any_safe_http_source(
application: QApplication,
) -> None:
dialog = ImagePreviewDialog(["file:///C:/private/tongue.jpg", ""])
assert dialog.has_images() is False
assert dialog.sources == []
assert dialog.current_source == ""
assert not dialog.external_button.isEnabled()
assert not dialog.next_button.isEnabled()
assert "HTTP(S)" in dialog.canvas.text()
dialog.close()
def test_notes_render_cover_thumbnail_and_keep_safe_open_and_single_delete(
application: QApplication,
monkeypatch: pytest.MonkeyPatch,
+144
View File
@@ -0,0 +1,144 @@
"""诊单字典 / 枚举 / 时间戳翻译的契约。"""
from __future__ import annotations
from typing import Any
import pytest
from doctor_workstation.services.mock_repository import DemoDoctorRepository
from doctor_workstation.services.repository import RemoteDoctorRepository
from doctor_workstation.ui.diagnosis_terms import (
DICTIONARY_TYPES,
MULTI_VALUE_DICTIONARIES,
SINGLE_VALUE_DICTIONARIES,
TermIndex,
format_timestamp,
unit_suffix,
)
def test_seed_dictionary_translates_codes_admin_shows_in_chinese() -> None:
terms = TermIndex()
assert terms.dictionary_label("appetite", "dry,bitter") == "干、苦"
assert terms.dictionary_label("appetite", ["dry", "greasy"]) == "干、腻"
assert terms.dictionary_label("weight_change", "lose_10_jin") == "瘦10斤"
assert terms.dictionary_label("fatty_liver_degree", "mild") == "轻度"
assert terms.dictionary_label("past_history", "hypertension、diabetes") == "高血压、糖尿病"
# 同一个 code 在不同字典里含义不同,翻译必须按字段所属字典走。
assert terms.dictionary_label("skin_condition", "dry") == "干燥"
assert terms.dictionary_label("eye_condition", "dry") == "干涩"
# 不在字典里的自定义值回显原值,不会被吞掉。
assert terms.dictionary_label("appetite", "自定义症状") == "自定义症状"
assert terms.dictionary_label("remark", "任意文本") is None
def test_backend_text_field_wins_over_dictionary_and_raw_value() -> None:
terms = TermIndex()
assert terms.display({"appetite": "dry", "appetite_text": "口干"}, "appetite") == "口干"
assert terms.display({"appetite": "dry"}, "appetite") == ""
assert terms.display({}, "appetite", default="未记录") == "未记录"
def test_live_dictionary_overrides_the_bundled_seed() -> None:
terms = TermIndex()
terms.merge({"appetite": [{"name": "口干", "value": "dry"}]})
assert terms.dictionary_label("appetite", "dry") == "口干"
# 实时字典没覆盖到的条目继续用种子。
assert terms.dictionary_label("appetite", "bitter") == ""
def test_enum_and_timestamp_fields_render_like_the_admin_readonly_page() -> None:
terms = TermIndex()
assert terms.value_label("gender", 1) == ""
assert terms.value_label("gender", "0") == ""
assert terms.value_label("marital_status", "1") == "已婚"
assert terms.value_label("allergy_history", "0") == ""
assert terms.value_label("family_history", 1) == ""
assert terms.value_label("diagnosis_type", "follow_up") == "复诊"
assert terms.value_label("create_source", "admin") == "后台创建"
assert terms.value_label("source", "1") == "患者自录"
assert terms.value_label("create_time", 1783838927) == format_timestamp(1783838927)
assert format_timestamp(1783838927) is not None
assert format_timestamp("2026-08-18 09:20") is None
assert format_timestamp(0) is None
def test_units_only_decorate_numeric_readonly_values() -> None:
assert unit_suffix("height", "162") == " cm"
assert unit_suffix("fasting_blood_sugar", "8.2") == " mmol/L"
assert unit_suffix("diabetes_discovery_year", "6") == ""
# 自由文本("17多"、"五年")不补单位,避免拼出错误的读数。
assert unit_suffix("fasting_blood_sugar", "17多") == ""
assert unit_suffix("diabetes_discovery_year", "五年") == ""
assert unit_suffix("remark", "123") == ""
def test_dictionary_types_cover_every_field_the_backend_translates() -> None:
# 与 AppointmentLogic::enrichDiagnosisLabels 的字段表保持同步。
assert set(SINGLE_VALUE_DICTIONARIES) == {
"diagnosis_type",
"syndrome_type",
"diabetes_type",
"water_intake",
"weight_change",
"fatty_liver_degree",
}
assert set(MULTI_VALUE_DICTIONARIES) == {
"past_history",
"appetite",
"diet_condition",
"body_feeling",
"sleep_condition",
"eye_condition",
"head_feeling",
"sweat_condition",
"skin_condition",
"urine_condition",
"stool_condition",
"kidney_condition",
}
assert set(DICTIONARY_TYPES) == set(SINGLE_VALUE_DICTIONARIES.values()) | set(
MULTI_VALUE_DICTIONARIES.values()
)
class _RecordingClient:
def __init__(self, payload: Any) -> None:
self.payload = payload
self.get_calls: list[tuple[str, dict[str, Any]]] = []
def get(self, endpoint: str, params: dict[str, Any] | None = None) -> Any:
self.get_calls.append((endpoint, dict(params or {})))
return self.payload
def test_remote_batches_every_dictionary_into_one_request() -> None:
client = _RecordingClient(
{
"appetite": [{"name": "口干", "value": "dry"}],
"weight_change": [{"name": "瘦10斤", "value": "lose_10_jin"}],
}
)
repository = RemoteDoctorRepository(client) # type: ignore[arg-type]
dictionaries = repository.get_dictionaries(["appetite", "weight_change", "appetite", ""])
assert client.get_calls == [("config/dict", {"type": "appetite,weight_change"})]
assert list(dictionaries) == ["appetite", "weight_change"]
terms = TermIndex()
terms.merge(dictionaries)
assert terms.dictionary_label("appetite", "dry") == "口干"
@pytest.mark.parametrize("dictionary_type", DICTIONARY_TYPES)
def test_demo_repository_answers_the_batch_dictionary_contract(dictionary_type: str) -> None:
repository = DemoDoctorRepository()
dictionaries = repository.get_dictionaries(DICTIONARY_TYPES)
assert dictionary_type in dictionaries
+29
View File
@@ -1018,6 +1018,35 @@ def test_prescription_detail_can_open_immutable_case_record_tab(
application.processEvents()
def test_case_record_translates_snapshot_dictionary_codes_to_chinese() -> None:
# 处方快照存的是开方当时的原始 code,没有后端补的 *_text。
prescription = {
"id": 21,
"diagnosis_id": 9,
"case_record": {
"diagnosis": {
"appetite": "dry,bitter",
"water_intake": "one_bottle",
"weight_change": "lose_10_jin",
"fatty_liver_degree": "mild",
"past_history": "hypertension,diabetes",
"sleep_condition": "many_dreams",
}
},
}
case_html = dialog_module.render_case_record_html(prescription)
assert "干、苦" in case_html
assert "1瓶矿泉水" in case_html
assert "瘦10斤" in case_html
assert "轻度" in case_html
assert "高血压、糖尿病" in case_html
assert "多梦" in case_html
for code in ("lose_10_jin", "one_bottle", "many_dreams"):
assert code not in case_html
def test_case_record_tab_exports_case_record_as_a3_pdf(
application: QApplication,
tmp_path: Any,
+52
View File
@@ -669,3 +669,55 @@ def test_shell_directional_controls_have_no_unicode_arrow_text(
if hasattr(button, "text") and callable(button.text)
for arrow in ("", "", "", "", "", "", "", "")
)
def test_chat_notification_takes_the_doctor_to_the_matching_workspace(
application: QApplication,
shell_window: ShellWindow,
monkeypatch: pytest.MonkeyPatch,
) -> None:
opened: list[Any] = []
monkeypatch.setattr(
ShellWindow,
"open_diagnosis_by_id",
lambda self, diagnosis_id, *, modeless=False: opened.append(diagnosis_id),
)
center = shell_window.chat_notifications
assert shell_window.navigate("consultations")
center.add_notifications(
[
{
"id": "n1",
"type": "patient_opened_chat",
"patient_name": "甘先生",
"created_at": 1787882294,
}
]
)
application.processEvents()
assert [item.id for item in center.pending] == ["n1"]
# 患者进入会话 → 直接落到接诊台。
next(iter(center._cards.values())).open_button.click()
application.processEvents()
assert shell_window._active_page_key == "reception"
assert center.pending == []
# 面诊结束 → 打开对应诊单。
center.add_notifications(
[
{
"id": "n2",
"type": "consultation_complete",
"patient_name": "甘先生",
"doctor_name": "陈医生",
"diagnosis_id": 8169,
"created_at": 1787882294,
}
]
)
next(iter(center._cards.values())).open_button.click()
application.processEvents()
assert opened == [8169]
assert center.pending == []
+73
View File
@@ -240,6 +240,7 @@ def test_real_demo_login_reaches_success_without_widget_adapter(
api_base_url="https://127.0.0.1:9",
request_timeout=30,
demo_mode=True,
debug_mode=True,
remembered_account="",
)
payloads: list[dict[str, Any]] = []
@@ -346,6 +347,7 @@ def test_server_settings_panel_keeps_controls_separated_at_minimum_window(
api_base_url="",
request_timeout=30,
demo_mode=False,
debug_mode=True,
remembered_account="",
)
window = LoginWindow(object(), config=config, settings=settings)
@@ -375,6 +377,49 @@ def test_server_settings_panel_keeps_controls_separated_at_minimum_window(
application.processEvents()
def test_production_login_hides_and_blocks_debug_controls(tmp_path: Any) -> None:
application = QApplication.instance() or QApplication([])
settings = QSettings(str(tmp_path / "production.ini"), QSettings.Format.IniFormat)
settings.setValue("server/base_url", "https://stale.example.test")
settings.setValue("server/verify_ssl", False)
remote_repository = object()
demo_repository = DemoDoctorRepository()
config = SimpleNamespace(
api_base_url="https://prod.example.test/adminapi",
request_timeout=30,
verify_ssl=True,
demo_mode=True,
debug_mode=False,
remembered_account="",
)
window = LoginWindow(
remote_repository,
config=config,
demo_repository=demo_repository,
settings=settings,
)
window.show()
application.processEvents()
assert not window.demo_check.isVisible()
assert not window.debug_settings_section.isVisible()
assert not window.server_toggle.isVisible()
assert not window.server_panel.isVisible()
assert not window.demo_check.isChecked()
assert window.active_repository is remote_repository
assert window.server_url_edit.text() == "https://prod.example.test/adminapi"
assert window._credential_scope() == "https://prod.example.test/adminapi"
window._on_demo_toggled(True)
window._toggle_server_panel(True)
assert not window.demo_check.isChecked()
assert window.active_repository is remote_repository
assert window.server_panel.isHidden()
window.close()
application.processEvents()
def test_server_settings_can_persist_self_signed_debug_mode(tmp_path: Any) -> None:
application = QApplication.instance() or QApplication([])
settings = QSettings(str(tmp_path / "self-signed.ini"), QSettings.Format.IniFormat)
@@ -383,6 +428,7 @@ def test_server_settings_can_persist_self_signed_debug_mode(tmp_path: Any) -> No
request_timeout=30,
verify_ssl=True,
demo_mode=False,
debug_mode=True,
remembered_account="",
)
window = LoginWindow(object(), config=config, settings=settings)
@@ -408,6 +454,7 @@ def test_login_applies_self_signed_setting_before_authentication(
config = AppConfig(
api_base_url="https://internal.example.test/adminapi",
demo_mode=False,
debug_mode=True,
verify_ssl=True,
)
calls: list[str] = []
@@ -500,6 +547,7 @@ def test_certificate_error_opens_server_settings(tmp_path: Any) -> None:
request_timeout=30,
verify_ssl=True,
demo_mode=False,
debug_mode=True,
remembered_account="",
)
window = LoginWindow(object(), config=config, settings=settings)
@@ -513,6 +561,31 @@ def test_certificate_error_opens_server_settings(tmp_path: Any) -> None:
application.processEvents()
def test_certificate_error_does_not_reveal_production_server_settings(tmp_path: Any) -> None:
application = QApplication.instance() or QApplication([])
settings = QSettings(str(tmp_path / "production-certificate.ini"), QSettings.Format.IniFormat)
config = SimpleNamespace(
api_base_url="https://prod.example.test/adminapi",
request_timeout=30,
verify_ssl=True,
demo_mode=False,
debug_mode=False,
remembered_account="",
)
window = LoginWindow(object(), config=config, settings=settings)
window.show()
window._on_login_error(RuntimeError("[SSL: CERTIFICATE_VERIFY_FAILED]"))
application.processEvents()
assert not window.server_toggle.isChecked()
assert not window.debug_settings_section.isVisible()
assert window.server_panel.isHidden()
assert "联系管理员" in window.error_banner.label.text()
window.close()
application.processEvents()
def test_business_dialogs_can_be_maximized_but_prompts_cannot() -> None:
"""Dense AI panels and editors were stuck at their constructed size."""