"""CLI（SPEC_03 の `tiasdata init`・「出力の規則」，error-handling.md の終了コード）."""
import hashlib
import os
import re
import sqlite3
import stat
import subprocess
import sys
from pathlib import Path
import pytest
from conftest import (
TONGUE_COLOR_FOLDERS,
ImportEnv,
write_dummy_png,
write_patient_map,
write_reviews,
)
from tiasdata import catalog
from tiasdata.cli import main
def _write_config(tmp_path: Path, extra: str = "") -> Path:
# 相対パスで書き，設定ファイルのあるフォルダ基準で解決させる
config_path = tmp_path / "conf" / "config.toml"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
'[catalog]\nroot = "../data/tiasdata"\nidentity = "../data/identity/identity.sqlite"\n'
+ extra,
encoding="utf-8",
)
return config_path
def _paths(tmp_path: Path) -> dict[str, Path]:
root = tmp_path / "data" / "tiasdata"
return {
"root": root,
"catalog": root / "catalog.sqlite",
"store": root / "store",
"bundles": root / "bundles",
"backups": root / "backups",
"identity": tmp_path / "data" / "identity" / "identity.sqlite",
}
def _assert_no_paths(text: str, tmp_path: Path) -> None:
# 出力にデータ・設定ファイルの場所を出さない（SPEC_03「出力の規則」）
for form in (str(tmp_path), tmp_path.as_posix(), str(tmp_path.resolve())):
assert form not in text
assert "catalog.sqlite" not in text
assert "identity.sqlite" not in text
assert "config.toml" not in text
@pytest.fixture
def config_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
config_path = _write_config(tmp_path)
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
return config_path
# ---------------------------------------------------------------------------
# init
# ---------------------------------------------------------------------------
class TestInit:
def test_init_creates_catalog_store_and_identity(
self, tmp_path: Path, config_env: Path
) -> None:
assert main(["init"]) == 0
p = _paths(tmp_path)
assert p["catalog"].is_file()
assert p["store"].is_dir()
assert p["bundles"].is_dir()
assert p["backups"].is_dir()
assert p["identity"].is_file()
def test_init_applies_latest_migrations(self, tmp_path: Path, config_env: Path) -> None:
assert main(["init"]) == 0
p = _paths(tmp_path)
with sqlite3.connect(p["catalog"]) as conn:
assert catalog.schema_version(conn) == catalog.latest_version("catalog")
assert conn.execute("SELECT COUNT(*) FROM kinds").fetchone()[0] == 6
with sqlite3.connect(p["identity"]) as conn:
assert catalog.schema_version(conn) == catalog.latest_version("identity")
def test_init_twice_gives_same_result(
self, tmp_path: Path, config_env: Path, capsys: pytest.CaptureFixture[str]
) -> None:
p = _paths(tmp_path)
assert main(["init"]) == 0
capsys.readouterr()
with sqlite3.connect(p["catalog"]) as conn:
schema = conn.execute("SELECT sql FROM sqlite_master ORDER BY name").fetchall()
kinds = conn.execute("SELECT * FROM kinds ORDER BY kind").fetchall()
# 何度実行しても同じ（既存の行を消さない）
with sqlite3.connect(p["catalog"]) as conn:
conn.execute(
"INSERT INTO subjects (subject_id, created_at)"
" VALUES ('S000001', '2026-10-06T00:00:00Z')"
)
assert main(["init"]) == 0
out = capsys.readouterr().out
with sqlite3.connect(p["catalog"]) as conn:
assert conn.execute("SELECT sql FROM sqlite_master ORDER BY name").fetchall() == schema
assert conn.execute("SELECT * FROM kinds ORDER BY kind").fetchall() == kinds
assert conn.execute("SELECT COUNT(*) FROM subjects").fetchone()[0] == 1
assert catalog.schema_version(conn) == catalog.latest_version("catalog")
assert "種類" in out
def test_init_reports_versions_and_kind_counts(
self, tmp_path: Path, config_env: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
out = capsys.readouterr().out
latest = catalog.latest_version("catalog")
assert f"{latest}" in out
assert "asset 4" in out
assert "annotation 2" in out
def test_init_output_contains_no_paths(
self, tmp_path: Path, config_env: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["init"]) == 0
captured = capsys.readouterr()
_assert_no_paths(captured.out + captured.err, tmp_path)
def test_init_rejects_catalog_newer_than_library(
self, tmp_path: Path, config_env: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
p = _paths(tmp_path)
newer = catalog.latest_version("catalog") + 1
with sqlite3.connect(p["catalog"]) as conn:
conn.execute(f"PRAGMA user_version = {newer}")
capsys.readouterr()
assert main(["init"]) == 1
captured = capsys.readouterr()
assert len(captured.err.strip().splitlines()) == 1
_assert_no_paths(captured.err, tmp_path)
# ---------------------------------------------------------------------------
# 入力の問題（終了コード 1・標準エラーに 1 行）
# ---------------------------------------------------------------------------
class TestInputErrors:
def test_missing_config_exits_1_with_one_line(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
monkeypatch.setenv("TIASDATA_CONFIG", str(tmp_path / "missing.toml"))
assert main(["init"]) == 1
captured = capsys.readouterr()
assert captured.out == ""
assert len(captured.err.strip().splitlines()) == 1
assert "Traceback" not in captured.err
# 何も作らない
assert not (tmp_path / "data").exists()
def test_broken_toml_exits_1_with_one_line(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
config_path = tmp_path / "config.toml"
config_path.write_text("[catalog\nroot = ", encoding="utf-8")
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
assert main(["init"]) == 1
captured = capsys.readouterr()
assert len(captured.err.strip().splitlines()) == 1
assert "Traceback" not in captured.err
def test_unknown_key_exits_1_without_input_values(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
secret = "DUMMY-SECRET-VALUE-12345"
config_path = _write_config(tmp_path, f'stroe = "{secret}"\n')
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
assert main(["init"]) == 1
captured = capsys.readouterr()
assert len(captured.err.strip().splitlines()) == 1
# キーの位置は出すが，入力値・パスは出さない
assert "stroe" in captured.err
assert secret not in captured.err
_assert_no_paths(captured.err, tmp_path)
assert not (tmp_path / "data").exists()
def test_missing_required_key_exits_1(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
config_path = tmp_path / "conf" / "config.toml"
config_path.parent.mkdir()
config_path.write_text('[catalog]\nroot = "../data/tiasdata"\n', encoding="utf-8")
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
assert main(["init"]) == 1
captured = capsys.readouterr()
assert len(captured.err.strip().splitlines()) == 1
assert "identity" in captured.err
_assert_no_paths(captured.err, tmp_path)
def test_multiple_validation_errors_fit_in_one_line(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
config_path = tmp_path / "conf" / "config.toml"
config_path.parent.mkdir()
config_path.write_text(
'x = 1\n[catalog]\nfoo = "a"\n[servers.s]\nhost = "h"\n', encoding="utf-8"
)
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
assert main(["init"]) == 1
assert len(capsys.readouterr().err.strip().splitlines()) == 1
# ---------------------------------------------------------------------------
# 引数
# ---------------------------------------------------------------------------
class TestArguments:
def test_unknown_command_exits_2(self) -> None:
with pytest.raises(SystemExit) as exc:
main(["no-such-command"])
assert exc.value.code == 2
def test_unknown_option_exits_2(self, config_env: Path) -> None:
with pytest.raises(SystemExit) as exc:
main(["init", "--no-such-option"])
assert exc.value.code == 2
def test_no_command_prints_help_and_exits_0(self, capsys: pytest.CaptureFixture[str]) -> None:
assert main([]) == 0
out = capsys.readouterr().out
assert "init" in out
# ---------------------------------------------------------------------------
# 別プロセスでの実行（出力の文字コード）
# ---------------------------------------------------------------------------
def _run_cli(args: list[str], config_path: Path, cwd: Path) -> subprocess.CompletedProcess[bytes]:
env = {k: v for k, v in os.environ.items() if k not in ("PYTHONIOENCODING", "PYTHONUTF8")}
env["TIASDATA_CONFIG"] = str(config_path)
# パイプにつないだときの既定の文字コード（Windows では cp932）に任せる
return subprocess.run(
[sys.executable, "-m", "tiasdata.cli", *args],
capture_output=True,
env=env,
cwd=cwd,
check=False,
timeout=60,
)
class TestSubprocess:
def test_init_stdout_is_utf8(self, tmp_path: Path) -> None:
config_path = _write_config(tmp_path)
result = _run_cli(["init"], config_path, tmp_path)
assert result.returncode == 0, result.stderr.decode("utf-8", errors="replace")
out = result.stdout.decode("utf-8") # UTF-8 でなければ例外
assert "種類" in out
_assert_no_paths(out, tmp_path)
def test_init_twice_in_separate_processes(self, tmp_path: Path) -> None:
config_path = _write_config(tmp_path)
assert _run_cli(["init"], config_path, tmp_path).returncode == 0
second = _run_cli(["init"], config_path, tmp_path)
assert second.returncode == 0
assert second.stderr == b""
def test_error_stderr_is_utf8_one_line(self, tmp_path: Path) -> None:
config_path = _write_config(tmp_path, 'stroe = "x"\n')
result = _run_cli(["init"], config_path, tmp_path)
assert result.returncode == 1
err = result.stderr.decode("utf-8") # UTF-8 でなければ例外
assert len(err.strip().splitlines()) == 1
assert "Traceback" not in err
# ---------------------------------------------------------------------------
# import・stats・check（SPEC_03「CLI」「出力の規則」，PLAN_02 Step 3，testing.md 必須対象 6）
# ---------------------------------------------------------------------------
_DUMMY_KEYS = [str(990001 + i) for i in range(7)]
_SOURCE_NAME_PATTERN = re.compile(r"[0-9]+\.png", re.IGNORECASE)
def _assert_no_personal_info(text: str, env: ImportEnv) -> None:
# 患者 ID・元のファイル名（フォルダ名・<番号>.png）・パスを出さない
for key in _DUMMY_KEYS:
assert key not in text
for folder in TONGUE_COLOR_FOLDERS:
assert folder not in text
assert not _SOURCE_NAME_PATTERN.search(text)
root = env.config_path.parents[1]
for form in (str(root), root.as_posix(), str(root.resolve())):
assert form not in text
def _line_with(text: str, needle: str) -> str:
lines = [line for line in text.splitlines() if needle in line]
assert len(lines) == 1, text
return lines[0]
def _map_rows(env: ImportEnv) -> list[tuple[object, ...]]:
return [(number, 1, key) for number, key in sorted(env.patient_map.patient_keys.items())]
class TestImportCommand:
def test_init_import_stats_check_flow(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
imported = capsys.readouterr()
assert main(["stats"]) == 0
stats = capsys.readouterr()
assert main(["check"]) == 0
checked = capsys.readouterr()
assert imported.err == stats.err == checked.err == ""
assert "dummy_tongue" in imported.out
assert "18" in imported.out
assert "tongue_crop_masked" in stats.out
for slug in ("pale", "light_red", "red", "dark_red", "purple"):
assert slug in stats.out
for text in (imported.out, stats.out, checked.out):
_assert_no_personal_info(text, import_env)
def test_import_reports_review_count(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
write_reviews(import_env.reviews_dir, "dummy_tongue", [3, 10])
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 0
out = capsys.readouterr().out
assert "2 件" in _line_with(out, "要確認として登録")
_assert_no_personal_info(out, import_env)
def test_second_import_reports_skipped(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 0
out = capsys.readouterr().out
assert "18 件" in _line_with(out, "登録済み")
_assert_no_personal_info(out, import_env)
def test_unused_map_rows_count_and_row_numbers_only_with_flag(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
# 見出し 1 行 + 18 行の後の 20・21 行目が使わない行
rows = _map_rows(import_env) + [("合計", None, "990001"), (2.5, 1, "990002")]
write_patient_map(import_env.patient_map.path, rows)
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 0
hidden = _line_with(capsys.readouterr().out, "対応表")
assert "2 件" in hidden
assert "20" not in hidden
assert "21" not in hidden
assert main(["import", "dummy_tongue", "--show-source-names"]) == 0
out = capsys.readouterr().out
shown = _line_with(out, "対応表")
assert "20" in shown
assert "21" in shown
for key in _DUMMY_KEYS:
assert key not in out
@pytest.mark.parametrize(
"case",
["unknown-file", "number-not-in-map", "non-numeric-patient", "review-not-in-source"],
)
def test_import_error_is_one_line_without_personal_info(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str], case: str
) -> None:
root = import_env.source.root
rows = _map_rows(import_env)
planted: list[str] = []
if case == "unknown-file":
(root / "紅＿あり" / "leak_zz.jpg").write_bytes(b"x")
planted.append("leak_zz.jpg")
elif case == "number-not-in-map":
write_dummy_png(root / "紅＿あり" / "19.png", seed=19)
elif case == "non-numeric-patient":
rows[0] = (1, 1, "LEAKID")
write_patient_map(import_env.patient_map.path, rows)
planted.append("LEAKID")
else:
write_reviews(import_env.reviews_dir, "dummy_tongue", [99])
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 1
captured = capsys.readouterr()
assert captured.out == ""
assert len(captured.err.strip().splitlines()) == 1
assert "Traceback" not in captured.err
_assert_no_personal_info(captured.err, import_env)
for name in planted:
assert name not in captured.err
def test_unknown_source_exits_1(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "no_such_source"]) == 1
assert len(capsys.readouterr().err.strip().splitlines()) == 1
def test_import_without_source_argument_exits_2(self, import_env: ImportEnv) -> None:
with pytest.raises(SystemExit) as exc:
main(["import"])
assert exc.value.code == 2
def test_import_before_init_exits_1(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["import", "dummy_tongue"]) == 1
captured = capsys.readouterr()
assert len(captured.err.strip().splitlines()) == 1
_assert_no_personal_info(captured.err, import_env)
def test_missing_openpyxl_is_one_line_error(
self,
import_env: ImportEnv,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
assert main(["init"]) == 0
capsys.readouterr()
monkeypatch.setitem(sys.modules, "openpyxl", None)
assert main(["import", "dummy_tongue"]) == 1
err = capsys.readouterr().err
assert len(err.strip().splitlines()) == 1
assert "uv sync --extra import" in err
conn = sqlite3.connect(import_env.catalog_root / "catalog.sqlite")
try:
assert conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0] == 0
finally:
conn.close()
class TestStatsAndCheckCommands:
def test_check_exit_1_and_counts_on_problem(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
store_dir = import_env.catalog_root / "store"
path = next(p for p in store_dir.rglob("*") if p.is_file())
path.chmod(path.stat().st_mode | stat.S_IWUSR)
capsys.readouterr()
assert main(["check"]) == 1
captured = capsys.readouterr()
assert "1 件" in _line_with(captured.out, "読み取り専用")
_assert_no_personal_info(captured.out + captured.err, import_env)
assert path.name not in captured.out
def test_stats_and_check_in_subprocess_are_utf8(
self, import_env: ImportEnv, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
# 取り込みは同じプロセスで行う（要確認リストの差し替えは別プロセスに効かないため）
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
for command in ("stats", "check"):
result = _run_cli([command], import_env.config_path, tmp_path)
assert result.returncode == 0, result.stderr.decode("utf-8", errors="replace")
out = result.stdout.decode("utf-8") # UTF-8 でなければ例外
_assert_no_personal_info(out, import_env)
# ---------------------------------------------------------------------------
# dataset list / show / freeze
# ---------------------------------------------------------------------------
def _versions(env: ImportEnv) -> list[tuple]:
conn = sqlite3.connect(env.catalog_root / "catalog.sqlite")
try:
return conn.execute(
"SELECT name, version, hub_commit, manifest_sha256, note FROM dataset_versions"
" ORDER BY name, version"
).fetchall()
finally:
conn.close()
def _rebuild(env: ImportEnv, version: int) -> bytes:
from tiasdata.datasets import rebuild_manifest
conn = catalog.connect(env.catalog_root / "catalog.sqlite")
try:
return rebuild_manifest(conn, "tongue_color", version)
finally:
conn.close()
def _assert_one_line_error(err: str, env: ImportEnv) -> None:
assert len(err.strip().splitlines()) == 1, err
assert err.startswith("エラー: ")
assert "Traceback" not in err
_assert_no_personal_info(err, env)
class TestDatasetCommands:
def test_list_freeze_show_flow(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
capsys.readouterr()
# 版の無い定義も「版なし」で表示する
assert main(["dataset", "list"]) == 0
out = capsys.readouterr().out
assert "データセット: 1 件" in out
assert "tongue_color: 版なし" in out
_assert_no_personal_info(out, import_env)
assert main(["dataset", "freeze", "tongue_color", "--note", "最初の版"]) == 0
out = capsys.readouterr().out
assert "版 1 を作った" in out
assert "ファイル 18 件・所見 18 件" in out
_assert_no_personal_info(out, import_env)
[(name, version, hub_commit, digest, note)] = _versions(import_env)
assert (name, version, note) == ("tongue_color", 1, "最初の版")
assert hub_commit == datasets_env.repo_state.commit
assert digest in out
# 構成が同じなら作らない
assert main(["dataset", "freeze", "tongue_color"]) == 0
out = capsys.readouterr().out
assert "作らなかった" in out
assert len(_versions(import_env)) == 1
assert main(["dataset", "show", "tongue_color"]) == 0
out = capsys.readouterr().out
assert "データセット: tongue_color" in out
assert "版: 1 件" in out
line = _line_with(out, "v1:")
assert "ファイル 18 件" in line
assert "所見 18 件" in line
assert "備考 最初の版" in line
assert digest in line
_assert_no_personal_info(out, import_env)
assert main(["dataset", "list"]) == 0
out = capsys.readouterr().out
assert "tongue_color: 版 1 件（最新 v1・ファイル 18 件）" in out
_assert_no_personal_info(out, import_env)
def test_new_import_creates_new_version_and_keeps_old(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
# 必須対象 1 を取り込みから通しで: 新しい画像を取り込んで版 2 を作っても版 1 は変わらない
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
assert main(["dataset", "freeze", "tongue_color"]) == 0
manifest_v1 = _rebuild(import_env, 1)
[v1] = _versions(import_env)
write_dummy_png(import_env.source.root / "紅＿あり" / "19.png", seed=19)
write_patient_map(import_env.patient_map.path, [*_map_rows(import_env), (19, 2, "990003")])
assert main(["import", "dummy_tongue"]) == 0
capsys.readouterr()
assert main(["dataset", "freeze", "tongue_color"]) == 0
out = capsys.readouterr().out
assert "版 2 を作った" in out
assert "ファイル 19 件" in out
versions_now = _versions(import_env)
assert versions_now[0] == v1
assert _rebuild(import_env, 1) == manifest_v1
assert hashlib.sha256(manifest_v1).hexdigest() == v1[3]
assert _rebuild(import_env, 2).count(b"\n") == 20
capsys.readouterr()
assert main(["check"]) == 0
assert "凍結した版の manifest の不一致: 0 件（版 2 件）" in capsys.readouterr().out
def test_freeze_with_uncommitted_changes_exits_1(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
from tiasdata.repo import RepoState
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
datasets_env.repo_state = RepoState(commit="a" * 40, uncommitted=3)
capsys.readouterr()
assert main(["dataset", "freeze", "tongue_color"]) == 1
captured = capsys.readouterr()
assert captured.out == ""
assert "未コミット" in captured.err
_assert_one_line_error(captured.err, import_env)
assert _versions(import_env) == []
def test_freeze_with_zero_files_exits_1(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
capsys.readouterr()
assert main(["dataset", "freeze", "tongue_color"]) == 1
captured = capsys.readouterr()
assert "0 件" in captured.err
_assert_one_line_error(captured.err, import_env)
def test_freeze_duplicate_annotation_exits_1_without_personal_info(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
conn = sqlite3.connect(import_env.catalog_root / "catalog.sqlite")
try:
conn.execute(
"INSERT INTO annotations (kind, capture_id, value, source, created_at)"
" SELECT kind, capture_id, value, 'test', created_at FROM annotations"
" ORDER BY annotation_id LIMIT 1"
)
conn.commit()
finally:
conn.close()
capsys.readouterr()
assert main(["dataset", "freeze", "tongue_color"]) == 1
captured = capsys.readouterr()
assert "2 つ以上" in captured.err
_assert_one_line_error(captured.err, import_env)
assert _versions(import_env) == []
@pytest.mark.parametrize("command", ["show", "freeze"])
@pytest.mark.parametrize("name", ["no_such", "../tongue_color", "Bad"])
def test_unknown_or_invalid_name_exits_1(
self,
import_env: ImportEnv,
datasets_env,
capsys: pytest.CaptureFixture[str],
command: str,
name: str,
) -> None:
assert main(["init"]) == 0
capsys.readouterr()
assert main(["dataset", command, name]) == 1
_assert_one_line_error(capsys.readouterr().err, import_env)
def test_invalid_definition_exits_1(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
(datasets_env.datasets_dir / "tongue_color.toml").write_bytes(
b'name = "tongue_color"\n[select]\nasset_kind = "tongue_crop_masked"\nfoo = 1\n'
)
capsys.readouterr()
assert main(["dataset", "freeze", "tongue_color"]) == 1
_assert_one_line_error(capsys.readouterr().err, import_env)
def test_dataset_before_init_exits_1(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
for args in (["list"], ["show", "tongue_color"], ["freeze", "tongue_color"]):
assert main(["dataset", *args]) == 1
_assert_one_line_error(capsys.readouterr().err, import_env)
def test_freeze_without_name_exits_2(self, import_env: ImportEnv, datasets_env) -> None:
with pytest.raises(SystemExit) as exc:
main(["dataset", "freeze"])
assert exc.value.code == 2
def test_list_keeps_dataset_whose_definition_was_removed(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
assert main(["dataset", "freeze", "tongue_color"]) == 0
(datasets_env.datasets_dir / "tongue_color.toml").unlink()
capsys.readouterr()
assert main(["dataset", "list"]) == 0
assert "tongue_color（定義ファイルなし）: 版 1 件" in capsys.readouterr().out
assert main(["dataset", "show", "tongue_color"]) == 0
assert "版: 1 件" in capsys.readouterr().out
# ---------------------------------------------------------------------------
# export（SPEC_03「CLI」「バンドル」，testing.md「CLI のエントリポイント」・必須対象 5・6）
# ---------------------------------------------------------------------------
class TestExportCommand:
def test_init_import_freeze_export_flow(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
from tiasdata import open_bundle
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
assert main(["dataset", "freeze", "tongue_color"]) == 0
[(_, _, _, digest, _)] = _versions(import_env)
capsys.readouterr()
assert main(["export", "tongue_color", "--version", "1"]) == 0
captured = capsys.readouterr()
assert captured.err == ""
assert "書き出した" in captured.out
assert "ファイル 18 件" in captured.out
assert digest in captured.out
_assert_no_personal_info(captured.out, import_env)
target = import_env.catalog_root / "bundles" / "tongue_color" / "v1"
bundle = open_bundle(target)
assert bundle.verify() is None
assert bundle.fingerprint == {
"name": "tongue_color",
"version": 1,
"manifest_sha256": digest,
}
assert len(bundle.rows) == 18
assert (target / "manifest.csv").read_bytes() == _rebuild(import_env, 1)
# 取り込んだ画像がそのままの中身で入っている
sources = {
hashlib.sha256(image.path.read_bytes()).hexdigest(): image
for image in import_env.source.images
}
for row in bundle.rows:
image = sources[row.sha256]
assert bundle.file_path(row).read_bytes() == image.path.read_bytes()
assert row.annotations["tongue_color"] == {
"label": image.label,
"severity": image.severity,
}
# 患者 ID・元のファイル名がバンドルのメタデータに入っていない
# （SHA-256 の 16 進が偶然 <数字>.png・患者 ID に見えないよう SHA-256 を除いて調べる）
for filename in ("dataset.json", "manifest.csv", "SHA256SUMS"):
text = (target / filename).read_bytes().decode("utf-8")
_assert_no_personal_info(re.sub(r"[0-9a-f]{64}", "<sha256>", text), import_env)
before = {p: p.read_bytes() for p in target.rglob("*") if p.is_file()}
# 2 回目は照合して使う（上書きしない）
assert main(["export", "tongue_color", "--version", "1"]) == 0
out = capsys.readouterr().out
assert "照合して使った" in out
_assert_no_personal_info(out, import_env)
assert {p: p.read_bytes() for p in target.rglob("*") if p.is_file()} == before
# 書き出したバンドルは check の対象外で，問題を増やさない
assert main(["check"]) == 0
capsys.readouterr()
# 改変した既存のバンドルは上書きせず終了コード 1
first = sorted((target / "files").iterdir())[0]
first.write_bytes(b"tampered")
assert main(["export", "tongue_color", "--version", "1"]) == 1
captured = capsys.readouterr()
assert captured.out == ""
_assert_one_line_error(captured.err, import_env)
assert first.read_bytes() == b"tampered"
# 存在しない版・データセット・不正な名前は終了コード 1
for args in (
["tongue_color", "--version", "2"],
["no_such", "--version", "1"],
["../tongue_color", "--version", "1"],
["tongue_color", "--version", "0"],
):
assert main(["export", *args]) == 1
captured = capsys.readouterr()
assert captured.out == ""
_assert_one_line_error(captured.err, import_env)
def test_export_before_init_exits_1(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["export", "tongue_color", "--version", "1"]) == 1
_assert_one_line_error(capsys.readouterr().err, import_env)
@pytest.mark.parametrize(
"args",
[
["export", "tongue_color"],
["export", "--version", "1"],
["export", "tongue_color", "--version", "abc"],
["export", "tongue_color", "--version"],
],
)
def test_export_argument_errors_exit_2(self, config_env: Path, args: list[str]) -> None:
with pytest.raises(SystemExit) as exc:
main(args)
assert exc.value.code == 2
# ---------------------------------------------------------------------------
# 要確認の印・撮影にひも付かないファイルの表示（2026-10-08 のユーザー判断）
# ---------------------------------------------------------------------------
_REASON = "理由のテキストLEAKzz"
def _share_first_image(env: ImportEnv, number: int) -> None:
# 番号 1 と同じ中身を同じフォルダ（同じラベル）に置き，対応表に行を足す
first = env.source.images[0]
(first.path.parent / f"{number}.png").write_bytes(first.path.read_bytes())
write_patient_map(env.patient_map.path, [*_map_rows(env), (number, 1, "990088")])
def _write_reviews_with_reason(env: ImportEnv, numbers: list[int]) -> None:
text = "".join(
f'[[review]]\nnumber = {n}\nreason = "{_REASON}"\nresolve = "確かめ方zz"\n'
"since = 2026-10-06\n\n"
for n in numbers
)
(env.reviews_dir / "dummy_tongue.toml").write_bytes(text.encode("utf-8"))
def _assert_review_safe(text: str, env: ImportEnv) -> None:
_assert_no_personal_info(text, env)
assert "990088" not in text
assert "保留" not in text
assert "hold" not in text
class TestReviewOutput:
def test_import_stats_check_freeze_with_reviews(
self, import_env: ImportEnv, datasets_env, capsys: pytest.CaptureFixture[str]
) -> None:
_share_first_image(import_env, 19)
_write_reviews_with_reason(import_env, [1, 5, 19])
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 0
out = capsys.readouterr().out
assert "3 件" in _line_with(out, "要確認リストの番号")
assert "reviews/dummy_tongue.toml" in _line_with(out, "要確認リストの番号")
assert _line_with(out, "要確認として登録") == (
"要確認として登録: 2 件（うち撮影にひも付けないファイル 1 件・番号 2 個）"
)
assert _line_with(out, "既存の撮影・ファイルに要確認の印を足した") == (
"既存の撮影・ファイルに要確認の印を足した: 0 件"
)
assert _line_with(out, "要確認リストから外れたが印が残っている") == (
"要確認リストから外れたが印が残っている: 0 件"
)
assert _REASON not in out
_assert_review_safe(out, import_env)
assert main(["stats"]) == 0
out = capsys.readouterr().out
assert _line_with(out, "撮影にひも付かないファイル") == "撮影にひも付かないファイル: 1 件"
assert _line_with(out, "review").strip() == (
"review: 2 件（撮影に付いたもの 1 件・ファイルに付いたもの 1 件）"
)
# review は件数だけ（reason の全文等の内訳を出さない）
assert _REASON not in out
assert "確かめ方zz" not in out
assert "2026-10-06" not in out
_assert_review_safe(out, import_env)
assert main(["check"]) == 0
out = capsys.readouterr().out
assert "撮影にひも付かないファイル（問題には数えない）: 1 件" in out
assert "問題: 0 件" in out
_assert_review_safe(out, import_env)
assert main(["dataset", "freeze", "tongue_color"]) == 0
out = capsys.readouterr().out
assert "版 1 を作った" in out
# 印の付いた撮影（番号 5）は版に入り，撮影にひも付かないファイルは除く
assert "ファイル 17 件" in out
assert "必須の所見が無く除いた 0 件" in out
assert (
_line_with(out, "撮影にひも付かないため除いた") == "撮影にひも付かないため除いた: 1 件"
)
_assert_review_safe(out, import_env)
assert _rebuild(import_env, 1).count(b"\n") == 18
def test_import_reports_added_and_stale_marks(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
_write_reviews_with_reason(import_env, [3, 10])
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 0
out = capsys.readouterr().out
assert "2 件" in _line_with(out, "既存の撮影・ファイルに要確認の印を足した")
assert "0 件" in _line_with(out, "要確認として登録")
_write_reviews_with_reason(import_env, [10])
assert main(["import", "dummy_tongue"]) == 0
out = capsys.readouterr().out
assert "0 件" in _line_with(out, "既存の撮影・ファイルに要確認の印を足した")
assert _line_with(out, "要確認リストから外れたが印が残っている") == (
"要確認リストから外れたが印が残っている: 1 件"
)
_assert_review_safe(out, import_env)
def test_stats_without_marks_has_no_review_line(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
assert main(["import", "dummy_tongue"]) == 0
capsys.readouterr()
assert main(["stats"]) == 0
out = capsys.readouterr().out
assert "review" not in out
assert _line_with(out, "撮影にひも付かないファイル") == "撮影にひも付かないファイル: 0 件"
def test_stats_of_empty_catalog_shows_none(
self, config_env: Path, capsys: pytest.CaptureFixture[str]
) -> None:
assert main(["init"]) == 0
capsys.readouterr()
assert main(["stats"]) == 0
out = capsys.readouterr().out
assert " なし" in out
assert "review" not in out
def test_different_labels_error_is_one_line(
self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
) -> None:
first = import_env.source.images[0]
(import_env.source.root / "紅＿あり" / "19.png").write_bytes(first.path.read_bytes())
write_patient_map(import_env.patient_map.path, [*_map_rows(import_env), (19, 1, "990088")])
_write_reviews_with_reason(import_env, [1, 19])
assert main(["init"]) == 0
capsys.readouterr()
assert main(["import", "dummy_tongue"]) == 1
captured = capsys.readouterr()
assert captured.out == ""
assert len(captured.err.strip().splitlines()) == 1
assert "Traceback" not in captured.err
_assert_review_safe(captured.err, import_env)
def test_import_help_mentions_reviews_not_holds(
self, capsys: pytest.CaptureFixture[str]
) -> None:
with pytest.raises(SystemExit) as excinfo:
main(["import", "--help"])
assert excinfo.value.code == 0
out = capsys.readouterr().out
assert "要確認" in out
assert "保留" not in out
assert "holds" not in out
