"""tias-dataset CLI のテスト. 仕様: SPEC_01「CLI」「実験の出力」「データセット集計の出力」「回収ホワイトリスト」． - build-manifest は既存の manifest があれば --overwrite 無しで止まり，ありなら上書きする - survey は runs/<実験ID>/ に dataset_summary.json・config.yaml・log.txt・figures/*.png を書く - 実験 ID は `YYYYmmdd-HHMMSS_survey_`（TIAS_GIT_COMMIT が無ければ nogit） - 出力（ファイル・ログ・標準出力）に患者 ID・ファイル名を出さない - 書き出すファイルはすべて回収ホワイトリストの対象（手元へ持ち帰れる集計値だけ） データは scripts/make_synthetic_dataset.py で tmp_path に作る合成データ（臨床データは使わない）． """ from __future__ import annotations import importlib.util import json import logging import re import sys from datetime import datetime from pathlib import Path from types import ModuleType import pytest import yaml from PIL import Image import tias_judge_model.cli.dataset as cli from tias_judge_model.data.manifest import read_manifest from tias_judge_model.remote.whitelist import is_pull_allowed SCRIPT_PATH = Path(__file__).resolve().parents[2] / "scripts" / "make_synthetic_dataset.py" SECRET = "SECRETPT" RUN_ID_RE = re.compile(r"^\d{8}-\d{6}_survey_(?P.+)$") @pytest.fixture(scope="module") def script() -> ModuleType: spec = importlib.util.spec_from_file_location("make_synthetic_dataset_cli", SCRIPT_PATH) assert spec is not None and spec.loader is not None module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module @pytest.fixture def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, script: ModuleType) -> Path: """合成データと設定 YAML を作り，作業ディレクトリに移る. ファイル名（＝患者 ID）には目印の文字列を含める． Returns: 設定 YAML のパス． """ data_root = tmp_path / "dataset" script.generate(data_root, per_dir=6, seed=0) (data_root / "manifest.csv").unlink() # ファイル名（= 患者 ID）に目印を入れ，出力に紛れ込んでいないか検索できるようにする for n, png in enumerate(sorted((data_root / "images").rglob("*.png"))): png.rename(png.with_name(f"{SECRET}{n:04d}.png")) # 読めない画像を 1 枚混ぜる（集計は止めずに件数として数える） (data_root / "images" / "red_mild" / f"{SECRET}BROKEN.png").write_bytes(b"not a png") config = tmp_path / "survey.yaml" config.write_text( f"data:\n root: {data_root.as_posix()}\n" "split:\n seed: 42\n val_ratio: 0.15\n test_ratio: 0.15\n n_folds: 5\n", encoding="utf-8", ) work = tmp_path / "work" work.mkdir() monkeypatch.chdir(work) for var in ("TIAS_DATA_ROOT", "TIAS_GIT_COMMIT"): monkeypatch.delenv(var, raising=False) return config def _manifest_path(config: Path) -> Path: return config.parent / "dataset" / "manifest.csv" def _run_dirs(root: Path) -> list[Path]: runs = root / "runs" return sorted(p for p in runs.iterdir()) if runs.exists() else [] # --------------------------------------------------------------------------- # build-manifest # --------------------------------------------------------------------------- class TestBuildManifest: def test_writes_manifest(self, workspace: Path) -> None: assert cli.main(["build-manifest", "--config", str(workspace)]) == 0 records = read_manifest(_manifest_path(workspace)) assert len(records) == 9 * 6 + 1 def test_existing_manifest_requires_overwrite(self, workspace: Path) -> None: manifest = _manifest_path(workspace) assert cli.main(["build-manifest", "--config", str(workspace)]) == 0 original = manifest.read_bytes() # 画像を 1 枚足してから再実行 → --overwrite 無しでは止まり，manifest は変わらない extra = manifest.parent / "images" / "pale_mild" / f"{SECRET}EXTRA.png" Image.new("RGB", (8, 8), (200, 180, 180)).save(extra) assert cli.main(["build-manifest", "--config", str(workspace)]) != 0 assert manifest.read_bytes() == original # --overwrite で上書きされる assert cli.main(["build-manifest", "--config", str(workspace), "--overwrite"]) == 0 assert len(read_manifest(manifest)) == 9 * 6 + 2 # --------------------------------------------------------------------------- # survey # --------------------------------------------------------------------------- def _survey( workspace: Path, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> tuple[int, Path | None, str]: """manifest を作ってから survey を実行し，(終了コード, 実験ディレクトリ, 出力全体) を返す.""" # log.txt に INFO が書かれるよう，ルートロガーの水準を INFO にする caplog.set_level(logging.INFO) assert cli.main(["build-manifest", "--config", str(workspace)]) == 0 rc = cli.main(["survey", "--config", str(workspace)]) captured = capsys.readouterr() dirs = _run_dirs(Path.cwd()) run_dir = dirs[0] if len(dirs) == 1 else None return rc, run_dir, captured.out + captured.err + caplog.text class TestSurvey: def test_writes_summary_files_under_run_id( self, workspace: Path, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> None: caplog.set_level(logging.INFO) assert cli.main(["build-manifest", "--config", str(workspace)]) == 0 capsys.readouterr() assert cli.main(["survey", "--config", str(workspace)]) == 0 out = capsys.readouterr().out dirs = _run_dirs(Path.cwd()) assert len(dirs) == 1 run_dir = dirs[0] match = RUN_ID_RE.match(run_dir.name) assert match is not None, run_dir.name assert match["git"] == "nogit" # TIAS_GIT_COMMIT が無い # 実験 ID を標準出力へ出す（tias-remote pull に使う） assert out.strip().splitlines()[-1] == run_dir.name assert (run_dir / "dataset_summary.json").is_file() assert (run_dir / "config.yaml").is_file() assert (run_dir / "log.txt").is_file() assert (run_dir / "log.txt").read_text(encoding="utf-8").strip() figures = sorted((run_dir / "figures").glob("*.png")) assert figures for fig in figures: with Image.open(fig) as img: assert img.format == "PNG" summary = json.loads((run_dir / "dataset_summary.json").read_text(encoding="utf-8")) assert summary["counts"]["total"]["images"] == 9 * 6 + 1 assert summary["loading"]["unreadable"] == 1 assert len(summary["delta_e"]["mild_comparison"]) == 4 def test_all_outputs_are_pull_allowed( self, workspace: Path, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> None: rc, run_dir, _ = _survey(workspace, caplog, capsys) assert rc == 0 and run_dir is not None files = [p.relative_to(run_dir).as_posix() for p in run_dir.rglob("*") if p.is_file()] assert "dataset_summary.json" in files for name in files: assert is_pull_allowed(name, with_checkpoints=False), name def test_no_patient_id_or_filename_in_outputs( self, workspace: Path, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> None: rc, run_dir, console = _survey(workspace, caplog, capsys) assert rc == 0 and run_dir is not None assert SECRET not in console for path in run_dir.rglob("*"): if path.is_file() and path.suffix != ".png": text = path.read_text(encoding="utf-8") assert SECRET not in text, path.name assert ".png" not in text, path.name assert SECRET not in path.name def test_git_commit_from_environment( self, workspace: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> None: monkeypatch.setenv("TIAS_GIT_COMMIT", "abc1234-dirty") rc, run_dir, _ = _survey(workspace, caplog, capsys) assert rc == 0 and run_dir is not None match = RUN_ID_RE.match(run_dir.name) assert match is not None and match["git"] == "abc1234-dirty" # config.yaml に git のハッシュと seed が記録される document = yaml.safe_load((run_dir / "config.yaml").read_text(encoding="utf-8")) text = json.dumps(document, ensure_ascii=False) assert "abc1234-dirty" in text assert document["config"]["split"]["seed"] == 42 @pytest.mark.parametrize("value", ["../evil", "a b", "x;rm -rf ~", "abc/def", "-dirty"]) def test_invalid_git_commit_is_rejected( self, workspace: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], value: str, ) -> None: monkeypatch.setenv("TIAS_GIT_COMMIT", value) rc, _run_dir, _ = _survey(workspace, caplog, capsys) assert rc != 0 assert _run_dirs(Path.cwd()) == [] assert not (Path.cwd().parent / "evil").exists() def test_severities_setting_is_recorded( self, workspace: Path, caplog: pytest.LogCaptureFixture, capsys: pytest.CaptureFixture[str], ) -> None: workspace.write_text( workspace.read_text(encoding="utf-8").replace( "data:\n", "data:\n severities: [strong, normal]\n", 1 ), encoding="utf-8", ) rc, run_dir, _ = _survey(workspace, caplog, capsys) assert rc == 0 and run_dir is not None summary = json.loads((run_dir / "dataset_summary.json").read_text(encoding="utf-8")) assert summary["configured_severities"] == ["strong", "normal"] # 集計は設定に関わらず全データで行い，条件ごとの件数を並べる assert summary["counts"]["total"]["images"] == 9 * 6 + 1 conditions = summary["counts"]["conditions"] assert conditions["with_mild"]["images"] == 9 * 6 + 1 assert conditions["without_mild"]["images"] == 5 * 6 def test_missing_manifest_is_error( self, workspace: Path, caplog: pytest.LogCaptureFixture ) -> None: caplog.set_level(logging.INFO) assert cli.main(["survey", "--config", str(workspace)]) != 0 # --------------------------------------------------------------------------- # 実験 ID # --------------------------------------------------------------------------- class TestRunId: def test_format(self) -> None: now = datetime(2026, 9, 27, 8, 5, 3) assert cli.make_run_id(now, "survey", "abc1234") == "20260927-080503_survey_abc1234" @pytest.mark.parametrize("environ", [{}, {"TIAS_GIT_COMMIT": ""}, {"OTHER": "x"}]) def test_nogit_when_absent(self, environ: dict[str, str]) -> None: assert cli.resolve_git_commit(environ) == "nogit" @pytest.mark.parametrize("value", ["abc1234", "abc1234-dirty", "0123456789abcdef"]) def test_valid_git_commit(self, value: str) -> None: assert cli.resolve_git_commit({"TIAS_GIT_COMMIT": value}) == value @pytest.mark.parametrize("value", ["../x", "a b", "a/b", "$(id)", "-x"]) def test_invalid_git_commit(self, value: str) -> None: with pytest.raises(ValueError): cli.resolve_git_commit({"TIAS_GIT_COMMIT": value})