feat: 装配两个库（不含 CEUS 结果）
按任务书 01 实现 delivery、library、answers、runs（快照部分）、build 五个模块， scripts/build.py、scripts/scan_library.py 与 config/build.yaml（decode_codes: false）。 交付包整表检查 T1–T4、映射表检查 T5、逐例检查 C01–C13，一例遇到第一项不过即整例跳过。 与任务书原文的出入： - C04 报告表名随视频名配对（负责人 2026-10-16 定）。 - 按两轮独立审阅加严 C11（更多换行符、比例号冒号）、C12（同义词与词根）、 C13（空格、周岁、中文数字、繁体）；泄露扫描做 NFKC；关键图只接受能保真重编码的 PNG； YAML 配置拒绝重复键和空路径。 - import-linter 第 1 条拆成互不依赖与不引用其他 cadx 模块两条。 文档：结构.md 第五、六、八部分按实际更新；新增 guides/装配两个库.md；WORK_LOG 更新。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,267 @@
|
||||
"""测试用假交付包：形状照 40 例交付包写（表头、工作表名、目录结构），内容全部编造。
|
||||
|
||||
这里的表头和文件名是按任务书 01 第二部分的表格独立写的，不取自 delivery 的常量，
|
||||
这样 delivery 的常量写错时测试能发现。各条测试在 `FakeDelivery` 上改出要测的情况，再 `write`。
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import openpyxl
|
||||
import yaml
|
||||
from PIL import Image, PngImagePlugin
|
||||
|
||||
from cadx.build import BuildSummary, run_build
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
DIAGNOSIS_COLUMN = "病理类型（1 IPMN，2 SPN，3 MCN，4 SCN）"
|
||||
CLINICAL_HEADER = [
|
||||
"Case_ID",
|
||||
"Sex",
|
||||
"Age",
|
||||
"Diagnosis",
|
||||
"Symptom_Pain",
|
||||
"Symptom_Jaundice",
|
||||
"Symptom_WeightLoss",
|
||||
"Symptom_Fever",
|
||||
"Symptom_Asymptomatic",
|
||||
"Hx_Pancreatitis",
|
||||
"Hx_Alcohol",
|
||||
"Hx_Gallstones",
|
||||
"Hx_Malignancy",
|
||||
"Hx_Family_PC",
|
||||
"CA199",
|
||||
"CA199_Level",
|
||||
"CEA",
|
||||
"CEA_Level",
|
||||
"Amylase",
|
||||
"Amylase_Level",
|
||||
"Lipase",
|
||||
"Lipase_Level",
|
||||
"TBil",
|
||||
"TBil_Level",
|
||||
]
|
||||
ULTRASOUND_HEADER = [
|
||||
"患者ID",
|
||||
"CEUS视频ID",
|
||||
"性别",
|
||||
"年龄",
|
||||
DIAGNOSIS_COLUMN,
|
||||
"推注开始时间s",
|
||||
"超声描述",
|
||||
]
|
||||
MR_HEADER = ["患者ID", "性别", "年龄", DIAGNOSIS_COLUMN, "MR描述"]
|
||||
|
||||
CLINICAL_PATH = "Clinical_agent/临床工作簿_40例.xlsx"
|
||||
ULTRASOUND_PATH = "CEUS_agent/超声报告文本_40例.xlsx"
|
||||
MR_PATH = "MR_agent/MR报告文本_40例.xlsx"
|
||||
|
||||
# 编造的报告文字：不含亚型名、结论段标题、性别字、「数字岁」和复核措辞。
|
||||
ULTRASOUND_TEXT = "胰体部见一无回声区，边界清晰，壁薄。造影后囊壁呈等增强，内部未见增强。"
|
||||
MR_TEXT = "胰体部见一类圆形长T2信号灶，边界清楚，内见细分隔；增强后分隔轻度强化。"
|
||||
|
||||
# 测试用的样例映射表：含义是编的，只用来验证解码路径。
|
||||
SAMPLE_CODE_MAPS: dict[str, dict[int, str]] = {
|
||||
"Sex": {1: "男", 2: "女"},
|
||||
"Diagnosis": {1: "IPMN", 2: "SPN", 3: "MCN", 4: "SCN"},
|
||||
**{
|
||||
column: {0: "否", 1: "是"}
|
||||
for column in (
|
||||
"Symptom_Pain",
|
||||
"Symptom_Jaundice",
|
||||
"Symptom_WeightLoss",
|
||||
"Symptom_Fever",
|
||||
"Symptom_Asymptomatic",
|
||||
"Hx_Pancreatitis",
|
||||
"Hx_Alcohol",
|
||||
"Hx_Gallstones",
|
||||
"Hx_Malignancy",
|
||||
"Hx_Family_PC",
|
||||
)
|
||||
},
|
||||
**{
|
||||
column: {0: "正常", 1: "异常"}
|
||||
for column in ("CA199_Level", "CEA_Level", "Amylase_Level", "Lipase_Level", "TBil_Level")
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeCase:
|
||||
"""一例：三张表各一行，病例目录里的文件名和单例报告表的行。"""
|
||||
|
||||
number: int
|
||||
clinical: dict[str, object]
|
||||
ultrasound: dict[str, object]
|
||||
mr: dict[str, object]
|
||||
dir_files: list[str]
|
||||
# None 表示单例报告表照抄超声汇总表那一行。
|
||||
case_report_rows: list[list[object]] | None = None
|
||||
case_report_sheet: str = "Sheet1"
|
||||
|
||||
@property
|
||||
def no(self) -> str:
|
||||
return f"{self.number:03d}"
|
||||
|
||||
|
||||
def make_case(number: int) -> FakeCase:
|
||||
"""一例全部检查都能通过的病例。诊断按病例号轮转 1–4。"""
|
||||
no = f"{number:03d}"
|
||||
sex = 1 + number % 2
|
||||
age = 30 + number
|
||||
diagnosis = 1 + (number - 1) % 4
|
||||
clinical: dict[str, object] = dict.fromkeys(CLINICAL_HEADER, 0)
|
||||
clinical.update(
|
||||
{
|
||||
"Case_ID": f"P{no}",
|
||||
"Sex": sex,
|
||||
"Age": age,
|
||||
"Diagnosis": diagnosis,
|
||||
"Symptom_Pain": 1,
|
||||
"CA199": 12.5,
|
||||
"CA199_Level": 0,
|
||||
"CEA": 3,
|
||||
"CEA_Level": 1,
|
||||
"Amylase": None,
|
||||
"Amylase_Level": None,
|
||||
"Lipase": None,
|
||||
"Lipase_Level": None,
|
||||
"TBil": 10.2,
|
||||
"TBil_Level": 0,
|
||||
}
|
||||
)
|
||||
sex_text = {1: "男", 2: "女"}[sex]
|
||||
ultrasound: dict[str, object] = {
|
||||
"患者ID": str(number),
|
||||
"CEUS视频ID": f"U{no}",
|
||||
"性别": sex_text,
|
||||
"年龄": f"{age}岁",
|
||||
DIAGNOSIS_COLUMN: diagnosis,
|
||||
"推注开始时间s": 5,
|
||||
"超声描述": ULTRASOUND_TEXT,
|
||||
}
|
||||
mr: dict[str, object] = {
|
||||
"患者ID": str(number),
|
||||
"性别": sex_text,
|
||||
"年龄": f"{age}岁",
|
||||
DIAGNOSIS_COLUMN: diagnosis,
|
||||
"MR描述": MR_TEXT,
|
||||
}
|
||||
files = [f"U{no}_key.png", f"U{no}_roi2.mp4", f"U{no}_roi2_report.xlsx"]
|
||||
return FakeCase(number, clinical, ultrasound, mr, files)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeDelivery:
|
||||
"""一个假交付包。`extra_rows` 是附加在汇总表末尾的整行（测试编号认不出的行）。"""
|
||||
|
||||
cases: list[FakeCase]
|
||||
sheet_names: dict[str, str] = field(
|
||||
default_factory=lambda: {
|
||||
CLINICAL_PATH: "Sheet",
|
||||
ULTRASOUND_PATH: "超声报告",
|
||||
MR_PATH: "MR报告",
|
||||
}
|
||||
)
|
||||
headers: dict[str, list[str]] = field(
|
||||
default_factory=lambda: {
|
||||
CLINICAL_PATH: list(CLINICAL_HEADER),
|
||||
ULTRASOUND_PATH: list(ULTRASOUND_HEADER),
|
||||
MR_PATH: list(MR_HEADER),
|
||||
}
|
||||
)
|
||||
extra_rows: dict[str, list[list[object]]] = field(
|
||||
default_factory=lambda: {CLINICAL_PATH: [], ULTRASOUND_PATH: [], MR_PATH: []}
|
||||
)
|
||||
|
||||
def case(self, number: int) -> FakeCase:
|
||||
return next(c for c in self.cases if c.number == number)
|
||||
|
||||
def write(self, root: Path) -> Path:
|
||||
tables = {
|
||||
CLINICAL_PATH: [c.clinical for c in self.cases],
|
||||
ULTRASOUND_PATH: [c.ultrasound for c in self.cases],
|
||||
MR_PATH: [c.mr for c in self.cases],
|
||||
}
|
||||
for path, rows in tables.items():
|
||||
header = self.headers[path]
|
||||
values = [[row.get(h) for h in header] for row in rows]
|
||||
_write_workbook(
|
||||
root / path, self.sheet_names[path], [header, *values, *self.extra_rows[path]]
|
||||
)
|
||||
for case in self.cases:
|
||||
folder = root / "CEUS_agent" / case.no
|
||||
folder.mkdir(parents=True, exist_ok=True)
|
||||
for name in case.dir_files:
|
||||
if name.endswith(".png"):
|
||||
_write_png(folder / name, case.number)
|
||||
elif name.endswith(".xlsx"):
|
||||
row = [case.ultrasound.get(h) for h in ULTRASOUND_HEADER]
|
||||
rows = case.case_report_rows or [list(ULTRASOUND_HEADER), row]
|
||||
_write_workbook(folder / name, case.case_report_sheet, rows)
|
||||
else:
|
||||
(folder / name).write_bytes(b"not a real video")
|
||||
(root / "交付说明_40例_20260908.md").write_text("说明不读", encoding="utf-8")
|
||||
return root
|
||||
|
||||
|
||||
def make_delivery(count: int) -> FakeDelivery:
|
||||
"""病例号 1..count、全部能通过检查的假交付包。"""
|
||||
return FakeDelivery(cases=[make_case(n) for n in range(1, count + 1)])
|
||||
|
||||
|
||||
def _write_workbook(path: Path, sheet: str, rows: Sequence[Sequence[object]]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
workbook = openpyxl.Workbook()
|
||||
worksheet = workbook.active
|
||||
assert worksheet is not None
|
||||
worksheet.title = sheet
|
||||
for row in rows:
|
||||
# 假数据里的值都是 openpyxl 认得的单元格类型（字符串、数字、None）。
|
||||
worksheet.append(cast("list[Any]", list(row)))
|
||||
workbook.save(path)
|
||||
|
||||
|
||||
def _write_png(path: Path, number: int) -> None:
|
||||
pixels = bytes(v for i in range(24) for v in (number * 10 % 256, i * 10, 255 - i * 10))
|
||||
image = Image.frombytes("RGB", (6, 4), pixels)
|
||||
info = PngImagePlugin.PngInfo()
|
||||
info.add_text("PatientName", "should-not-survive")
|
||||
image.save(path, format="PNG", pnginfo=info)
|
||||
|
||||
|
||||
def write_config(
|
||||
tmp_path: Path, delivery_dir: Path, seed: int, code_maps: dict[str, dict[int, str]] | None
|
||||
) -> Path:
|
||||
"""在 tmp_path 写一份 build.yaml（路径都写绝对路径）；给了映射表就同时写 code_maps.yaml。"""
|
||||
maps_path = None
|
||||
if code_maps is not None:
|
||||
maps_path = tmp_path / "code_maps.yaml"
|
||||
maps_path.write_text(yaml.safe_dump(code_maps, allow_unicode=True), encoding="utf-8")
|
||||
config = {
|
||||
"delivery_dir": str(delivery_dir),
|
||||
"library_root": str(tmp_path / "library"),
|
||||
"answers_root": str(tmp_path / "answers"),
|
||||
"alias_seed": seed,
|
||||
"decode_codes": code_maps is not None,
|
||||
"code_maps": None if maps_path is None else str(maps_path),
|
||||
}
|
||||
path = tmp_path / "build.yaml"
|
||||
path.write_text(yaml.safe_dump(config, allow_unicode=True), encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def build(
|
||||
tmp_path: Path,
|
||||
delivery: FakeDelivery,
|
||||
seed: int,
|
||||
code_maps: dict[str, dict[int, str]] | None,
|
||||
now: datetime,
|
||||
) -> BuildSummary:
|
||||
"""写出假交付包和配置，跑一次构建。"""
|
||||
delivery_dir = delivery.write(tmp_path / "delivery")
|
||||
return run_build(write_config(tmp_path, delivery_dir, seed, code_maps), REPO_ROOT, now)
|
||||
@@ -0,0 +1,352 @@
|
||||
"""构建产物：可见性、泄露扫描、化名、解码开关、关键图、清单与读取、不覆盖、答案库内容。"""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import re
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from cadx.answers import read_aliases, read_diagnoses
|
||||
from cadx.build import make_aliases, run_build
|
||||
from cadx.library import (
|
||||
list_aliases,
|
||||
read_clinical,
|
||||
read_mr,
|
||||
read_ultrasound,
|
||||
scan_for_leaks,
|
||||
)
|
||||
from fake_delivery import (
|
||||
CLINICAL_HEADER,
|
||||
MR_TEXT,
|
||||
REPO_ROOT,
|
||||
SAMPLE_CODE_MAPS,
|
||||
ULTRASOUND_TEXT,
|
||||
build,
|
||||
make_delivery,
|
||||
write_config,
|
||||
)
|
||||
|
||||
NOW = datetime(2026, 10, 16, 9, 30, 0).astimezone()
|
||||
|
||||
# 独立写的泄露词表，不取自 delivery 的常量。
|
||||
SUBTYPES = (
|
||||
"IPMN",
|
||||
"SPN",
|
||||
"MCN",
|
||||
"SCN",
|
||||
"导管内乳头状黏液性肿瘤",
|
||||
"实性假乳头状瘤",
|
||||
"黏液性囊性肿瘤",
|
||||
"浆液性囊性肿瘤",
|
||||
"浆液性囊腺瘤",
|
||||
)
|
||||
DIAGNOSIS_COLUMNS = ("病理类型（1 IPMN，2 SPN，3 MCN，4 SCN）", "病理类型", "Diagnosis", "Case_ID")
|
||||
DELIVERY_FILE_NAMES = ("临床工作簿_40例.xlsx", "超声报告文本_40例.xlsx", "MR报告文本_40例.xlsx")
|
||||
FORBIDDEN_KEYS = ("diagnosis", "label", "gold", "pathology", "name", "row")
|
||||
|
||||
|
||||
def _forbidden_strings(case_nos: list[str]) -> list[str]:
|
||||
per_case = [
|
||||
term
|
||||
for no in case_nos
|
||||
for term in (
|
||||
f"P{no}",
|
||||
f"U{no}",
|
||||
f"U{no}_key.png",
|
||||
f"U{no}_roi2.mp4",
|
||||
f"U{no}_roi2_report.xlsx",
|
||||
)
|
||||
]
|
||||
return [*per_case, *DELIVERY_FILE_NAMES, *SUBTYPES, *DIAGNOSIS_COLUMNS]
|
||||
|
||||
|
||||
def _case_dirs(library_dir: Path) -> list[Path]:
|
||||
return sorted((library_dir / "cases").iterdir())
|
||||
|
||||
|
||||
def test_visibility_sex_and_age_only_in_clinical_route(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
summary = build(tmp_path, delivery, 1, SAMPLE_CODE_MAPS, NOW)
|
||||
clinical_names = [*CLINICAL_HEADER, "男", "女", "岁"]
|
||||
for case_dir in _case_dirs(summary.library_dir):
|
||||
files = sorted(p.relative_to(case_dir).as_posix() for p in case_dir.rglob("*.*"))
|
||||
assert files == [
|
||||
"clinical/clinical.json",
|
||||
"mr/report.txt",
|
||||
"ultrasound/key.png",
|
||||
"ultrasound/report.txt",
|
||||
]
|
||||
clinical = json.loads((case_dir / "clinical/clinical.json").read_text(encoding="utf-8"))
|
||||
assert set(clinical["fields"]) == set(CLINICAL_HEADER) - {"Case_ID", "Diagnosis"}
|
||||
for report in ("ultrasound/report.txt", "mr/report.txt"):
|
||||
text = (case_dir / report).read_text(encoding="utf-8")
|
||||
assert [name for name in clinical_names if name in text] == []
|
||||
ages = {f"{30 + n}" for n in range(1, 5)}
|
||||
for alias in list_aliases(summary.library_dir):
|
||||
assert str(read_clinical(summary.library_dir, alias).fields["Age"]) in ages
|
||||
assert read_ultrasound(summary.library_dir, alias).report == ULTRASOUND_TEXT
|
||||
assert read_mr(summary.library_dir, alias).report == MR_TEXT
|
||||
|
||||
|
||||
def test_leak_scan_finds_nothing_in_clean_build(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(4), 1, None, NOW)
|
||||
case_nos = [f"{n:03d}" for n in range(1, 5)]
|
||||
hits = scan_for_leaks(summary.library_dir, _forbidden_strings(case_nos), FORBIDDEN_KEYS)
|
||||
assert hits == []
|
||||
|
||||
|
||||
def test_leak_scan_detects_planted_leaks(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(3), 1, None, NOW)
|
||||
case_dir = _case_dirs(summary.library_dir)[0]
|
||||
(case_dir / "mr/report.txt").write_text(MR_TEXT + "见 P002 。", encoding="utf-8")
|
||||
(case_dir / "ultrasound/note.json").write_text('{"Label": "x"}', encoding="utf-8")
|
||||
(case_dir / "clinical/extra.json").write_text(
|
||||
'{"k": "\\u6d46\\u6db2\\u6027\\u56ca\\u6027\\u80bf\\u7624"}', encoding="utf-8"
|
||||
)
|
||||
hits = scan_for_leaks(summary.library_dir, _forbidden_strings(["002"]), FORBIDDEN_KEYS)
|
||||
found = {(Path(h.file).name, h.kind, h.term) for h in hits}
|
||||
assert found == {
|
||||
("report.txt", "字符串", "P002"),
|
||||
("note.json", "键名", "Label"),
|
||||
("extra.json", "字符串", "浆液性囊性肿瘤"),
|
||||
}
|
||||
|
||||
|
||||
def test_leak_scan_normalizes_fullwidth_and_case(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
case_dir = _case_dirs(summary.library_dir)[0]
|
||||
leaked = MR_TEXT + "形态似ｉｐｍｎ，见ｐ００２。"
|
||||
(case_dir / "mr/report.txt").write_text(leaked, encoding="utf-8")
|
||||
hits = scan_for_leaks(summary.library_dir, ["IPMN", "P002", "Ｕ００２"], FORBIDDEN_KEYS)
|
||||
assert {(Path(h.file).name, h.term) for h in hits} == {
|
||||
("report.txt", "IPMN"),
|
||||
("report.txt", "P002"),
|
||||
}
|
||||
(case_dir / "mr/report.txt").write_text(MR_TEXT + "U002", encoding="utf-8")
|
||||
hits = scan_for_leaks(summary.library_dir, ["Ｕ００２"], FORBIDDEN_KEYS)
|
||||
assert [(Path(h.file).name, h.term) for h in hits] == [("report.txt", "Ｕ００２")]
|
||||
|
||||
|
||||
def test_aliases_are_reproducible_and_well_formed(tmp_path: Path) -> None:
|
||||
first = build(tmp_path / "a", make_delivery(5), 11, None, NOW)
|
||||
second = build(tmp_path / "b", make_delivery(5), 11, None, NOW)
|
||||
other = build(tmp_path / "c", make_delivery(5), 12, None, NOW)
|
||||
aliases = list_aliases(first.library_dir)
|
||||
assert aliases == list_aliases(second.library_dir)
|
||||
assert aliases != list_aliases(other.library_dir)
|
||||
assert len(set(aliases)) == 5
|
||||
assert all(re.fullmatch(r"[a-z0-9]{4}", a) for a in aliases)
|
||||
|
||||
|
||||
def test_aliases_avoid_subtype_substrings_and_repeats() -> None:
|
||||
case_nos = [f"{n:03d}" for n in range(1, 1000)]
|
||||
aliases = make_aliases(case_nos, 3)
|
||||
assert len(set(aliases.values())) == 999
|
||||
banned = ("ipmn", "spn", "mcn", "scn")
|
||||
assert [a for a in aliases.values() if any(b in a for b in banned)] == []
|
||||
|
||||
|
||||
def test_alias_order_does_not_depend_on_input_order() -> None:
|
||||
case_nos = ["003", "001", "002"]
|
||||
assert make_aliases(case_nos, 5) == make_aliases(sorted(case_nos), 5)
|
||||
|
||||
|
||||
def test_decode_off_keeps_raw_numbers(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(3), 1, None, NOW)
|
||||
for alias in list_aliases(summary.library_dir):
|
||||
clinical = read_clinical(summary.library_dir, alias)
|
||||
assert clinical.codes_decoded is False
|
||||
assert clinical.fields["Sex"] in (1, 2)
|
||||
assert clinical.fields["Symptom_Pain"] == 1
|
||||
assert clinical.fields["CEA_Level"] == 1
|
||||
assert [d.name for d in read_diagnoses(summary.answers_dir)] == [None, None, None]
|
||||
manifest = json.loads((summary.library_dir / "manifest.json").read_text(encoding="utf-8"))
|
||||
assert manifest["codes_decoded"] is False
|
||||
assert manifest["uncompared"] == ["临床性别未比对"]
|
||||
|
||||
|
||||
def test_decode_on_uses_code_maps(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(3), 1, SAMPLE_CODE_MAPS, NOW)
|
||||
by_case = {row.case_no: row.alias for row in read_aliases(summary.answers_dir)}
|
||||
# 假交付包：病例 n 的性别编码是 1 + n % 2，诊断编码是 1 + (n - 1) % 4。
|
||||
expected_sex = {"001": "女", "002": "男", "003": "女"}
|
||||
for case_no, alias in by_case.items():
|
||||
clinical = read_clinical(summary.library_dir, alias)
|
||||
assert clinical.codes_decoded is True
|
||||
assert clinical.fields["Sex"] == expected_sex[case_no]
|
||||
assert clinical.fields["Symptom_Pain"] == "是"
|
||||
assert clinical.fields["CEA_Level"] == "异常"
|
||||
assert clinical.fields["Amylase_Level"] is None
|
||||
assert clinical.fields["Age"] == 30 + int(case_no)
|
||||
names = {d.case_no: d.name for d in read_diagnoses(summary.answers_dir)}
|
||||
assert names == {"001": "IPMN", "002": "SPN", "003": "MCN"}
|
||||
|
||||
|
||||
def test_key_image_pixels_kept_and_text_chunk_dropped(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
by_case = {row.case_no: row.alias for row in read_aliases(summary.answers_dir)}
|
||||
for case_no, alias in by_case.items():
|
||||
source = tmp_path / "delivery" / "CEUS_agent" / case_no / f"U{case_no}_key.png"
|
||||
with Image.open(source) as original:
|
||||
assert original.info.get("PatientName") == "should-not-survive"
|
||||
original_mode, original_pixels = original.mode, original.tobytes()
|
||||
copied = read_ultrasound(summary.library_dir, alias).key_image
|
||||
with Image.open(copied) as image:
|
||||
assert image.mode == original_mode
|
||||
assert image.tobytes() == original_pixels
|
||||
assert "PatientName" not in image.info
|
||||
assert _png_chunks(copied.read_bytes()) == ["IHDR", "IDAT", "IEND"]
|
||||
|
||||
|
||||
def _png_chunks(data: bytes) -> list[str]:
|
||||
"""按 PNG 格式逐块读出块类型（8 字节签名后：4 字节长度、4 字节类型、数据、4 字节 CRC）。"""
|
||||
chunks: list[str] = []
|
||||
offset = 8
|
||||
while offset < len(data):
|
||||
length = int.from_bytes(data[offset : offset + 4], "big")
|
||||
chunks.append(data[offset + 4 : offset + 8].decode("ascii"))
|
||||
offset += 12 + length
|
||||
return chunks
|
||||
|
||||
|
||||
def _build_with_key_image(tmp_path: Path, image: Image.Image) -> None:
|
||||
root = make_delivery(2).write(tmp_path / "delivery")
|
||||
image.save(root / "CEUS_agent" / "002" / "U002_key.png", format="PNG")
|
||||
run_build(write_config(tmp_path, root, 1, None), REPO_ROOT, NOW)
|
||||
|
||||
|
||||
def test_key_image_16_bit_source_stops_build(tmp_path: Path) -> None:
|
||||
image = Image.new("I;16", (4, 4))
|
||||
with pytest.raises(ValueError, match=r"U002_key\.png"):
|
||||
_build_with_key_image(tmp_path, image)
|
||||
assert not (tmp_path / "library" / f"{NOW:%Y%m%d-%H%M%S}").exists()
|
||||
assert [p for p in (tmp_path / "library").iterdir() if not p.name.endswith(".partial")] == []
|
||||
|
||||
|
||||
def test_key_image_with_transparency_stops_build(tmp_path: Path) -> None:
|
||||
image = Image.new("RGB", (4, 4))
|
||||
image.info["transparency"] = (0, 0, 0)
|
||||
with pytest.raises(ValueError, match=r"U002_key\.png"):
|
||||
_build_with_key_image(tmp_path, image)
|
||||
|
||||
|
||||
def test_key_image_palette_source_stops_build(tmp_path: Path) -> None:
|
||||
with pytest.raises(ValueError, match=r"U002_key\.png"):
|
||||
_build_with_key_image(tmp_path, Image.new("P", (4, 4)))
|
||||
|
||||
|
||||
def test_key_image_one_bit_kept_as_is(tmp_path: Path) -> None:
|
||||
source = Image.frombytes("1", (8, 2), bytes([0b10110010, 0b01001101]))
|
||||
_build_with_key_image(tmp_path, source)
|
||||
library_root = tmp_path / "library"
|
||||
(library_dir,) = list(library_root.iterdir())
|
||||
by_case = {r.case_no: r.alias for r in read_aliases(tmp_path / "answers" / library_dir.name)}
|
||||
copied = read_ultrasound(library_dir, by_case["002"]).key_image
|
||||
with Image.open(copied) as image:
|
||||
assert image.mode == "1"
|
||||
assert image.tobytes() == source.tobytes()
|
||||
assert _png_chunks(copied.read_bytes()) == ["IHDR", "IDAT", "IEND"]
|
||||
|
||||
|
||||
def test_library_read_rejects_modified_file(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
alias = list_aliases(summary.library_dir)[0]
|
||||
report = summary.library_dir / "cases" / alias / "mr" / "report.txt"
|
||||
report.write_text(MR_TEXT + "改", encoding="utf-8")
|
||||
with pytest.raises(ValueError, match="SHA-256"):
|
||||
read_mr(summary.library_dir, alias)
|
||||
|
||||
|
||||
def test_library_read_rejects_partial_directory(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
partial = summary.library_dir.with_name(summary.library_dir.name + ".partial")
|
||||
summary.library_dir.rename(partial)
|
||||
with pytest.raises(ValueError, match="partial"):
|
||||
list_aliases(partial)
|
||||
|
||||
|
||||
def test_answers_read_rejects_modified_file(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
path = summary.answers_dir / "diagnoses.csv"
|
||||
path.write_text(path.read_text(encoding="utf-8-sig") + "x\n", encoding="utf-8-sig")
|
||||
with pytest.raises(ValueError, match="SHA-256"):
|
||||
read_diagnoses(summary.answers_dir)
|
||||
|
||||
|
||||
def test_existing_build_directory_is_not_overwritten(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, None, NOW)
|
||||
manifest = (summary.library_dir / "manifest.json").read_bytes()
|
||||
with pytest.raises(FileExistsError):
|
||||
build(tmp_path, make_delivery(3), 1, None, NOW) # 同一时刻、同一提交 → 同一构建号
|
||||
assert (summary.library_dir / "manifest.json").read_bytes() == manifest
|
||||
assert len(list_aliases(summary.library_dir)) == 2
|
||||
|
||||
|
||||
def _csv_rows(path: Path) -> list[list[str]]:
|
||||
with path.open(encoding="utf-8-sig", newline="") as handle:
|
||||
return list(csv.reader(handle))
|
||||
|
||||
|
||||
def test_answers_contents(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(3).mr["年龄"] = "99岁" # C10
|
||||
delivery.case(1).ultrasound["超声描述"] = "胰体部见无回声区。考虑囊性病灶;边界清。"
|
||||
summary = build(tmp_path, delivery, 1, None, NOW)
|
||||
|
||||
assert summary.total_cases == 4
|
||||
assert summary.passed_cases == 3
|
||||
assert summary.skipped_cases == 1
|
||||
assert summary.skipped_by_check == {"C10": 1}
|
||||
assert summary.review_rows == 1
|
||||
|
||||
skipped = _csv_rows(summary.answers_dir / "skipped.csv")
|
||||
assert skipped[0] == ["病例", "检查编号", "检查说明", "文件", "列", "涉及值"]
|
||||
assert [row[:2] for row in skipped[1:]] == [["003", "C10"]]
|
||||
assert json.loads(skipped[1][5])["年龄"]["MR 表"] == "99岁"
|
||||
|
||||
review = _csv_rows(summary.answers_dir / "review.csv")
|
||||
alias_of_1 = {r.case_no: r.alias for r in read_aliases(summary.answers_dir)}["001"]
|
||||
assert review[1:] == [[alias_of_1, "超声", "考虑", "考虑囊性病灶"]]
|
||||
|
||||
assert [r.case_no for r in read_aliases(summary.answers_dir)] == ["001", "002", "004"]
|
||||
assert [r.video_id for r in read_aliases(summary.answers_dir)] == ["U001", "U002", "U004"]
|
||||
codes = [
|
||||
(d.clinical_code, d.ultrasound_code, d.mr_code) for d in read_diagnoses(summary.answers_dir)
|
||||
]
|
||||
assert codes == [(1, 1, 1), (2, 2, 2), (4, 4, 4)]
|
||||
|
||||
manifest = json.loads((summary.answers_dir / "manifest.json").read_text(encoding="utf-8"))
|
||||
assert manifest["build_id"] == summary.build_id
|
||||
assert "不代表没有问题" in manifest["skipped_note"]
|
||||
assert sorted(manifest["files"]) == [
|
||||
"aliases.csv",
|
||||
"diagnoses.csv",
|
||||
"review.csv",
|
||||
"skipped.csv",
|
||||
"snapshot/build.yaml",
|
||||
"snapshot/snapshot.json",
|
||||
]
|
||||
|
||||
|
||||
def test_snapshot_records_config_and_git(tmp_path: Path) -> None:
|
||||
summary = build(tmp_path, make_delivery(2), 1, SAMPLE_CODE_MAPS, NOW)
|
||||
snapshot_dir = summary.answers_dir / "snapshot"
|
||||
assert (snapshot_dir / "build.yaml").read_bytes() == (tmp_path / "build.yaml").read_bytes()
|
||||
assert (snapshot_dir / "code_maps.yaml").read_bytes() == (
|
||||
tmp_path / "code_maps.yaml"
|
||||
).read_bytes()
|
||||
snapshot = json.loads((snapshot_dir / "snapshot.json").read_text(encoding="utf-8"))
|
||||
assert snapshot["build_id"] == summary.build_id
|
||||
assert re.fullmatch(r"[0-9a-f]{40}", snapshot["git_commit"])
|
||||
assert summary.build_id.startswith("20261016-093000-" + snapshot["git_commit"][:7])
|
||||
assert summary.build_id.endswith("-dirty") == (snapshot["git_clean"] is False)
|
||||
assert set(snapshot["config_sha256"]) == {"build.yaml", "code_maps.yaml"}
|
||||
library_manifest = json.loads(
|
||||
(summary.library_dir / "manifest.json").read_text(encoding="utf-8")
|
||||
)
|
||||
assert library_manifest["config_sha256"] == snapshot["config_sha256"]
|
||||
assert library_manifest["ceus_source"] is None
|
||||
assert library_manifest["aliases"] == sorted(library_manifest["aliases"])
|
||||
@@ -0,0 +1,187 @@
|
||||
"""整张表的问题 T1–T5 与配置错误：整次构建停下，两个库下不留完成的构建目录。"""
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from cadx.build import load_build_config, load_code_maps, run_build
|
||||
from cadx.delivery import DeliveryError
|
||||
from fake_delivery import (
|
||||
CLINICAL_PATH,
|
||||
MR_PATH,
|
||||
REPO_ROOT,
|
||||
SAMPLE_CODE_MAPS,
|
||||
ULTRASOUND_PATH,
|
||||
make_delivery,
|
||||
write_config,
|
||||
)
|
||||
|
||||
NOW = datetime(2026, 10, 16, 9, 0, 0).astimezone()
|
||||
|
||||
|
||||
def _assert_no_finished_build(tmp_path: Path) -> None:
|
||||
for root in (tmp_path / "library", tmp_path / "answers"):
|
||||
if root.exists():
|
||||
assert all(p.name.endswith(".partial") for p in root.iterdir())
|
||||
|
||||
|
||||
def _run(tmp_path: Path, root: Path, code_maps: dict[str, dict[int, str]] | None) -> None:
|
||||
run_build(write_config(tmp_path, root, 7, code_maps), REPO_ROOT, NOW)
|
||||
|
||||
|
||||
def test_t1_missing_summary_table(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
(root / MR_PATH).unlink()
|
||||
with pytest.raises(DeliveryError, match=r"^T1"):
|
||||
_run(tmp_path, root, None)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t2_wrong_sheet_name(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.sheet_names[ULTRASOUND_PATH] = "Sheet1"
|
||||
root = delivery.write(tmp_path / "delivery")
|
||||
with pytest.raises(DeliveryError, match=r"^T2"):
|
||||
_run(tmp_path, root, None)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t3_header_differs(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.headers[MR_PATH][3] = "病理类型" # 诊断列名里写的编码含义丢了
|
||||
root = delivery.write(tmp_path / "delivery")
|
||||
with pytest.raises(DeliveryError, match=r"^T3"):
|
||||
_run(tmp_path, root, None)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t4_extra_file_in_clinical_folder(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
(root / CLINICAL_PATH).with_name("备注.txt").write_text("x", encoding="utf-8")
|
||||
with pytest.raises(DeliveryError, match=r"^T4"):
|
||||
_run(tmp_path, root, None)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t4_unexpected_entry_in_ceus_folder(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
(root / "CEUS_agent" / "41").mkdir() # 不是三位数字
|
||||
with pytest.raises(DeliveryError, match=r"^T4"):
|
||||
_run(tmp_path, root, None)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t5_code_maps_missing_a_column(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
maps = {k: v for k, v in SAMPLE_CODE_MAPS.items() if k != "TBil_Level"}
|
||||
with pytest.raises(ValueError, match=r"^T5"):
|
||||
_run(tmp_path, root, maps)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t5_code_maps_extra_column(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
maps = {**SAMPLE_CODE_MAPS, "Age": {1: "一"}}
|
||||
with pytest.raises(ValueError, match=r"^T5"):
|
||||
_run(tmp_path, root, maps)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t5_sex_must_map_to_chinese_characters(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
maps = {**SAMPLE_CODE_MAPS, "Sex": {1: "M", 2: "F"}}
|
||||
with pytest.raises(ValueError, match=r"^T5"):
|
||||
_run(tmp_path, root, maps)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def test_t5_diagnosis_must_map_to_four_subtypes(tmp_path: Path) -> None:
|
||||
root = make_delivery(3).write(tmp_path / "delivery")
|
||||
maps = {**SAMPLE_CODE_MAPS, "Diagnosis": {1: "IPMN", 2: "SPN", 3: "MCN", 4: "MCN"}}
|
||||
with pytest.raises(ValueError, match=r"^T5"):
|
||||
_run(tmp_path, root, maps)
|
||||
_assert_no_finished_build(tmp_path)
|
||||
|
||||
|
||||
def _write_yaml(tmp_path: Path, body: str) -> Path:
|
||||
path = tmp_path / "build.yaml"
|
||||
path.write_text(body, encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
_BASE = "delivery_dir: d\nlibrary_root: l\nanswers_root: a\nalias_seed: 1\n"
|
||||
|
||||
|
||||
def test_config_decode_true_needs_code_maps(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: true\ncode_maps: null\n")
|
||||
with pytest.raises(ValidationError):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_decode_false_needs_null_code_maps(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: false\ncode_maps: maps.yaml\n")
|
||||
with pytest.raises(ValidationError):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_field_missing(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: false\n")
|
||||
with pytest.raises(ValidationError):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_extra_field(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: false\ncode_maps: null\nseed: 2\n")
|
||||
with pytest.raises(ValidationError):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_seed_must_be_int_not_string(tmp_path: Path) -> None:
|
||||
body = _BASE.replace("alias_seed: 1", "alias_seed: '1'")
|
||||
path = _write_yaml(tmp_path, body + "decode_codes: false\ncode_maps: null\n")
|
||||
with pytest.raises(ValidationError):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_relative_paths_resolve_against_repo_root(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: false\ncode_maps: null\n")
|
||||
config = load_build_config(path, Path("/repo"))
|
||||
assert config.delivery_dir == Path("/repo/d")
|
||||
assert config.library_root == Path("/repo/l")
|
||||
assert config.answers_root == Path("/repo/a")
|
||||
assert config.code_maps is None
|
||||
|
||||
|
||||
def test_committed_config_is_valid() -> None:
|
||||
config = load_build_config(REPO_ROOT / "config" / "build.yaml", REPO_ROOT)
|
||||
assert config.decode_codes is False
|
||||
assert config.code_maps is None
|
||||
|
||||
|
||||
def test_config_duplicate_key(tmp_path: Path) -> None:
|
||||
body = _BASE + "decode_codes: false\ncode_maps: null\nalias_seed: 2\n"
|
||||
path = _write_yaml(tmp_path, body)
|
||||
with pytest.raises(ValueError, match="重复的键"):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_empty_path(tmp_path: Path) -> None:
|
||||
body = _BASE.replace("delivery_dir: d", "delivery_dir: ''")
|
||||
path = _write_yaml(tmp_path, body + "decode_codes: false\ncode_maps: null\n")
|
||||
with pytest.raises(ValidationError, match="空串"):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_config_empty_code_maps_path(tmp_path: Path) -> None:
|
||||
path = _write_yaml(tmp_path, _BASE + "decode_codes: true\ncode_maps: ''\n")
|
||||
with pytest.raises(ValidationError, match="空串"):
|
||||
load_build_config(path, tmp_path)
|
||||
|
||||
|
||||
def test_code_maps_duplicate_key(tmp_path: Path) -> None:
|
||||
path = tmp_path / "code_maps.yaml"
|
||||
path.write_text("Sex: {1: 男, 1: 女}\n", encoding="utf-8")
|
||||
with pytest.raises(ValueError, match="重复的键"):
|
||||
load_code_maps(path)
|
||||
@@ -0,0 +1,420 @@
|
||||
"""逐例检查 C01–C13：只让一例违反一项，这一例跳过、记对检查编号，其余病例照常通过。
|
||||
|
||||
预期值照任务书 01 第二部分「单个病例的问题」的文字写。
|
||||
"""
|
||||
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from cadx.delivery import DeliveryResult, read_delivery
|
||||
from cadx.delivery._text import conclusion_headings
|
||||
from fake_delivery import (
|
||||
CLINICAL_PATH,
|
||||
DIAGNOSIS_COLUMN,
|
||||
MR_PATH,
|
||||
MR_TEXT,
|
||||
SAMPLE_CODE_MAPS,
|
||||
ULTRASOUND_HEADER,
|
||||
ULTRASOUND_PATH,
|
||||
ULTRASOUND_TEXT,
|
||||
FakeDelivery,
|
||||
make_delivery,
|
||||
)
|
||||
|
||||
|
||||
def _read(
|
||||
tmp_path: Path, delivery: FakeDelivery, code_maps: dict[str, dict[int, str]] | None
|
||||
) -> DeliveryResult:
|
||||
return read_delivery(delivery.write(tmp_path / "delivery"), code_maps)
|
||||
|
||||
|
||||
def _assert_only_skipped(result: DeliveryResult, case: str, check: str, total: int) -> None:
|
||||
assert [(s.case, s.check) for s in result.skipped] == [(case, check)]
|
||||
passed = [f"{n:03d}" for n in range(1, total + 1) if f"{n:03d}" != case]
|
||||
assert [c.case_no for c in result.cases] == passed
|
||||
assert result.total_cases == total
|
||||
|
||||
|
||||
def test_all_pass(tmp_path: Path) -> None:
|
||||
result = _read(tmp_path, make_delivery(4), None)
|
||||
assert [c.case_no for c in result.cases] == ["001", "002", "003", "004"]
|
||||
assert result.skipped == ()
|
||||
assert result.unrecognized_rows == 0
|
||||
|
||||
|
||||
def test_c01_unrecognized_clinical_id_is_its_own_row_and_case_missing(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(2).clinical["Case_ID"] = "P2" # 临床表要求 P 加三位数字
|
||||
result = _read(tmp_path, delivery, None)
|
||||
# 第 1 行是表头，病例 2 在第 3 行；同一个病人因此在 C03（缺在临床表）再记一条。
|
||||
assert [(s.case, s.check) for s in result.skipped] == [
|
||||
(f"{CLINICAL_PATH} 第 3 行", "C01"),
|
||||
("002", "C03"),
|
||||
]
|
||||
assert result.unrecognized_rows == 1
|
||||
assert result.total_cases == 4
|
||||
assert [c.case_no for c in result.cases] == ["001", "003", "004"]
|
||||
|
||||
|
||||
def test_c01_float_report_id_is_unrecognized(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(3).mr["患者ID"] = 3.5 # 浮点不是认得出的写法
|
||||
result = _read(tmp_path, delivery, None)
|
||||
assert [(s.case, s.check) for s in result.skipped] == [
|
||||
(f"{MR_PATH} 第 4 行", "C01"),
|
||||
("003", "C03"),
|
||||
]
|
||||
|
||||
|
||||
def test_c01_report_ids_accept_int_and_zero_padded_strings(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["患者ID"] = 1
|
||||
delivery.case(2).mr["患者ID"] = "002"
|
||||
delivery.case(2).ultrasound["患者ID"] = "02"
|
||||
result = _read(tmp_path, delivery, None)
|
||||
assert result.skipped == ()
|
||||
assert len(result.cases) == 3
|
||||
|
||||
|
||||
def test_c02_duplicate_id_in_one_table(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
duplicate = delivery.case(2).mr
|
||||
delivery.extra_rows[MR_PATH].append([duplicate[h] for h in delivery.headers[MR_PATH]])
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C02", 4)
|
||||
|
||||
|
||||
def test_c03_case_directory_missing(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
root = delivery.write(tmp_path / "delivery")
|
||||
shutil.rmtree(root / "CEUS_agent" / "003")
|
||||
_assert_only_skipped(read_delivery(root, None), "003", "C03", 4)
|
||||
|
||||
|
||||
def test_c04_missing_key_image(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(4).dir_files.remove("U004_key.png")
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C04", 4)
|
||||
|
||||
|
||||
def test_c04_extra_file(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).dir_files.append("U001.mp4") # 两个视频共四个文件
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C04", 3)
|
||||
|
||||
|
||||
def test_c04_roi2_video_with_roi2_report_passes(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
assert delivery.case(2).dir_files == ["U002_key.png", "U002_roi2.mp4", "U002_roi2_report.xlsx"]
|
||||
assert _read(tmp_path, delivery, None).skipped == ()
|
||||
|
||||
|
||||
def test_c04_plain_video_with_plain_report_passes(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(3).dir_files = ["U003_key.png", "U003.mp4", "U003_report.xlsx"]
|
||||
result = _read(tmp_path, delivery, None)
|
||||
assert result.skipped == ()
|
||||
assert len(result.cases) == 3
|
||||
|
||||
|
||||
def test_c04_roi2_video_with_plain_report_is_skipped(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
# 文件名被 gitleaks 的通用密钥规则误报，不是密钥。
|
||||
delivery.case(1).dir_files = [
|
||||
"U001_key.png",
|
||||
"U001_roi2.mp4", # gitleaks:allow
|
||||
"U001_report.xlsx",
|
||||
]
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C04", 3)
|
||||
|
||||
|
||||
def test_c04_plain_video_with_roi2_report_is_skipped(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).dir_files = ["U002_key.png", "U002.mp4", "U002_roi2_report.xlsx"]
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C04", 3)
|
||||
|
||||
|
||||
def test_c06_reads_plain_report_when_paired_with_plain_video(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
case = delivery.case(3)
|
||||
case.dir_files = ["U003_key.png", "U003.mp4", "U003_report.xlsx"]
|
||||
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
|
||||
row[-1] = ULTRASOUND_TEXT + "补充一句。"
|
||||
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C06", 3)
|
||||
|
||||
|
||||
def test_c05_video_id_mismatch(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(2).ultrasound["CEUS视频ID"] = "U003"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C05", 4)
|
||||
|
||||
|
||||
def test_c06_case_report_cell_differs(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
case = delivery.case(3)
|
||||
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
|
||||
row[-1] = ULTRASOUND_TEXT + "补充一句。"
|
||||
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "003", "C06", 4)
|
||||
assert result.skipped[0].columns == ("超声描述",)
|
||||
|
||||
|
||||
def test_c06_case_report_cell_type_differs(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
case = delivery.case(1)
|
||||
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
|
||||
row[0] = 1 # 汇总表里是字符串 "1"
|
||||
case.case_report_rows = [list(ULTRASOUND_HEADER), row]
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C06", 3)
|
||||
|
||||
|
||||
def test_c06_case_report_wrong_sheet(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).case_report_sheet = "Sheet2"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C06", 3)
|
||||
|
||||
|
||||
def test_c06_case_report_two_data_rows(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
case = delivery.case(2)
|
||||
row = [case.ultrasound[h] for h in ULTRASOUND_HEADER]
|
||||
case.case_report_rows = [list(ULTRASOUND_HEADER), row, row]
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C06", 3)
|
||||
|
||||
|
||||
def test_c07_bool_sex_is_not_an_int(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).clinical["Sex"] = True # bool 是 int 的子类，不能当作 1
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C07", 3)
|
||||
|
||||
|
||||
def test_c01_bool_report_id_is_unrecognized(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["患者ID"] = True
|
||||
result = _read(tmp_path, delivery, None)
|
||||
assert [(s.case, s.check) for s in result.skipped] == [
|
||||
(f"{ULTRASOUND_PATH} 第 2 行", "C01"),
|
||||
("001", "C03"),
|
||||
]
|
||||
|
||||
|
||||
def test_c07_sex_out_of_range(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(1).clinical["Sex"] = 3
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C07", 4)
|
||||
|
||||
|
||||
def test_c07_non_integer_age(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).clinical["Age"] = 32.5
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C07", 3)
|
||||
|
||||
|
||||
def test_c07_symptom_empty(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).clinical["Symptom_Fever"] = None
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C07", 3)
|
||||
|
||||
|
||||
def test_c07_lab_only_one_column_empty(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(2).clinical["CA199"] = None # 档位列仍是 0
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "002", "C07", 4)
|
||||
assert set(result.skipped[0].columns) == {"CA199", "CA199_Level"}
|
||||
|
||||
|
||||
def test_c07_lab_both_empty_is_fine(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).clinical["TBil"] = None
|
||||
delivery.case(1).clinical["TBil_Level"] = None
|
||||
assert _read(tmp_path, delivery, None).skipped == ()
|
||||
|
||||
|
||||
def test_c07_value_missing_from_code_map(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
maps = {**SAMPLE_CODE_MAPS, "Hx_Alcohol": {0: "否"}} # 映射表里没有 1
|
||||
delivery.case(2).clinical["Hx_Alcohol"] = 1
|
||||
_assert_only_skipped(_read(tmp_path, delivery, maps), "002", "C07", 3)
|
||||
|
||||
|
||||
def test_c08_ultrasound_sex_invalid(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(4).ultrasound["性别"] = "M"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C08", 4)
|
||||
|
||||
|
||||
def test_c08_negative_bolus_time(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(3).ultrasound["推注开始时间s"] = -1
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C08", 3)
|
||||
|
||||
|
||||
def test_c08_empty_description(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["超声描述"] = None
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C08", 3)
|
||||
|
||||
|
||||
def test_c09_mr_age_not_number_plus_sui(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(3).mr["年龄"] = "33"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C09", 4)
|
||||
|
||||
|
||||
def test_c09_mr_diagnosis_out_of_range(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).mr[DIAGNOSIS_COLUMN] = 5
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C09", 3)
|
||||
|
||||
|
||||
def test_c10_diagnosis_mismatch(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(1).clinical["Diagnosis"] = 2 # 两张报告表仍是 1
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "001", "C10", 4)
|
||||
assert result.skipped[0].columns == ("诊断",)
|
||||
|
||||
|
||||
def test_c10_age_mismatch(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(2).mr["年龄"] = "99岁"
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "002", "C10", 4)
|
||||
assert result.skipped[0].columns == ("年龄",)
|
||||
|
||||
|
||||
def test_c10_sex_mismatch_between_reports(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(3).mr["性别"] = "男" if delivery.case(3).ultrasound["性别"] == "女" else "女"
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "003", "C10", 4)
|
||||
assert result.skipped[0].columns == ("性别",)
|
||||
|
||||
|
||||
def test_c10_clinical_sex_compared_only_when_decoding(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
case = delivery.case(2)
|
||||
case.clinical["Sex"] = 1 if case.clinical["Sex"] == 2 else 2 # 报告表的性别不变
|
||||
undecoded = _read(tmp_path / "a", delivery, None)
|
||||
assert undecoded.skipped == ()
|
||||
assert undecoded.uncompared == ("临床性别未比对",)
|
||||
decoded = _read(tmp_path / "b", delivery, SAMPLE_CODE_MAPS)
|
||||
_assert_only_skipped(decoded, "002", "C10", 3)
|
||||
assert decoded.uncompared == ()
|
||||
|
||||
|
||||
def test_c11_heading_at_start_with_fullwidth_colon(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(2).ultrasound["超声描述"] = "结论：胰体部囊性病灶。" + ULTRASOUND_TEXT
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C11", 4)
|
||||
|
||||
|
||||
def test_c11_heading_after_newline(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(3).mr["MR描述"] = MR_TEXT + "\n印象 : 胰体部囊性病灶。"
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "003", "C11", 4)
|
||||
assert result.skipped[0].columns == ("MR描述",)
|
||||
|
||||
|
||||
def test_c11_heading_mid_text_is_not_a_failure(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["超声描述"] = "胰体部见一无回声区。造影提示：囊壁呈等增强。"
|
||||
result = _read(tmp_path, delivery, None)
|
||||
assert result.skipped == ()
|
||||
assert len(result.cases) == 3
|
||||
|
||||
|
||||
def test_c12_subtype_name_fullwidth_lowercase(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(4).mr["MR描述"] = MR_TEXT + "形态类似ｉｐｍｎ。"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "004", "C12", 4)
|
||||
|
||||
|
||||
def test_c12_chinese_subtype_name(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["超声描述"] = ULTRASOUND_TEXT + "符合浆液性囊腺瘤表现。"
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C12", 3)
|
||||
|
||||
|
||||
def test_c11_heading_after_vertical_tab_or_form_feed() -> None:
|
||||
# xlsx（XML）里写不进 \x0b、\x0c，openpyxl 拒绝生成，所以直接测规则函数。
|
||||
for separator in ("\x0b", "\x0c"):
|
||||
text = MR_TEXT + separator + "诊断：胰体部囊性病灶。"
|
||||
assert [k for k, _ in conclusion_headings(text)] == ["诊断"]
|
||||
|
||||
|
||||
def test_c11_heading_after_unicode_line_separator(tmp_path: Path) -> None:
|
||||
for i, separator in enumerate(("\x85", "\u2028", "\u2029")):
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).mr["MR描述"] = MR_TEXT + separator + "诊断：胰体部囊性病灶。"
|
||||
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "002", "C11", 3)
|
||||
|
||||
|
||||
def test_c11_ratio_sign_as_colon(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["超声描述"] = "超声提示∶胰体部囊性病灶。" + ULTRASOUND_TEXT
|
||||
result = _read(tmp_path, delivery, None)
|
||||
_assert_only_skipped(result, "001", "C11", 3)
|
||||
assert '"关键词": "超声提示"' in result.skipped[0].value_json
|
||||
|
||||
|
||||
def test_c12_new_subtype_terms(tmp_path: Path) -> None:
|
||||
terms = (
|
||||
"SPT",
|
||||
"ｓｐｔｐ",
|
||||
"IPMT",
|
||||
"导管内乳头",
|
||||
"实性假乳头",
|
||||
"黏液囊性",
|
||||
"粘液囊性",
|
||||
"囊腺瘤",
|
||||
"浆液性微囊",
|
||||
)
|
||||
for i, term in enumerate(terms):
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(3).ultrasound["超声描述"] = ULTRASOUND_TEXT + f"形态与{term}相似。"
|
||||
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "003", "C12", 3)
|
||||
|
||||
|
||||
def test_c12_serous_alone_is_fine(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(1).ultrasound["超声描述"] = ULTRASOUND_TEXT + "囊液呈浆液性。"
|
||||
assert _read(tmp_path, delivery, None).skipped == ()
|
||||
|
||||
|
||||
def test_c13_age_in_report(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(4)
|
||||
delivery.case(1).ultrasound["超声描述"] = "患者４５岁。" + ULTRASOUND_TEXT
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "001", "C13", 4)
|
||||
|
||||
|
||||
def test_c13_age_variants(tmp_path: Path) -> None:
|
||||
for i, age in enumerate(("四十五岁", "45 岁", "6周岁", "三十歲", "１２周岁")):
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).mr["MR描述"] = f"患者{age}。" + MR_TEXT
|
||||
_assert_only_skipped(_read(tmp_path / str(i), delivery, None), "002", "C13", 3)
|
||||
|
||||
|
||||
def test_c13_sex_character_in_report(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(3).mr["MR描述"] = "女性患者。" + MR_TEXT
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "003", "C13", 3)
|
||||
|
||||
|
||||
def test_only_first_failing_check_is_recorded(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).ultrasound["性别"] = "M" # C08
|
||||
delivery.case(2).mr["年龄"] = "99岁" # C10
|
||||
_assert_only_skipped(_read(tmp_path, delivery, None), "002", "C08", 3)
|
||||
|
||||
|
||||
def test_skip_record_names_file_and_column(tmp_path: Path) -> None:
|
||||
delivery = make_delivery(3)
|
||||
delivery.case(2).ultrasound["推注开始时间s"] = 2.5
|
||||
record = _read(tmp_path, delivery, None).skipped[0]
|
||||
assert record.files == (ULTRASOUND_PATH,)
|
||||
assert record.columns == ("推注开始时间s",)
|
||||
assert record.value_json == '{"推注开始时间s": 2.5}'
|
||||
Reference in New Issue
Block a user