338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""Data provenance and native UI checks for the prescription overview page."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from copy import deepcopy
|
||
|
||
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
||
|
||
import pytest
|
||
from PySide6.QtCore import Qt
|
||
from PySide6.QtWidgets import QApplication, QLabel, QVBoxLayout, QWidget
|
||
|
||
from doctor_workstation.ui.dialogs.issued_prescription_ai_workspace import (
|
||
PrescriptionReviewWorkspace,
|
||
checklist_items,
|
||
diagnosis_text,
|
||
dose_deltas,
|
||
gap_counts,
|
||
review_rows,
|
||
)
|
||
|
||
|
||
def batch(count: int = 3) -> dict:
|
||
rows = []
|
||
herbs = []
|
||
for index in range(count):
|
||
herb = {"name": f"药材{index}", "dosage": "15.00", "unit": "g", "dose_basis": "per_dose", "formula_type": "主方", "processing": "生品"}
|
||
rows.append({"key": f"saved-identity-{index}", "name": herb["name"], "doctor": {**herb, "dosage": "30.00"}, "candidate": {**herb, "source_rows": [index]}, "match_type": "matched"})
|
||
herbs.append(herb)
|
||
model = {"status": "succeeded", "report": {"diagnosis": "已保存的辨证意见", "summary": "不能当作诊断的摘要", "missing_information": ["缺少舌脉记录"]}, "candidate": {"status": "available_for_review", "herbs": herbs}, "comparison": {"status": "comparable", "rows": rows}}
|
||
return {"id": 4, "validity": "current", "models": {"qwen": deepcopy(model), "openai": deepcopy(model)}, "doctor_snapshot": {"patient": {"name": "测试患者", "gender": "male", "age": 50}, "diagnosis": {"western_diagnosis": "已记录西医诊断", "tcm_diagnosis": "已记录中医诊断", "syndrome": "已记录证候"}, "prescription": {"herbs": deepcopy(herbs), "usage_instruction": "水煎服", "usage_days": 7, "times_per_day": 2}}}
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def application():
|
||
return QApplication.instance() or QApplication([])
|
||
|
||
|
||
@pytest.fixture
|
||
def workspace(application):
|
||
host = QWidget()
|
||
host.resize(1024, 760)
|
||
layout = QVBoxLayout(host)
|
||
layout.setContentsMargins(0, 0, 0, 0)
|
||
widget = PrescriptionReviewWorkspace(host)
|
||
layout.addWidget(widget)
|
||
host.show()
|
||
application.processEvents()
|
||
yield widget
|
||
host.close()
|
||
host.deleteLater()
|
||
application.processEvents()
|
||
|
||
|
||
def test_three_series_need_saved_unique_identity_and_same_baseline():
|
||
source = batch()
|
||
before = deepcopy(source)
|
||
rows, _ = review_rows(source)
|
||
assert source == before
|
||
assert len(rows) == 3
|
||
assert all(set(row.doses) == {"doctor", "qwen", "openai"} for row in rows)
|
||
assert rows[0].scale == ("克", "每剂")
|
||
assert rows[0].changed
|
||
assert "15.00 克 / 每剂" in rows[0].description
|
||
assert "候选原方记录：第 1 项 药材0" in rows[0].description
|
||
assert "saved-identity" not in rows[0].description
|
||
|
||
|
||
@pytest.mark.parametrize("mutation", ["no_keys", "different_baseline", "duplicate_keys", "different_units", "different_identity"])
|
||
def test_no_guessed_cross_model_join(mutation):
|
||
source = batch(1)
|
||
target = source["models"]["openai"]["comparison"]["rows"][0]
|
||
if mutation == "no_keys":
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0].pop("key")
|
||
elif mutation == "different_baseline":
|
||
target["doctor"]["dosage"] = 31
|
||
elif mutation == "duplicate_keys":
|
||
source["models"]["openai"]["comparison"]["rows"].append(deepcopy(target))
|
||
elif mutation == "different_units":
|
||
target["candidate"]["unit"] = "mg"
|
||
else:
|
||
target["candidate"]["processing"] = "炙品"
|
||
rows, _ = review_rows(source)
|
||
assert len(rows) >= 2
|
||
assert all(len(row.entries) == 1 for row in rows)
|
||
|
||
|
||
@pytest.mark.parametrize("value", ["nan", "-3", "1/2", "Infinity", None])
|
||
def test_invalid_numbers_and_missing_are_never_zero(value):
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0]["candidate"]["dosage"] = value
|
||
rows, _ = review_rows(source)
|
||
assert rows[0].scale is None
|
||
assert not rows[0].changed
|
||
if value is None:
|
||
assert rows[0].entries["qwen"].candidate.label == "—"
|
||
|
||
|
||
@pytest.mark.parametrize("field,value", [("validity", "stale"), ("validity", "source_updated"), ("validity", None), ("status", "running"), ("status", "failed"), ("comparison", "not_comparable"), ("candidate", "withheld_for_risk")])
|
||
def test_pending_historical_and_error_are_text_only(field, value):
|
||
source = batch()
|
||
if field == "validity":
|
||
source[field] = value
|
||
else:
|
||
for model in source["models"].values():
|
||
if field == "status":
|
||
model[field] = value
|
||
else:
|
||
model[field]["status"] = value
|
||
rows, _ = review_rows(source)
|
||
assert rows
|
||
assert all(row.scale is None for row in rows)
|
||
assert all(not row.changed for row in rows)
|
||
|
||
|
||
def test_unaccounted_candidates_and_missing_trace_survive():
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["candidate"]["herbs"].append({"name": "未规范炮制药", "dosage": "2.5", "unit": "g"})
|
||
model["comparison"]["rows"][0]["candidate"].pop("source_rows")
|
||
rows, _ = review_rows(source)
|
||
assert len(rows) == 5
|
||
originals = [row for row in rows if next(iter(row.entries.values())).origin != "comparison"]
|
||
assert len(originals) == 4
|
||
assert all(row.scale is None for row in originals)
|
||
assert sum(row.name == "未规范炮制药" for row in originals) == 2
|
||
|
||
|
||
def test_diagnosis_never_fabricated_from_summary():
|
||
assert diagnosis_text({"report": {"summary": "摘要疾病"}}) == "诊断意见未保存"
|
||
assert diagnosis_text({"report": {"diagnosis": "辨证原文"}}) == "辨证原文"
|
||
assert diagnosis_text({"report": {"diagnosis": {"western_diagnosis": "诊断原值"}}}) == "西医诊断：诊断原值"
|
||
|
||
|
||
def test_missing_side_is_not_a_zero_dose_difference():
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0]["doctor"] = None
|
||
rows, _ = review_rows(source)
|
||
assert rows[0].doctor is None
|
||
assert not rows[0].changed
|
||
|
||
|
||
def test_historical_report_states_itself_instead_of_reporting_zero_differences(workspace):
|
||
source = batch()
|
||
source["validity"] = "stale"
|
||
workspace.set_batch(source)
|
||
assert "历史或失效报告" in workspace.summary_text
|
||
assert workspace.doses.rows == [] # 不可比就不画，不是画成 0
|
||
assert workspace.doses.empty.isVisible()
|
||
|
||
|
||
def test_每味药与原方的距离按共同标尺画出(workspace):
|
||
workspace.set_batch(batch())
|
||
assert [delta.name for delta in workspace.doses.rows] == ["药材0", "药材1", "药材2"]
|
||
assert all(delta.deltas["qwen"] == -15 for delta in workspace.doses.rows)
|
||
assert "3 项剂量差异" in workspace.summary_text
|
||
axis = workspace.doses.body.findChildren(QWidget)
|
||
described = [widget.accessibleDescription() for widget in axis if widget.accessibleName() == "剂量差异条"]
|
||
assert "药材0 千问 −15克 OpenAI −15克" in described
|
||
|
||
|
||
def test_a_herb_one_model_never_listed_is_marked_absent_not_zero(workspace):
|
||
source = batch(1)
|
||
source["models"]["openai"]["comparison"]["rows"][0]["candidate"] = None
|
||
source["models"]["openai"]["candidate"]["herbs"] = []
|
||
workspace.set_batch(source)
|
||
delta = workspace.doses.rows[0]
|
||
assert delta.deltas["openai"] is None
|
||
assert delta.badge == "OpenAI 未收录"
|
||
assert delta.deltas["qwen"] == -15
|
||
|
||
|
||
def test_an_unreadable_saved_dose_takes_the_whole_row_off_the_chart(workspace):
|
||
source = batch(1)
|
||
source["models"]["openai"]["comparison"]["rows"][0]["candidate"]["dosage"] = "nan"
|
||
workspace.set_batch(source)
|
||
assert workspace.doses.rows == []
|
||
|
||
|
||
def test_an_incomparable_basis_keeps_the_row_off_the_chart(workspace):
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0]["doctor"]["dose_basis"] = "per_day"
|
||
workspace.set_batch(source)
|
||
assert workspace.doses.rows == []
|
||
|
||
|
||
def test_a_pending_model_adds_no_row_and_no_difference(workspace):
|
||
source = batch(1)
|
||
source["models"]["openai"] = {"status": "running"}
|
||
workspace.set_batch(source)
|
||
assert len(workspace.doses.rows) == 1
|
||
assert workspace.doses.rows[0].deltas["openai"] is None
|
||
|
||
|
||
def test_new_herbs_carry_their_full_dose_and_say_which_model_added_them(workspace):
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0]["doctor"] = None
|
||
workspace.set_batch(source)
|
||
delta = workspace.doses.rows[0]
|
||
assert delta.doctor is None
|
||
assert delta.deltas["qwen"] == 15
|
||
|
||
|
||
def test_gap_counts_split_the_saved_gaps_into_three_buckets() -> None:
|
||
source = batch(1)
|
||
source["missing"] = [{"code": "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "critical": True},
|
||
{"code": "ATTACHMENT_STORAGE_RESTRICTED"},
|
||
{"code": "ARCHIVE_SYNC_WATERMARK_UNAVAILABLE"}]
|
||
assert gap_counts(source) == {"critical": 1, "attachment": 1, "other": 1}
|
||
assert gap_counts(batch(1)) == {"critical": 0, "attachment": 0, "other": 0}
|
||
|
||
|
||
def _conclusion_texts(workspace) -> list[str]:
|
||
"""The numbered sentences, skipping the hairlines the design puts between them."""
|
||
|
||
texts = []
|
||
for index in range(workspace.conclusions.body_layout.count()):
|
||
widget = workspace.conclusions.body_layout.itemAt(index).widget()
|
||
labels = widget.findChildren(QLabel) if widget is not None else []
|
||
if labels:
|
||
texts.append(labels[-1].text())
|
||
return texts
|
||
|
||
|
||
def test_conclusions_state_the_counts_they_were_built_from(workspace) -> None:
|
||
workspace.set_batch(batch(3))
|
||
texts = _conclusion_texts(workspace)
|
||
assert any("两个模型都给出了候选方" in text for text in texts)
|
||
assert any("剂量偏差共" in text for text in texts)
|
||
assert any("没有记录资料缺口" in text for text in texts)
|
||
|
||
|
||
def test_conclusions_never_invent_a_score_comparison(workspace) -> None:
|
||
source = batch(1)
|
||
source["models"]["openai"]["comparison"]["status"] = "not_comparable"
|
||
workspace.set_batch(source)
|
||
texts = _conclusion_texts(workspace)
|
||
assert any("只有可比的一侧有分数" in text for text in texts)
|
||
|
||
|
||
def test_attribution_groups_every_herb_exactly_once(workspace) -> None:
|
||
source = batch(2)
|
||
source["models"]["openai"]["candidate"]["herbs"] = [{"name": "药材0", "dosage": "15.00", "unit": "g"}]
|
||
workspace.set_batch(source)
|
||
assert workspace.attribution.rows["all"]["count"].text() == "1 味"
|
||
assert workspace.attribution.rows["doctor_only"]["count"].text() == "0 味"
|
||
assert workspace.attribution.rows["openai_only"]["count"].text() == "0 味"
|
||
# 药材1 只有医方与千问共用，四个分组都不含它，标题必须说明这一点
|
||
assert "合计 2 味" in workspace.attribution.total_note.text()
|
||
assert "另 1 味为医方与单一模型共用" in workspace.attribution.total_note.text()
|
||
|
||
|
||
def test_risk_panel_flags_only_large_changes(workspace) -> None:
|
||
small = batch(1)
|
||
for model in small["models"].values():
|
||
model["comparison"]["rows"][0]["candidate"]["dosage"] = "29.00"
|
||
workspace.set_batch(small)
|
||
assert workspace.risks.empty.isVisible() # 差 1 克不进清单
|
||
large = batch(1)
|
||
for model in large["models"].values():
|
||
model["comparison"]["rows"][0]["candidate"]["dosage"] = "12.00"
|
||
workspace.set_batch(large)
|
||
assert workspace.risks.body.isVisible()
|
||
assert "2 项" in workspace.risks.hint.text()
|
||
|
||
|
||
def test_checklist_merges_the_models_and_keeps_the_severe_items_first():
|
||
source = batch(1)
|
||
source["models"]["qwen"]["report"]["risk_assessment"] = [{"level": "high", "label": "血压数据缺失"}]
|
||
source["missing"] = [{"code": "TRANSCRIPT_NOT_VERIFIED_COMPLETE", "critical": True}]
|
||
items = checklist_items(source)
|
||
assert items[0][0] == "血压数据缺失"
|
||
assert items[0][1] == "千问 关键"
|
||
shared = next(item for item in items if item[0] == "缺少舌脉记录")
|
||
assert shared[1] == "千问、OpenAI"
|
||
assert any("转写" in item[0] for item in items)
|
||
|
||
|
||
def test_checklist_shows_the_saved_items_and_says_so_when_empty(workspace):
|
||
workspace.set_batch(batch())
|
||
assert "1 条 · 按严重度排序" in workspace.checklist.count_note.text()
|
||
assert not workspace.checklist.empty.isVisible()
|
||
workspace.set_batch({"id": 9, "validity": "current", "models": {}})
|
||
assert workspace.checklist.empty.isVisible()
|
||
assert workspace.checklist.count_note.text() == "暂无待确认项"
|
||
|
||
|
||
def test_rerendering_leaves_no_stale_row_widgets(workspace, application):
|
||
workspace.set_batch(batch(20))
|
||
application.processEvents()
|
||
assert len(_axes(workspace)) == 20
|
||
workspace.set_batch(batch(3))
|
||
application.processEvents()
|
||
assert len(_axes(workspace)) == 3
|
||
|
||
|
||
def _axes(workspace):
|
||
return [widget for widget in workspace.doses.body.findChildren(QWidget)
|
||
if widget.accessibleName() == "剂量差异条" and widget.parentWidget() is not None]
|
||
|
||
|
||
def test_page_fits_a_1024_window_without_horizontal_scrolling(workspace, application):
|
||
workspace.set_batch(batch())
|
||
application.processEvents()
|
||
assert workspace.minimumSizeHint().width() <= 900
|
||
assert workspace.width() == 1024
|
||
assert workspace.doses.scroll.horizontalScrollBarPolicy() == Qt.ScrollBarPolicy.ScrollBarAlwaysOff
|
||
assert workspace.checklist.scroll.horizontalScrollBarPolicy() == Qt.ScrollBarPolicy.ScrollBarAlwaysOff
|
||
|
||
|
||
def test_dose_deltas_ignore_rows_without_a_shared_scale():
|
||
source = batch(1)
|
||
for model in source["models"].values():
|
||
model["comparison"]["rows"][0]["candidate"]["unit"] = "mg"
|
||
rows, _states = review_rows(source)
|
||
assert dose_deltas(rows) == []
|
||
|
||
|
||
def test_review_slot_has_no_visible_parentless_widget(workspace, application):
|
||
before = {widget for widget in application.topLevelWidgets() if widget.isVisible()}
|
||
first = QLabel("第一模型复核", workspace)
|
||
second = QLabel("双模型复核", workspace)
|
||
workspace.set_review_widget(first)
|
||
workspace.set_review_widget(second)
|
||
application.processEvents()
|
||
assert second.parentWidget() is workspace.checklist.review_slot
|
||
assert not first.isVisible()
|
||
assert first.parentWidget() is workspace.checklist.review_slot
|
||
assert {widget for widget in application.topLevelWidgets() if widget.isVisible()} == before
|
||
|
||
|