Newer
Older
TIASDataHub / tests / test_cli.py
"""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_holds,
    write_patient_map,
)

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] == 5
        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 1" 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_held_count(
        self, import_env: ImportEnv, capsys: pytest.CaptureFixture[str]
    ) -> None:
        write_holds(import_env.holds_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", "hold-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_holds(import_env.holds_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