"""回収ホワイトリストの判定のテスト（SPEC_01「回収ホワイトリスト」・SPEC_03「pull」）. 最重要: 名前が一致するファイルだけが回収対象になり，分割結果・個票の予測・画像・manifest 等は 回収されないこと．testing ルール必須対象 4． """ import io import tarfile import pytest from tias_judge_model.remote.whitelist import ( check_pull_members, is_pull_allowed, parse_find_listing, plan_pull, validate_run_id, ) RUN_ID = "20260927-120000_newton_abc1234" ROOT_FILES = ["config.yaml", "history.csv", "metrics.json", "confusion_matrix.csv", "log.txt"] # 列挙外（回収してはならない）ファイル．患者 ID・画像名を含みうる名前も混ぜる NOT_ALLOWED = [ "split.json", "predictions.csv", "per_image_predictions.csv", "manifest.csv", "confusion_matrix.png", # figures/ 以外にある *.png "P0001_tongue.png", "P0001_tongue.jpg", "images/pale_1/P0001.png", "misclassified/P0001.png", "data/manifest.csv", "config.yml", "config.yaml.bak", "metrics.json.tmp", "sub/config.yaml", "figures/config.yaml", "figures/cm.jpg", "figures/sub/cm.png", "checkpoints/best.pth", "checkpoints/sub/best.pt", "best.pt", ] class TestIsPullAllowed: @pytest.mark.parametrize("name", ROOT_FILES) def test_root_files_are_allowed(self, name: str) -> None: assert is_pull_allowed(name, with_checkpoints=False) @pytest.mark.parametrize("name", ["figures/confusion_matrix.png", "figures/learning_curve.png"]) def test_figures_png_is_allowed(self, name: str) -> None: assert is_pull_allowed(name, with_checkpoints=False) @pytest.mark.parametrize("name", NOT_ALLOWED) def test_names_outside_whitelist_are_not_allowed(self, name: str) -> None: assert not is_pull_allowed(name, with_checkpoints=False) # --with-checkpoints は checkpoints/*.pt 以外の判定を変えない assert not is_pull_allowed(name, with_checkpoints=True) def test_checkpoints_only_with_flag(self) -> None: assert not is_pull_allowed("checkpoints/best.pt", with_checkpoints=False) assert is_pull_allowed("checkpoints/best.pt", with_checkpoints=True) @pytest.mark.parametrize( "name", [ "../config.yaml", "figures/../config.yaml", "/config.yaml", "C:config.yaml", "C:/config.yaml", "figures\\cm.png", "..\\config.yaml", ], ) def test_unsafe_paths_are_not_allowed(self, name: str) -> None: assert not is_pull_allowed(name, with_checkpoints=True) class TestValidateRunId: def test_accepts_normal_run_id(self) -> None: assert validate_run_id(RUN_ID) == RUN_ID @pytest.mark.parametrize( "run_id", ["a/b", "a\\b", "..", "../x", "x/..", "a..b", "/abs", "runs/x", ""] ) def test_rejects_run_id_with_separator_or_dotdot(self, run_id: str) -> None: with pytest.raises(ValueError): validate_run_id(run_id) class TestPlanPull: def _listing(self, names: list[str], kind: str = "f") -> list[tuple[str, str]]: return [(kind, f"{RUN_ID}/{name}") for name in names] def test_selects_only_whitelisted_files(self) -> None: allowed = [*ROOT_FILES, "figures/cm.png"] listing = self._listing(allowed + NOT_ALLOWED + ["checkpoints/best.pt"]) plan = plan_pull(listing, RUN_ID, with_checkpoints=False) assert sorted(plan.files) == sorted(f"{RUN_ID}/{name}" for name in allowed) assert plan.skipped == len(NOT_ALLOWED) + 1 def test_with_checkpoints_includes_pt(self) -> None: listing = self._listing(["config.yaml", "checkpoints/best.pt", "split.json"]) plan = plan_pull(listing, RUN_ID, with_checkpoints=True) assert sorted(plan.files) == [f"{RUN_ID}/checkpoints/best.pt", f"{RUN_ID}/config.yaml"] assert plan.skipped == 1 def test_links_are_not_selected_even_with_whitelisted_name(self) -> None: listing = self._listing(["config.yaml"], kind="l") plan = plan_pull(listing, RUN_ID, with_checkpoints=False) assert plan.files == () assert plan.skipped == 1 def test_files_of_other_run_are_not_selected(self) -> None: plan = plan_pull([("f", "other_run/config.yaml")], RUN_ID, with_checkpoints=False) assert plan.files == () def test_parse_find_listing_round_trip(self) -> None: data = f"f {RUN_ID}/config.yaml\0f {RUN_ID}/figures/a b.png\0l {RUN_ID}/x\0".encode() assert parse_find_listing(data) == [ ("f", f"{RUN_ID}/config.yaml"), ("f", f"{RUN_ID}/figures/a b.png"), ("l", f"{RUN_ID}/x"), ] def _regular(name: str) -> tarfile.TarInfo: info = tarfile.TarInfo(name) info.size = 0 return info def _link(name: str, target: str, kind: bytes) -> tarfile.TarInfo: info = tarfile.TarInfo(name) info.type = kind info.linkname = target return info class TestCheckPullMembers: def test_accepts_only_whitelisted_regular_files(self) -> None: members = [_regular(f"{RUN_ID}/{name}") for name in [*ROOT_FILES, "figures/cm.png"]] assert check_pull_members(members, RUN_ID, with_checkpoints=False) == len(members) @pytest.mark.parametrize( "bad", [ *(_regular(f"{RUN_ID}/{name}") for name in NOT_ALLOWED), _regular(f"{RUN_ID}/checkpoints/best.pt"), _regular("other_run/config.yaml"), _regular("config.yaml"), _regular(f"{RUN_ID}/../config.yaml"), _regular(f"/{RUN_ID}/config.yaml"), _regular(f"C:/{RUN_ID}/config.yaml"), _regular(f"{RUN_ID}\\config.yaml"), _link(f"{RUN_ID}/config.yaml", "/etc/passwd", tarfile.SYMTYPE), _link(f"{RUN_ID}/metrics.json", f"{RUN_ID}/config.yaml", tarfile.LNKTYPE), ], ) def test_rejects_whole_archive_if_any_member_is_bad(self, bad: tarfile.TarInfo) -> None: members = [_regular(f"{RUN_ID}/config.yaml"), bad, _regular(f"{RUN_ID}/log.txt")] with pytest.raises(ValueError): check_pull_members(members, RUN_ID, with_checkpoints=False) def test_error_message_does_not_contain_member_names(self) -> None: secret = "P0001_yamada_tongue" members = [ _regular(f"{RUN_ID}/{secret}.png"), _link(f"{RUN_ID}/{secret}_link", "/etc/passwd", tarfile.SYMTYPE), _regular(f"../{secret}.csv"), ] with pytest.raises(ValueError) as excinfo: check_pull_members(members, RUN_ID, with_checkpoints=False) assert secret not in str(excinfo.value) def test_checkpoints_accepted_only_with_flag(self) -> None: members = [_regular(f"{RUN_ID}/checkpoints/best.pt")] assert check_pull_members(members, RUN_ID, with_checkpoints=True) == 1 with pytest.raises(ValueError): check_pull_members(members, RUN_ID, with_checkpoints=False) def test_members_from_real_tar_file(self) -> None: buffer = io.BytesIO() with tarfile.open(fileobj=buffer, mode="w") as tar: tar.addfile(_regular(f"{RUN_ID}/config.yaml"), io.BytesIO(b"")) tar.addfile(_link(f"{RUN_ID}/log.txt", "/etc/shadow", tarfile.SYMTYPE)) buffer.seek(0) with tarfile.open(fileobj=buffer, mode="r") as tar, pytest.raises(ValueError): check_pull_members(tar.getmembers(), RUN_ID, with_checkpoints=False)