"""データセットの定義・版の作成（SPEC_03「データセットの定義」「版の作成」，SPEC_02「datasets」）.
必須対象 1（版の凍結．最重要）: 所見を訂正・データを追加した後も，過去の版から同じ構成・
同じ manifest.csv（SHA-256 一致）が得られること．構成が同じなら新しい版を作らないこと．
目録はダミーの行を直接登録して作る（取り込みを繰り返さず速くするため．取り込みからの流れは
`test_cli.py` の統合テストで確かめる）．臨床データは使わない．
"""
import csv
import hashlib
import io
import sqlite3
from datetime import UTC, datetime
from pathlib import Path
import pytest
from pydantic import BaseModel, ConfigDict
from tiasdata import catalog, kinds
from tiasdata.catalog import versions
from tiasdata.datasets import (
DatasetDefinition,
check_definition_kinds,
collect_composition,
count_manifest_mismatches,
freeze_version,
list_definition_names,
load_definition,
parse_definition,
rebuild_manifest,
)
from tiasdata.manifest import manifest_sha256
from tiasdata.repo import RepoState
REPO_ROOT = Path(__file__).resolve().parents[1]
COMMIT = "0123456789abcdef0123456789abcdef01234567"
CLEAN = RepoState(commit=COMMIT, uncommitted=0)
NOW = datetime(2026, 10, 6, 9, 0, 0, tzinfo=UTC)
# 元のファイル名に仕込む架空の患者 ID（manifest・メッセージに出ないことを確かめる）
LEAK_KEY = "990077"
TONGUE_COLOR_TOML = """\
name = "tongue_color"
description = "舌色 5 分類＋程度（テスト）"
[select]
asset_kind = "tongue_crop_masked"
require = ["tongue_color"]
[columns]
tongue_color = ["label", "severity"]
"""
FROZEN_TABLES = ("datasets", "dataset_versions", "dataset_members", "dataset_member_annotations")
# ---------------------------------------------------------------------------
# ダミーの目録
# ---------------------------------------------------------------------------
def _sha(seed: object) -> str:
return hashlib.sha256(f"dummy-{seed}".encode()).hexdigest()
def _tc(label: str, severity: str) -> str:
return kinds.dump_annotation_value(
kinds.validate_annotation_value("tongue_color", {"label": label, "severity": severity})
)
class Cat:
"""ダミーの目録を作る補助．"""
def __init__(self, conn: sqlite3.Connection) -> None:
self.conn = conn
def subject(self, subject_id: str) -> None:
self.conn.execute(
"INSERT OR IGNORE INTO subjects (subject_id, created_at) VALUES (?, ?)",
(subject_id, "2026-10-01T00:00:00Z"),
)
self.conn.commit()
def capture(self, capture_id: str, subject_id: str) -> None:
self.subject(subject_id)
self.conn.execute(
"INSERT INTO captures (capture_id, subject_id) VALUES (?, ?)", (capture_id, subject_id)
)
self.conn.commit()
def asset(
self,
seed: object,
capture_id: str | None,
*,
kind: str = "tongue_crop_masked",
ext: str = "png",
) -> str:
sha = _sha(seed)
self.conn.execute(
"INSERT INTO assets (sha256, kind, capture_id, ext, size_bytes, source, source_name,"
" registered_at) VALUES (?, ?, ?, ?, 10, 'dummy_src', ?, '2026-10-01T00:00:00Z')",
(sha, kind, capture_id, ext, f"患者{LEAK_KEY}/{seed}_original.png"),
)
self.conn.commit()
return sha
def annotate(
self,
value: str,
*,
kind: str = "tongue_color",
capture_id: str | None = None,
sha256: str | None = None,
) -> int:
cursor = self.conn.execute(
"INSERT INTO annotations (kind, capture_id, sha256, value, source, created_at)"
" VALUES (?, ?, ?, ?, 'dummy_src', '2026-10-01T00:00:00Z')",
(kind, capture_id, sha256, value),
)
self.conn.commit()
return int(cursor.lastrowid)
def correct(self, old_id: int, value: str) -> int:
"""所見を訂正する（新しい行を足し，古い行の superseded_by を埋める）．"""
row = self.conn.execute(
"SELECT kind, capture_id, sha256 FROM annotations WHERE annotation_id = ?", (old_id,)
).fetchone()
new_id = self.annotate(value, kind=row[0], capture_id=row[1], sha256=row[2])
self.conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?", (new_id, old_id)
)
self.conn.commit()
return new_id
def tagged(
self, n: int, subject_id: str, label: str = "red", severity: str = "mild"
) -> tuple[str, int]:
"""撮影 1 つ・ファイル 1 つ・撮影に付いた舌色の所見 1 つを足す．"""
capture_id = f"C{n:06d}"
self.capture(capture_id, subject_id)
sha = self.asset(n, capture_id)
annotation_id = self.annotate(_tc(label, severity), capture_id=capture_id)
return sha, annotation_id
@pytest.fixture
def cat(catalog_conn: sqlite3.Connection) -> Cat:
"""撮影 4 つ（症例・撮影・SHA-256 の順が食い違う）を登録した目録．
- C000003（S000001）・C000004（S000001）・C000001（S000002）: 舌色あり
- C000002（S000003）: 舌色なし（必須の所見が無いので版に入らない）
"""
# テストの速さのため同期を省く（目録の中身・制約には影響しない）
catalog_conn.execute("PRAGMA synchronous = OFF")
c = Cat(catalog_conn)
c.tagged(3, "S000001", "red", "mild")
c.tagged(4, "S000001", "light_red", "normal")
c.tagged(1, "S000002", "pale", "strong")
c.capture("C000002", "S000003")
c.asset(2, "C000002")
return c
def _loaded(tmp_path: Path, text: str = TONGUE_COLOR_TOML, name: str = "tongue_color"):
directory = tmp_path / "datasets"
directory.mkdir(exist_ok=True)
# 改行を変換しないように bytes で書く
(directory / f"{name}.toml").write_bytes(text.encode("utf-8"))
return load_definition(directory, name)
def _freeze(conn: sqlite3.Connection, loaded, **kwargs):
kwargs.setdefault("repo_state", CLEAN)
kwargs.setdefault("now", NOW)
return freeze_version(conn, loaded, **kwargs)
def _snapshot(conn: sqlite3.Connection, name: str, version: int) -> dict:
return {
"version": conn.execute(
"SELECT * FROM dataset_versions WHERE name = ? AND version = ?", (name, version)
).fetchall(),
"members": conn.execute(
"SELECT sha256 FROM dataset_members WHERE name = ? AND version = ? ORDER BY sha256",
(name, version),
).fetchall(),
"member_annotations": conn.execute(
"SELECT sha256, annotation_id FROM dataset_member_annotations"
" WHERE name = ? AND version = ? ORDER BY sha256, annotation_id",
(name, version),
).fetchall(),
}
def _count(conn: sqlite3.Connection, table: str) -> int:
return conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
def _rows(data: bytes) -> list[dict[str, str]]:
return list(csv.DictReader(io.StringIO(data.decode("utf-8"), newline="")))
# ---------------------------------------------------------------------------
# 定義の形
# ---------------------------------------------------------------------------
class TestParseDefinition:
def test_repository_definition_is_valid(self) -> None:
text = (REPO_ROOT / "datasets" / "tongue_color.toml").read_bytes().decode("utf-8")
definition = parse_definition(text, "tongue_color")
assert definition.name == "tongue_color"
assert definition.select.asset_kind == "tongue_crop_masked"
assert definition.select.require == ["tongue_color"]
assert definition.columns == {"tongue_color": ["label", "severity"]}
assert definition.description
def test_repository_definition_uses_lf(self) -> None:
assert b"\r" not in (REPO_ROOT / "datasets" / "tongue_color.toml").read_bytes()
def test_columns_keep_order(self) -> None:
text = TONGUE_COLOR_TOML.replace(
'tongue_color = ["label", "severity"]', 'tongue_color = ["severity", "label"]'
)
assert parse_definition(text, "tongue_color").columns == {
"tongue_color": ["severity", "label"]
}
@pytest.mark.parametrize(
"text",
[
pytest.param(TONGUE_COLOR_TOML + "\nunknown = 1\n", id="unknown-top-key"),
pytest.param(
TONGUE_COLOR_TOML.replace("[select]\n", "[select]\nfilter = 1\n"),
id="unknown-select-key",
),
pytest.param(
TONGUE_COLOR_TOML.replace('description = "舌色 5 分類＋程度（テスト）"\n', ""),
id="missing-description",
),
pytest.param(
TONGUE_COLOR_TOML.replace('"舌色 5 分類＋程度（テスト）"', '""'),
id="empty-description",
),
pytest.param(
TONGUE_COLOR_TOML.replace('asset_kind = "tongue_crop_masked"\n', ""),
id="missing-asset-kind",
),
pytest.param('name = "tongue_color"\ndescription = "x"\n', id="missing-select"),
pytest.param(
TONGUE_COLOR_TOML.replace('["label", "severity"]', "[]"), id="empty-columns-fields"
),
pytest.param(
TONGUE_COLOR_TOML.replace('["label", "severity"]', '["label", "label"]'),
id="duplicated-column-field",
),
pytest.param(
TONGUE_COLOR_TOML.replace('["tongue_color"]', '["tongue_color", "tongue_color"]'),
id="duplicated-require",
),
pytest.param("name = [", id="broken-toml"),
],
)
def test_invalid_definition_is_value_error(self, text: str) -> None:
with pytest.raises(ValueError):
parse_definition(text, "tongue_color")
def test_name_must_match_file_name(self) -> None:
with pytest.raises(ValueError, match="一致しない"):
parse_definition(TONGUE_COLOR_TOML, "other_name")
def test_message_is_one_line(self) -> None:
text = TONGUE_COLOR_TOML + "\nunknown = 1\nfoo = 2\n"
with pytest.raises(ValueError) as exc:
parse_definition(text, "tongue_color")
assert "\n" not in str(exc.value)
class TestLoadDefinition:
def test_keeps_text_as_is(self, tmp_path: Path) -> None:
loaded = _loaded(tmp_path)
assert loaded.text == TONGUE_COLOR_TOML
assert loaded.definition.name == "tongue_color"
def test_missing_file(self, tmp_path: Path) -> None:
(tmp_path / "datasets").mkdir()
with pytest.raises(ValueError, match="無い"):
load_definition(tmp_path / "datasets", "tongue_color")
@pytest.mark.parametrize("name", ["../tongue_color", "Tongue", "1abc", "a-b", ""])
def test_invalid_name(self, tmp_path: Path, name: str) -> None:
_loaded(tmp_path)
with pytest.raises(ValueError):
load_definition(tmp_path / "datasets", name)
def test_list_definition_names(self, tmp_path: Path) -> None:
_loaded(tmp_path)
_loaded(tmp_path, TONGUE_COLOR_TOML.replace('"tongue_color"\n', '"abc"\n', 1), "abc")
(tmp_path / "datasets" / "Bad-Name.toml").write_bytes(b"")
(tmp_path / "datasets" / "readme.md").write_bytes(b"")
assert list_definition_names(tmp_path / "datasets") == ["abc", "tongue_color"]
class TestCheckDefinitionKinds:
@pytest.mark.parametrize(
("old", "new"),
[
pytest.param('"tongue_crop_masked"', '"no_such_asset"', id="unknown-asset-kind"),
pytest.param(
'asset_kind = "tongue_crop_masked"',
'asset_kind = "tongue_color"',
id="asset-kind-is-annotation",
),
pytest.param(
'require = ["tongue_color"]', 'require = ["no_such"]', id="unknown-require"
),
pytest.param(
'require = ["tongue_color"]', 'require = ["tongue_raw"]', id="require-is-asset"
),
pytest.param('["label", "severity"]', '["label", "colour"]', id="unknown-field"),
pytest.param("tongue_color = [", "no_such = [", id="unknown-column-kind"),
],
)
def test_rejects_unknown_kinds_and_fields(
self, catalog_conn: sqlite3.Connection, old: str, new: str
) -> None:
definition = parse_definition(TONGUE_COLOR_TOML.replace(old, new), "tongue_color")
with pytest.raises(ValueError):
check_definition_kinds(definition, versions.kind_categories(catalog_conn))
def test_accepts_repository_definition(self, catalog_conn: sqlite3.Connection) -> None:
definition = parse_definition(TONGUE_COLOR_TOML, "tongue_color")
check_definition_kinds(definition, versions.kind_categories(catalog_conn))
def test_unknown_kind_stops_freeze_without_rows(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path, TONGUE_COLOR_TOML.replace('"tongue_crop_masked"', '"no_such"'))
with pytest.raises(ValueError):
_freeze(cat.conn, loaded)
assert all(_count(cat.conn, t) == 0 for t in FROZEN_TABLES)
# ---------------------------------------------------------------------------
# 版の作成
# ---------------------------------------------------------------------------
class TestFreeze:
def test_creates_first_version(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
result = _freeze(cat.conn, loaded, note="最初の版")
assert result.created
assert result.version == 1
assert result.candidates == 4
assert result.members == 3
assert result.member_annotations == 3
record = versions.get_version(cat.conn, "tongue_color", 1)
assert record is not None
assert record.definition == TONGUE_COLOR_TOML
assert record.hub_commit == COMMIT
assert record.created_at == "2026-10-06T09:00:00Z"
assert record.note == "最初の版"
assert record.manifest_sha256 == result.manifest_sha256
# 記録した SHA-256 は構成から作った manifest のもの
assert (
manifest_sha256(rebuild_manifest(cat.conn, "tongue_color", 1)) == record.manifest_sha256
)
members = {m.sha256 for m in versions.version_members(cat.conn, "tongue_color", 1)}
assert members == {_sha(1), _sha(3), _sha(4)}
dataset = versions.get_dataset(cat.conn, "tongue_color")
assert dataset is not None
assert dataset.description == "舌色 5 分類＋程度（テスト）"
def test_manifest_content(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
data = rebuild_manifest(cat.conn, "tongue_color", 1)
assert not data.startswith(b"\xef\xbb\xbf")
assert b"\r" not in data
text = data.decode("utf-8")
assert text.splitlines()[0] == (
"path,sha256,kind,subject_id,capture_id,tongue_color.label,tongue_color.severity"
)
rows = _rows(data)
# subject_id・capture_id・sha256 の順
assert [(r["subject_id"], r["capture_id"]) for r in rows] == [
("S000001", "C000003"),
("S000001", "C000004"),
("S000002", "C000001"),
]
assert rows[0]["path"] == f"files/{_sha(3)}.png"
assert rows[0]["sha256"] == _sha(3)
assert rows[0]["kind"] == "tongue_crop_masked"
assert (rows[0]["tongue_color.label"], rows[0]["tongue_color.severity"]) == ("red", "mild")
assert (rows[1]["tongue_color.label"], rows[1]["tongue_color.severity"]) == (
"light_red",
"normal",
)
# 元のファイル名・患者 ID は出さない
assert LEAK_KEY not in text
assert "original" not in text
assert "dummy_src" not in text
def test_columns_follow_definition_order(self, tmp_path: Path, cat: Cat) -> None:
text = TONGUE_COLOR_TOML.replace('["label", "severity"]', '["severity", "label"]')
_freeze(cat.conn, _loaded(tmp_path, text))
header = rebuild_manifest(cat.conn, "tongue_color", 1).decode().splitlines()[0]
assert header.endswith("tongue_color.severity,tongue_color.label")
def test_same_composition_does_not_create_version(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
first = _freeze(cat.conn, loaded)
before = {t: _count(cat.conn, t) for t in FROZEN_TABLES}
second = _freeze(
cat.conn,
loaded,
repo_state=RepoState("f" * 40, 0),
note="x",
now=datetime(2027, 1, 1, tzinfo=UTC),
)
assert not second.created
assert second.version == 1
assert second.manifest_sha256 == first.manifest_sha256
assert {t: _count(cat.conn, t) for t in FROZEN_TABLES} == before
def test_file_without_required_annotation_is_not_a_change(
self, tmp_path: Path, cat: Cat
) -> None:
loaded = _loaded(tmp_path)
_freeze(cat.conn, loaded)
cat.capture("C000005", "S000004")
cat.asset(5, "C000005") # 舌色なし
cat.asset(6, "C000005", kind="tongue_raw") # 種類が違う
result = _freeze(cat.conn, loaded)
assert not result.created
assert result.candidates == 5
def test_definition_change_creates_version_even_if_composition_same(
self, tmp_path: Path, cat: Cat
) -> None:
_freeze(cat.conn, _loaded(tmp_path))
changed = TONGUE_COLOR_TOML.replace("（テスト）", "（説明だけ変更）")
result = _freeze(cat.conn, _loaded(tmp_path, changed))
assert result.created
assert result.version == 2
assert versions.get_version(cat.conn, "tongue_color", 2).definition == changed
# 構成は同じ
assert (
_snapshot(cat.conn, "tongue_color", 1)["members"]
== _snapshot(cat.conn, "tongue_color", 2)["members"]
)
# datasets の説明は最初の版のまま
assert versions.get_dataset(cat.conn, "tongue_color").description == (
"舌色 5 分類＋程度（テスト）"
)
# 同じ定義でもう一度は作らない
assert not _freeze(cat.conn, _loaded(tmp_path, changed)).created
def test_definition_whitespace_change_creates_version(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
assert _freeze(cat.conn, _loaded(tmp_path, TONGUE_COLOR_TOML + "\n")).created
def test_empty_note_is_stored_as_none(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path), note="")
assert versions.get_version(cat.conn, "tongue_color", 1).note is None
def test_versions_are_numbered_per_dataset(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
other = TONGUE_COLOR_TOML.replace('name = "tongue_color"', 'name = "other_set"')
result = _freeze(cat.conn, _loaded(tmp_path, other, "other_set"))
assert result.created
assert result.version == 1
def test_datasets_dir_with_crlf_definition_is_recorded_as_is(
self, tmp_path: Path, cat: Cat
) -> None:
crlf = TONGUE_COLOR_TOML.replace("\n", "\r\n")
_freeze(cat.conn, _loaded(tmp_path, crlf))
assert versions.get_version(cat.conn, "tongue_color", 1).definition == crlf
class TestFrozenVersion:
"""必須対象 1: 過去の版は所見の訂正・データの追加の後も変わらない（最重要）．"""
def test_freeze_keeps_past_version_unchanged(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
v1 = _freeze(cat.conn, loaded)
snapshot = _snapshot(cat.conn, "tongue_color", 1)
manifest_v1 = rebuild_manifest(cat.conn, "tongue_color", 1)
# 所見の訂正（C000003: red/mild → dark_red/strong）
old_id = cat.conn.execute(
"SELECT annotation_id FROM annotations WHERE capture_id = 'C000003'"
).fetchone()[0]
cat.correct(old_id, _tc("dark_red", "strong"))
# 新しい画像（既存の症例の新しい撮影）と，既存の撮影に後から舌色を付ける
cat.tagged(7, "S000002", "purple", "mild")
cat.annotate(_tc("pale", "mild"), capture_id="C000002")
v2 = _freeze(cat.conn, loaded)
assert v2.created
assert v2.version == 2
assert v2.members == 5
# 版 1 は構成・manifest のバイト列・SHA-256 が変わらない
assert _snapshot(cat.conn, "tongue_color", 1) == snapshot
rebuilt = rebuild_manifest(cat.conn, "tongue_color", 1)
assert rebuilt == manifest_v1
assert manifest_sha256(rebuilt) == v1.manifest_sha256
assert (
versions.get_version(cat.conn, "tongue_color", 1).manifest_sha256 == v1.manifest_sha256
)
# 版 1 は古い値を返す
row_v1 = {r["capture_id"]: r for r in _rows(rebuilt)}
assert row_v1["C000003"]["tongue_color.label"] == "red"
assert "C000007" not in row_v1
assert "C000002" not in row_v1
assert old_id in [a for (_, a) in snapshot["member_annotations"]]
# 版 2 には変更が反映される
manifest_v2 = rebuild_manifest(cat.conn, "tongue_color", 2)
assert manifest_sha256(manifest_v2) == v2.manifest_sha256
row_v2 = {r["capture_id"]: r for r in _rows(manifest_v2)}
assert (
row_v2["C000003"]["tongue_color.label"],
row_v2["C000003"]["tongue_color.severity"],
) == (
"dark_red",
"strong",
)
assert row_v2["C000007"]["tongue_color.label"] == "purple"
assert row_v2["C000002"]["tongue_color.label"] == "pale"
ids_v2 = [a for (_, a) in _snapshot(cat.conn, "tongue_color", 2)["member_annotations"]]
assert old_id not in ids_v2
# 変更が無ければ版 3 は作らない
assert not _freeze(cat.conn, loaded).created
assert _count(cat.conn, "dataset_versions") == 2
assert count_manifest_mismatches(cat.conn) == (2, 0)
def test_correction_alone_creates_new_version(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
v1 = _freeze(cat.conn, loaded)
old_id = cat.conn.execute(
"SELECT annotation_id FROM annotations WHERE capture_id = 'C000001'"
).fetchone()[0]
cat.correct(old_id, _tc("pale", "mild"))
v2 = _freeze(cat.conn, loaded)
assert v2.created
assert v2.manifest_sha256 != v1.manifest_sha256
assert manifest_sha256(rebuild_manifest(cat.conn, "tongue_color", 1)) == v1.manifest_sha256
def test_correction_to_same_value_still_changes_composition(
self, tmp_path: Path, cat: Cat
) -> None:
# 構成は「ファイルと採用した所見（の行）」で比べる．値が同じでも行が変われば新しい版
loaded = _loaded(tmp_path)
v1 = _freeze(cat.conn, loaded)
old_id = cat.conn.execute(
"SELECT annotation_id FROM annotations WHERE capture_id = 'C000001'"
).fetchone()[0]
cat.correct(old_id, _tc("pale", "strong"))
v2 = _freeze(cat.conn, loaded)
assert v2.created
# manifest の中身は同じ
assert v2.manifest_sha256 == v1.manifest_sha256
def test_past_version_survives_definition_change(self, tmp_path: Path, cat: Cat) -> None:
v1 = _freeze(cat.conn, _loaded(tmp_path))
manifest_v1 = rebuild_manifest(cat.conn, "tongue_color", 1)
changed = TONGUE_COLOR_TOML.replace('["label", "severity"]', '["severity"]')
_freeze(cat.conn, _loaded(tmp_path, changed))
assert rebuild_manifest(cat.conn, "tongue_color", 1) == manifest_v1
assert manifest_sha256(manifest_v1) == v1.manifest_sha256
assert (
rebuild_manifest(cat.conn, "tongue_color", 2)
.decode()
.splitlines()[0]
.endswith("capture_id,tongue_color.severity")
)
@pytest.mark.parametrize(
"sql",
[
"UPDATE dataset_versions SET note = 'x'",
"UPDATE dataset_versions SET manifest_sha256 = '" + "0" * 64 + "'",
"DELETE FROM dataset_versions",
"UPDATE dataset_members SET sha256 = sha256",
"DELETE FROM dataset_members",
"UPDATE dataset_member_annotations SET annotation_id = annotation_id",
"DELETE FROM dataset_member_annotations",
],
)
def test_frozen_rows_cannot_be_updated_or_deleted(
self, tmp_path: Path, cat: Cat, sql: str
) -> None:
_freeze(cat.conn, _loaded(tmp_path))
snapshot = _snapshot(cat.conn, "tongue_color", 1)
with pytest.raises(sqlite3.DatabaseError):
cat.conn.execute(sql)
cat.conn.rollback()
assert _snapshot(cat.conn, "tongue_color", 1) == snapshot
def test_annotation_used_by_version_cannot_be_rewritten(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
with pytest.raises(sqlite3.DatabaseError):
cat.conn.execute(f"UPDATE annotations SET value = '{_tc('purple', 'mild')}'")
cat.conn.rollback()
class TestFreezeStops:
def _assert_nothing_created(self, conn: sqlite3.Connection) -> None:
assert all(_count(conn, t) == 0 for t in FROZEN_TABLES)
def test_uncommitted_changes_stop(self, tmp_path: Path, cat: Cat) -> None:
with pytest.raises(ValueError, match="未コミット"):
_freeze(cat.conn, _loaded(tmp_path), repo_state=RepoState(COMMIT, 2))
self._assert_nothing_created(cat.conn)
assert not cat.conn.in_transaction
def test_uncommitted_is_checked_before_catalog(
self, tmp_path: Path, catalog_conn: sqlite3.Connection
) -> None:
# 目録に条件に合うファイルが無くても，まず未コミットの変更で止まる
with pytest.raises(ValueError, match="未コミット"):
_freeze(catalog_conn, _loaded(tmp_path), repo_state=RepoState(COMMIT, 1))
def test_uncommitted_after_version_does_not_create(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
_freeze(cat.conn, loaded)
cat.tagged(9, "S000009")
with pytest.raises(ValueError, match="未コミット"):
_freeze(cat.conn, loaded, repo_state=RepoState(COMMIT, 1))
assert _count(cat.conn, "dataset_versions") == 1
def test_zero_files_stops(self, tmp_path: Path, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(ValueError, match="0 件"):
_freeze(catalog_conn, _loaded(tmp_path))
self._assert_nothing_created(catalog_conn)
def test_zero_files_with_required_annotation_stops(
self, tmp_path: Path, catalog_conn: sqlite3.Connection
) -> None:
c = Cat(catalog_conn)
c.capture("C000001", "S000001")
c.asset(1, "C000001")
with pytest.raises(ValueError, match="0 件"):
_freeze(catalog_conn, _loaded(tmp_path))
self._assert_nothing_created(catalog_conn)
def test_file_without_capture_is_excluded_and_counted(self, tmp_path: Path, cat: Cat) -> None:
# 2026-10-08 の仕様変更: 撮影にひも付かないファイルは停止せずに除き，数を報告する
sha = cat.asset(20, None)
cat.annotate(_tc("red", "mild"), sha256=sha) # ファイルに付いた所見で条件に合う
result = _freeze(cat.conn, _loaded(tmp_path))
assert result.members == 3
assert result.unlinked == 1
def test_file_without_capture_and_no_require_is_excluded(
self, tmp_path: Path, cat: Cat
) -> None:
cat.asset(21, None)
text = TONGUE_COLOR_TOML.replace('require = ["tongue_color"]', "require = []")
result = _freeze(cat.conn, _loaded(tmp_path, text))
assert result.unlinked == 1
def test_file_without_capture_not_matching_is_ignored(self, tmp_path: Path, cat: Cat) -> None:
cat.asset(22, None) # 必須の所見が無いので版に入らない
assert _freeze(cat.conn, _loaded(tmp_path)).members == 3
def test_two_valid_capture_annotations_stop(self, tmp_path: Path, cat: Cat) -> None:
cat.annotate(_tc("purple", "strong"), capture_id="C000003")
with pytest.raises(ValueError, match="2 つ以上") as exc:
_freeze(cat.conn, _loaded(tmp_path))
assert "C000003" in str(exc.value)
assert LEAK_KEY not in str(exc.value)
self._assert_nothing_created(cat.conn)
def test_capture_and_file_annotations_are_counted_together(
self, tmp_path: Path, cat: Cat
) -> None:
cat.annotate(_tc("purple", "strong"), sha256=_sha(4))
with pytest.raises(ValueError, match="2 つ以上"):
_freeze(cat.conn, _loaded(tmp_path))
self._assert_nothing_created(cat.conn)
def test_two_file_annotations_stop(self, tmp_path: Path, cat: Cat) -> None:
sha = cat.asset(23, "C000002")
cat.annotate(_tc("red", "mild"), sha256=sha)
cat.annotate(_tc("red", "strong"), sha256=sha)
with pytest.raises(ValueError, match="2 つ以上"):
_freeze(cat.conn, _loaded(tmp_path))
def test_superseded_annotation_is_not_counted(self, tmp_path: Path, cat: Cat) -> None:
old_id = cat.conn.execute(
"SELECT annotation_id FROM annotations WHERE capture_id = 'C000003'"
).fetchone()[0]
cat.correct(old_id, _tc("purple", "strong"))
result = _freeze(cat.conn, _loaded(tmp_path))
assert result.created
assert result.members == 3
def test_duplicate_after_version_does_not_change_past(self, tmp_path: Path, cat: Cat) -> None:
loaded = _loaded(tmp_path)
v1 = _freeze(cat.conn, loaded)
cat.annotate(_tc("purple", "strong"), capture_id="C000003")
with pytest.raises(ValueError, match="2 つ以上"):
_freeze(cat.conn, loaded)
assert _count(cat.conn, "dataset_versions") == 1
assert manifest_sha256(rebuild_manifest(cat.conn, "tongue_color", 1)) == v1.manifest_sha256
def test_catalog_is_not_in_transaction_after_failure(
self, tmp_path: Path, catalog_conn: sqlite3.Connection
) -> None:
with pytest.raises(ValueError):
_freeze(catalog_conn, _loaded(tmp_path))
assert not catalog_conn.in_transaction
class TestRebuild:
def test_unknown_version(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
with pytest.raises(ValueError):
rebuild_manifest(cat.conn, "tongue_color", 2)
with pytest.raises(ValueError):
rebuild_manifest(cat.conn, "no_such", 1)
def test_rebuild_is_deterministic(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
assert rebuild_manifest(cat.conn, "tongue_color", 1) == rebuild_manifest(
cat.conn, "tongue_color", 1
)
def test_composition_is_independent_of_insert_order(
self, tmp_path: Path, catalog_conn: sqlite3.Connection, tmp_path_factory
) -> None:
# 登録の順が違っても同じ構成なら同じ manifest になる
other_path = tmp_path_factory.mktemp("other") / "catalog.sqlite"
other = catalog.connect(other_path, create=True)
try:
catalog.apply_migrations(other, "catalog")
a, b = Cat(catalog_conn), Cat(other)
for n, s in [(3, "S000001"), (1, "S000002"), (4, "S000001")]:
a.tagged(n, s)
for n, s in [(4, "S000001"), (1, "S000002"), (3, "S000001")]:
b.tagged(n, s)
loaded = _loaded(tmp_path)
assert (
_freeze(catalog_conn, loaded).manifest_sha256
== _freeze(other, loaded).manifest_sha256
)
finally:
other.close()
class TestCountManifestMismatches:
def test_no_versions(self, catalog_conn: sqlite3.Connection) -> None:
assert count_manifest_mismatches(catalog_conn) == (0, 0)
def test_after_freeze_and_corrections(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
old_id = cat.conn.execute(
"SELECT annotation_id FROM annotations WHERE capture_id = 'C000004'"
).fetchone()[0]
cat.correct(old_id, _tc("red", "mild"))
assert count_manifest_mismatches(cat.conn) == (1, 0)
def test_asset_kind_changed_is_mismatch(self, tmp_path: Path, cat: Cat) -> None:
# assets は凍結の対象外なので，参照先を書き換えると作り直した manifest が変わる
_freeze(cat.conn, _loaded(tmp_path))
cat.conn.execute("UPDATE assets SET kind = 'tongue_raw' WHERE sha256 = ?", (_sha(3),))
cat.conn.commit()
assert count_manifest_mismatches(cat.conn) == (1, 1)
def test_capture_subject_changed_is_mismatch(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
cat.subject("S000099")
cat.conn.execute("UPDATE captures SET subject_id = 'S000099' WHERE capture_id = 'C000001'")
cat.conn.commit()
assert count_manifest_mismatches(cat.conn) == (1, 1)
def test_tampered_manifest_sha256_is_mismatch(self, tmp_path: Path, cat: Cat) -> None:
# トリガを外して記録を壊す（点検が記録との不一致を見つけることを確かめるため）
_freeze(cat.conn, _loaded(tmp_path))
cat.conn.execute("DROP TRIGGER dataset_versions_no_update")
cat.conn.execute("UPDATE dataset_versions SET manifest_sha256 = ?", ("0" * 64,))
cat.conn.commit()
assert count_manifest_mismatches(cat.conn) == (1, 1)
def test_unparsable_recorded_definition_is_mismatch(self, tmp_path: Path, cat: Cat) -> None:
_freeze(cat.conn, _loaded(tmp_path))
cat.conn.execute("DROP TRIGGER dataset_versions_no_update")
cat.conn.execute("UPDATE dataset_versions SET definition = 'name = ['")
cat.conn.commit()
assert count_manifest_mismatches(cat.conn) == (1, 1)
# ---------------------------------------------------------------------------
# 拡張: 所見の種類を足しても既存の表を変えずに済む（PLAN_02 Step 4 の完了条件）
# ---------------------------------------------------------------------------
class TongueCoating(BaseModel):
"""テスト用のダミーの所見の種類（舌苔）．"""
model_config = ConfigDict(extra="forbid", frozen=True)
thickness: str
moist: str | None = None
COATING_TOML = """\
name = "coating"
description = "舌苔（テスト用のダミーの種類）"
[select]
asset_kind = "tongue_crop_masked"
require = ["tongue_coating"]
[columns]
tongue_coating = ["thickness", "moist"]
tongue_color = ["label"]
"""
def _schema(conn: sqlite3.Connection) -> dict:
tables = [
r[0]
for r in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name")
]
return {
"sql": conn.execute(
"SELECT type, name, sql FROM sqlite_master ORDER BY type, name"
).fetchall(),
"columns": {t: conn.execute(f"PRAGMA table_info({t})").fetchall() for t in tables},
}
class TestNewAnnotationKind:
def test_dataset_with_new_annotation_kind(
self, tmp_path: Path, cat: Cat, monkeypatch: pytest.MonkeyPatch
) -> None:
conn = cat.conn
tc = _freeze(conn, _loaded(tmp_path))
manifest_tc = rebuild_manifest(conn, "tongue_color", 1)
schema = _schema(conn)
# 種類の行と値の型を足す（表の構造は変えない）
conn.execute(
"INSERT INTO kinds (kind, category, description, value_schema_version)"
" VALUES ('tongue_coating', 'annotation', 'テスト用の舌苔', 1)"
)
conn.commit()
monkeypatch.setitem(kinds.ANNOTATION_VALUE_TYPES, "tongue_coating", TongueCoating)
# 撮影に付く苔の所見（C000003・C000002），ファイルに付く苔の所見（C000001 のファイル）
cat.annotate(
'{"moist":"wet","thickness":"thick"}', kind="tongue_coating", capture_id="C000003"
)
cat.annotate('{"thickness":"thin"}', kind="tongue_coating", capture_id="C000002")
cat.annotate('{"moist":null,"thickness":"none"}', kind="tongue_coating", sha256=_sha(1))
result = _freeze(conn, _loaded(tmp_path, COATING_TOML, "coating"))
assert result.created
assert result.members == 3
# 舌色は columns だけ（必須ではない）なので，あるファイルだけ採用する
assert result.member_annotations == 3 + 2
rows = {r["capture_id"]: r for r in _rows(rebuild_manifest(conn, "coating", 1))}
assert list(next(iter(rows.values())).keys()) == [
"path",
"sha256",
"kind",
"subject_id",
"capture_id",
"tongue_coating.thickness",
"tongue_coating.moist",
"tongue_color.label",
]
assert (
rows["C000003"]["tongue_coating.thickness"],
rows["C000003"]["tongue_coating.moist"],
) == (
"thick",
"wet",
)
assert rows["C000003"]["tongue_color.label"] == "red"
# 値が無い所見・項目は空
assert rows["C000002"]["tongue_coating.moist"] == ""
assert rows["C000002"]["tongue_color.label"] == ""
assert rows["C000001"]["tongue_coating.moist"] == ""
assert rows["C000001"]["tongue_coating.thickness"] == "none"
# 既存の表の構造は変わらず，既存のデータセットの版も変わらない
assert _schema(conn) == schema
assert rebuild_manifest(conn, "tongue_color", 1) == manifest_tc
assert not _freeze(conn, _loaded(tmp_path)).created
assert count_manifest_mismatches(conn) == (2, 0)
assert manifest_sha256(manifest_tc) == tc.manifest_sha256
def test_new_kind_without_value_type_is_rejected(self, tmp_path: Path, cat: Cat) -> None:
cat.conn.execute(
"INSERT INTO kinds (kind, category, description, value_schema_version)"
" VALUES ('tongue_coating', 'annotation', 'テスト用', 1)"
)
cat.conn.commit()
with pytest.raises(ValueError):
_freeze(cat.conn, _loaded(tmp_path, COATING_TOML, "coating"))
assert _count(cat.conn, "dataset_versions") == 0
# ---------------------------------------------------------------------------
# 純関数の当てはめ
# ---------------------------------------------------------------------------
def test_collect_composition_does_not_write(tmp_path: Path, cat: Cat) -> None:
definition: DatasetDefinition = _loaded(tmp_path).definition
composition = collect_composition(cat.conn, definition)
assert composition.candidates == 4
assert len(composition.members) == 3
assert composition.excluded == 1
assert all(_count(cat.conn, t) == 0 for t in FROZEN_TABLES)
# ---------------------------------------------------------------------------
# 撮影にひも付かないファイル・要確認の印（2026-10-08 のユーザー判断）
# ---------------------------------------------------------------------------
def _review(numbers: list[int]) -> str:
return kinds.dump_annotation_value(
kinds.validate_annotation_value(
"review",
{"reason": "理由", "resolve": "確かめ方", "since": "2026-10-06", "numbers": numbers},
)
)
class TestUnlinkedAndReview:
def test_unlinked_with_duplicated_annotations_is_excluded_not_stopped(
self, tmp_path: Path, cat: Cat
) -> None:
# 版に入れないファイルは所見の重複の確認の対象にしない
sha = cat.asset(20, None)
cat.annotate(_tc("red", "mild"), sha256=sha)
cat.annotate(_tc("pale", "mild"), sha256=sha)
result = _freeze(cat.conn, _loaded(tmp_path))
assert result.created
assert result.members == 3
assert result.unlinked == 1
assert result.candidates == 5
manifest = rebuild_manifest(cat.conn, "tongue_color", 1)
assert sha.encode() not in manifest
def test_composition_reports_unlinked_separately_from_excluded(
self, tmp_path: Path, cat: Cat
) -> None:
sha = cat.asset(20, None)
cat.annotate(_tc("red", "mild"), sha256=sha)
cat.asset(21, None) # 必須の所見が無い（撮影にひも付かないため除いた，には数えない）
composition = collect_composition(cat.conn, _loaded(tmp_path).definition)
assert composition.candidates == 6
assert len(composition.members) == 3
assert composition.unlinked == 1
# C000002 の 1 件と，所見の無い撮影なしのファイル 1 件
assert composition.excluded == 2
def test_only_unlinked_files_stops_with_zero(
self, tmp_path: Path, catalog_conn: sqlite3.Connection
) -> None:
c = Cat(catalog_conn)
for seed in (1, 2):
sha = c.asset(seed, None)
c.annotate(_tc("red", "mild"), sha256=sha)
with pytest.raises(ValueError, match="0 件") as excinfo:
_freeze(catalog_conn, _loaded(tmp_path))
message = str(excinfo.value)
assert "2 件" in message
assert "\n" not in message
assert LEAK_KEY not in message
assert "original" not in message
for table in FROZEN_TABLES:
assert _count(catalog_conn, table) == 0
def test_review_marked_capture_is_in_version(self, tmp_path: Path, cat: Cat) -> None:
# 印による除外はしない（印の付いた撮影も版に入る）．印は版の列にならない
cat.annotate(_review([3]), kind="review", capture_id="C000003")
result = _freeze(cat.conn, _loaded(tmp_path))
assert result.members == 3
assert result.member_annotations == 3
rows = {r["capture_id"]: r for r in _rows(rebuild_manifest(cat.conn, "tongue_color", 1))}
assert "C000003" in rows
assert not any("review" in column for column in rows["C000003"])
def test_marks_and_unlinked_files_added_later_keep_past_version(
self, tmp_path: Path, cat: Cat
) -> None:
# 必須対象 1: 版の後に印・撮影にひも付かないファイルを足しても，版 1 は変わらず
# 構成も変わらないので新しい版は作らない
loaded = _loaded(tmp_path)
v1 = _freeze(cat.conn, loaded)
snapshot = _snapshot(cat.conn, "tongue_color", 1)
manifest_v1 = rebuild_manifest(cat.conn, "tongue_color", 1)
cat.annotate(_review([3]), kind="review", capture_id="C000003")
sha = cat.asset(30, None)
cat.annotate(_tc("red", "mild"), sha256=sha)
cat.annotate(_review([30, 31]), kind="review", sha256=sha)
again = _freeze(cat.conn, loaded)
assert not again.created
assert again.version == 1
assert again.unlinked == 1
assert again.manifest_sha256 == v1.manifest_sha256
assert _snapshot(cat.conn, "tongue_color", 1) == snapshot
assert rebuild_manifest(cat.conn, "tongue_color", 1) == manifest_v1
assert count_manifest_mismatches(cat.conn) == (1, 0)
