"""构建产物：可见性、泄露扫描、化名、解码开关、关键图、清单与读取、不覆盖、答案库内容。""" 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 ( LibraryCase, list_aliases, read_clinical, read_mr, read_ultrasound, scan_for_leaks, write_library, ) 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 _assert_key_image_skipped(tmp_path: Path, reason: str) -> None: """002 这一例记 C14 跳过、涉及值是原因；001 照常进库。""" (library_dir,) = list((tmp_path / "library").iterdir()) answers_dir = tmp_path / "answers" / library_dir.name assert [r.case_no for r in read_aliases(answers_dir)] == ["001"] assert len(list_aliases(library_dir)) == 1 skipped = _csv_rows(answers_dir / "skipped.csv") assert [row[:4] for row in skipped[1:]] == [ ["002", "C14", "关键图能原样重编码", "CEUS_agent/002/U002_key.png"] ] assert json.loads(skipped[1][5]) == reason def test_key_image_16_bit_source_skips_case(tmp_path: Path) -> None: _build_with_key_image(tmp_path, Image.new("I;16", (4, 4))) _assert_key_image_skipped(tmp_path, "位深 16、颜色类型 0 不能逐像素保真重编码") def test_key_image_with_transparency_skips_case(tmp_path: Path) -> None: image = Image.new("RGB", (4, 4)) image.info["transparency"] = (0, 0, 0) _build_with_key_image(tmp_path, image) _assert_key_image_skipped(tmp_path, "带 tRNS 透明色，重编码会丢透明") def test_key_image_palette_source_skips_case(tmp_path: Path) -> None: _build_with_key_image(tmp_path, Image.new("P", (4, 4))) # 只有一种颜色的调色板图，Pillow 存成 1 位。 _assert_key_image_skipped(tmp_path, "位深 1、颜色类型 3 不能逐像素保真重编码") def test_key_image_not_png_skips_case(tmp_path: Path) -> None: root = make_delivery(2).write(tmp_path / "delivery") Image.new("RGB", (4, 4)).save(root / "CEUS_agent" / "002" / "U002_key.png", format="JPEG") run_build(write_config(tmp_path, root, 1, None), REPO_ROOT, NOW) _assert_key_image_skipped(tmp_path, "不是 PNG") def test_write_library_rejects_unchecked_key_image(tmp_path: Path) -> None: """build 漏了 C14 检查时，library 写图那一步报错停下。""" source = tmp_path / "key.png" Image.new("P", (4, 4)).save(source, format="PNG") target = tmp_path / "out" target.mkdir() case = LibraryCase( alias="abcd", clinical_fields={}, ultrasound_report="", key_image_source=source, mr_report="", ) with pytest.raises(ValueError, match="check_key_image"): write_library(target, "b", "c", True, {}, False, (), None, [case]) 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"])