"""目録・対応表のスキーマ（SPEC_02 データモデル）.
マイグレーション適用後の DB に対して，表の構成と制約（外部キー・CHECK・トリガ）を確かめる．
"""
import json
import sqlite3
import pytest
TS = "2026-10-06T00:00:00Z"
SHA_A = "a" * 64
SHA_B = "b" * 64
SHA_M = "c" * 64
EXPECTED_COLUMNS = {
"kinds": {"kind", "category", "description", "value_schema_version"},
"subjects": {"subject_id", "created_at", "note"},
"captures": {"capture_id", "subject_id", "captured_on", "device", "note"},
"assets": {
"sha256",
"kind",
"capture_id",
"ext",
"size_bytes",
"source",
"source_name",
"registered_at",
},
"derivations": {"derivation_id", "tool", "tool_version", "mode", "performed_at", "note"},
"derivation_inputs": {"derivation_id", "sha256", "role"},
"derivation_outputs": {"derivation_id", "sha256", "role"},
"annotations": {
"annotation_id",
"kind",
"capture_id",
"sha256",
"value",
"annotator",
"source",
"created_at",
"superseded_by",
},
"datasets": {"name", "description", "created_at"},
"dataset_versions": {
"name",
"version",
"definition",
"hub_commit",
"manifest_sha256",
"created_at",
"note",
},
"dataset_members": {"name", "version", "sha256"},
"dataset_member_annotations": {"name", "version", "sha256", "annotation_id"},
"deliveries": {
"delivery_id",
"name",
"version",
"server",
"path",
"delivered_at",
"verified",
},
}
TONGUE_COLOR = json.dumps({"label": "red", "severity": "mild"})
# ---------------------------------------------------------------------------
# ダミーの行を入れる補助
# ---------------------------------------------------------------------------
def _subject(conn: sqlite3.Connection, subject_id: str = "S000001") -> None:
conn.execute("INSERT INTO subjects (subject_id, created_at) VALUES (?, ?)", (subject_id, TS))
def _capture(
conn: sqlite3.Connection, capture_id: str = "C000001", subject_id: str = "S000001"
) -> None:
conn.execute(
"INSERT INTO captures (capture_id, subject_id) VALUES (?, ?)", (capture_id, subject_id)
)
def _asset(
conn: sqlite3.Connection,
sha256: str = SHA_A,
kind: str = "tongue_crop_masked",
capture_id: str | None = "C000001",
ext: str = "png",
registered_at: str = TS,
) -> None:
conn.execute(
"INSERT INTO assets (sha256, kind, capture_id, ext, size_bytes, source, source_name,"
" registered_at) VALUES (?, ?, ?, ?, 10, 'dummy_source', 'dummy.png', ?)",
(sha256, kind, capture_id, ext, registered_at),
)
def _annotation(
conn: sqlite3.Connection,
*,
capture_id: str | None = "C000001",
sha256: str | None = None,
kind: str = "tongue_color",
value: str = TONGUE_COLOR,
created_at: str = TS,
) -> int:
cur = conn.execute(
"INSERT INTO annotations (kind, capture_id, sha256, value, source, created_at)"
" VALUES (?, ?, ?, ?, 'dummy_source', ?)",
(kind, capture_id, sha256, value, created_at),
)
return cur.lastrowid
def _base(conn: sqlite3.Connection) -> None:
"""症例 1・撮影 1・ファイル 1 を入れる．"""
_subject(conn)
_capture(conn)
_asset(conn)
def _frozen_version(conn: sqlite3.Connection) -> int:
"""データセットの版 1 を作り，ファイル 1 件と所見 1 件を構成に入れる．"""
_base(conn)
annotation_id = _annotation(conn)
conn.execute(
"INSERT INTO datasets (name, description, created_at) VALUES ('tongue_color', 'd', ?)",
(TS,),
)
conn.execute(
"INSERT INTO dataset_versions (name, version, definition, hub_commit, manifest_sha256,"
" created_at) VALUES ('tongue_color', 1, 'name = \"tongue_color\"', 'abc123', ?, ?)",
(SHA_M, TS),
)
conn.execute(
"INSERT INTO dataset_members (name, version, sha256) VALUES ('tongue_color', 1, ?)",
(SHA_A,),
)
conn.execute(
"INSERT INTO dataset_member_annotations (name, version, sha256, annotation_id)"
" VALUES ('tongue_color', 1, ?, ?)",
(SHA_A, annotation_id),
)
conn.commit()
return annotation_id
# ---------------------------------------------------------------------------
# 表の構成
# ---------------------------------------------------------------------------
class TestTables:
def test_catalog_has_the_13_tables(self, catalog_conn: sqlite3.Connection) -> None:
rows = catalog_conn.execute(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
)
assert {name for (name,) in rows} == set(EXPECTED_COLUMNS)
assert len(EXPECTED_COLUMNS) == 13
@pytest.mark.parametrize("table", sorted(EXPECTED_COLUMNS))
def test_table_has_spec_columns(self, catalog_conn: sqlite3.Connection, table: str) -> None:
columns = {row[1] for row in catalog_conn.execute(f"PRAGMA table_info({table})")}
assert columns == EXPECTED_COLUMNS[table]
def test_identity_map_columns(self, identity_conn: sqlite3.Connection) -> None:
columns = {row[1] for row in identity_conn.execute("PRAGMA table_info(identity_map)")}
assert columns == {"patient_key", "subject_id", "created_at"}
def test_catalog_has_no_patient_key_column(self, catalog_conn: sqlite3.Connection) -> None:
# 実際の患者 ID は目録に入れない（対応表にだけ置く）
for table in EXPECTED_COLUMNS:
columns = {row[1] for row in catalog_conn.execute(f"PRAGMA table_info({table})")}
assert "patient_key" not in columns
def test_dataset_versions_primary_key_is_name_and_version(
self, catalog_conn: sqlite3.Connection
) -> None:
pk = {
row[1]: row[5]
for row in catalog_conn.execute("PRAGMA table_info(dataset_versions)")
if row[5] > 0
}
assert set(pk) == {"name", "version"}
# ---------------------------------------------------------------------------
# kinds
# ---------------------------------------------------------------------------
class TestKinds:
def test_initial_kinds(self, catalog_conn: sqlite3.Connection) -> None:
rows = catalog_conn.execute(
"SELECT kind, category, value_schema_version FROM kinds"
).fetchall()
assert {(k, c) for k, c, _ in rows} == {
("tongue_raw", "asset"),
("color_chart", "asset"),
("tongue_crop_masked", "asset"),
("tongue_mask", "asset"),
("tongue_color", "annotation"),
("review", "annotation"),
}
def test_only_annotation_kinds_have_value_schema_version(
self, catalog_conn: sqlite3.Connection
) -> None:
for _kind, category, version in catalog_conn.execute(
"SELECT kind, category, value_schema_version FROM kinds"
):
if category == "annotation":
assert version is not None and version >= 1
else:
assert version is None
def test_unknown_category_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description) VALUES ('x', 'other', 'd')"
)
def test_asset_kind_with_value_schema_version_is_rejected(
self, catalog_conn: sqlite3.Connection
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description, value_schema_version)"
" VALUES ('x', 'asset', 'd', 1)"
)
def test_annotation_kind_without_value_schema_version_is_rejected(
self, catalog_conn: sqlite3.Connection
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description) VALUES ('x', 'annotation', 'd')"
)
@pytest.mark.parametrize("kind", ["Tongue", "1abc", "tongue-color", "_x", ""])
def test_invalid_kind_name_is_rejected(
self, catalog_conn: sqlite3.Connection, kind: str
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description) VALUES (?, 'asset', 'd')",
(kind,),
)
def test_new_kind_can_be_added_as_row(self, catalog_conn: sqlite3.Connection) -> None:
# 拡張は行の追加で行う（SPEC_02「拡張の例」）
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description, value_schema_version)"
" VALUES ('coating_color', 'annotation', 'd', 1)"
)
catalog_conn.execute(
"INSERT INTO kinds (kind, category, description) VALUES ('analysis_csv', 'asset', 'd')"
)
# ---------------------------------------------------------------------------
# ID・値の書式
# ---------------------------------------------------------------------------
class TestFormats:
def test_valid_rows_are_accepted(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
_annotation(catalog_conn)
catalog_conn.execute(
"UPDATE captures SET captured_on = '2026-10-01', device = 'TIAS'"
" WHERE capture_id = 'C000001'"
)
@pytest.mark.parametrize("subject_id", ["S12345", "S0000001", "s000001", "X000001", "S00000a"])
def test_invalid_subject_id_is_rejected(
self, catalog_conn: sqlite3.Connection, subject_id: str
) -> None:
with pytest.raises(sqlite3.IntegrityError):
_subject(catalog_conn, subject_id)
@pytest.mark.parametrize("capture_id", ["C12345", "C0000001", "c000001", "S000001"])
def test_invalid_capture_id_is_rejected(
self, catalog_conn: sqlite3.Connection, capture_id: str
) -> None:
_subject(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_capture(catalog_conn, capture_id)
@pytest.mark.parametrize(
"sha256", ["A" * 64, "a" * 63, "a" * 65, "g" * 64, ""], ids=["upper", "63", "65", "g", "e"]
)
def test_invalid_sha256_is_rejected(
self, catalog_conn: sqlite3.Connection, sha256: str
) -> None:
_subject(catalog_conn)
_capture(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_asset(catalog_conn, sha256=sha256)
@pytest.mark.parametrize("ext", ["PNG", ".png", "", "p g"])
def test_invalid_ext_is_rejected(self, catalog_conn: sqlite3.Connection, ext: str) -> None:
_subject(catalog_conn)
_capture(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_asset(catalog_conn, ext=ext)
@pytest.mark.parametrize(
"created_at",
[
"2026-10-06 00:00:00",
"2026-10-06T00:00:00",
"2026-10-06T00:00:00+09:00",
"2026-10-06T00:00:00.123Z",
"2026-10-06",
],
)
def test_invalid_datetime_is_rejected(
self, catalog_conn: sqlite3.Connection, created_at: str
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO subjects (subject_id, created_at) VALUES ('S000001', ?)",
(created_at,),
)
@pytest.mark.parametrize("captured_on", ["2026/10/01", "2026-10-01T00:00:00Z", "20261001"])
def test_invalid_captured_on_is_rejected(
self, catalog_conn: sqlite3.Connection, captured_on: str
) -> None:
_subject(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO captures (capture_id, subject_id, captured_on)"
" VALUES ('C000001', 'S000001', ?)",
(captured_on,),
)
def test_captured_on_may_be_null(self, catalog_conn: sqlite3.Connection) -> None:
_subject(catalog_conn)
_capture(catalog_conn)
row = catalog_conn.execute("SELECT captured_on FROM captures").fetchone()
assert row[0] is None
@pytest.mark.parametrize("name", ["Tongue", "1tongue", "tongue-color", "_tongue", ""])
def test_invalid_dataset_name_is_rejected(
self, catalog_conn: sqlite3.Connection, name: str
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO datasets (name, created_at) VALUES (?, ?)", (name, TS)
)
def test_dataset_version_starts_from_one(self, catalog_conn: sqlite3.Connection) -> None:
catalog_conn.execute("INSERT INTO datasets (name, created_at) VALUES ('ds', ?)", (TS,))
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_versions (name, version, definition, hub_commit,"
" manifest_sha256, created_at) VALUES ('ds', 0, '', 'c', ?, ?)",
(SHA_M, TS),
)
@pytest.mark.parametrize("verified", [2, -1])
def test_delivery_verified_must_be_boolean(
self, catalog_conn: sqlite3.Connection, verified: int
) -> None:
_frozen_version(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO deliveries (name, version, server, path, delivered_at, verified)"
" VALUES ('tongue_color', 1, 'newton', '~/data/tiasdata/tongue_color/v1', ?, ?)",
(TS, verified),
)
def test_delivery_is_recorded(self, catalog_conn: sqlite3.Connection) -> None:
_frozen_version(catalog_conn)
catalog_conn.execute(
"INSERT INTO deliveries (name, version, server, path, delivered_at, verified)"
" VALUES ('tongue_color', 1, 'newton', '~/data/tiasdata/tongue_color/v1', ?, 1)",
(TS,),
)
# ---------------------------------------------------------------------------
# ID・名前の列は NULL を受け入れない
# （SQLite では INTEGER 以外の PRIMARY KEY が NULL を受け入れるため，NOT NULL が要る）
# ---------------------------------------------------------------------------
class TestPrimaryKeyNotNull:
@pytest.mark.parametrize(
("needs_subject", "sql", "params"),
[
(
False,
"INSERT INTO kinds (kind, category, description) VALUES (?, 'asset', 'd')",
(None,),
),
(False, "INSERT INTO subjects (subject_id, created_at) VALUES (?, ?)", (None, TS)),
(
True,
"INSERT INTO captures (capture_id, subject_id) VALUES (?, 'S000001')",
(None,),
),
(
False,
"INSERT INTO assets (sha256, kind, capture_id, ext, size_bytes, source,"
" source_name, registered_at) VALUES (?, 'color_chart', NULL, 'png', 10,"
" 'dummy_source', 'dummy.png', ?)",
(None, TS),
),
(False, "INSERT INTO datasets (name, created_at) VALUES (?, ?)", (None, TS)),
],
ids=[
"kinds.kind",
"subjects.subject_id",
"captures.capture_id",
"assets.sha256",
"datasets.name",
],
)
def test_null_key_is_rejected(
self,
catalog_conn: sqlite3.Connection,
needs_subject: bool,
sql: str,
params: tuple,
) -> None:
if needs_subject:
_subject(catalog_conn)
table = sql.split()[2]
before = catalog_conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
catalog_conn.execute(sql, params)
assert catalog_conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == before
def test_null_patient_key_is_rejected(self, identity_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
identity_conn.execute(
"INSERT INTO identity_map (patient_key, subject_id, created_at) VALUES (?, ?, ?)",
(None, "S000001", TS),
)
assert identity_conn.execute("SELECT COUNT(*) FROM identity_map").fetchone()[0] == 0
# ---------------------------------------------------------------------------
# 外部キー
# ---------------------------------------------------------------------------
class TestForeignKeys:
def test_capture_requires_existing_subject(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
_capture(catalog_conn, subject_id="S999999")
def test_asset_requires_existing_capture(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
_asset(catalog_conn, capture_id="C999999")
def test_asset_without_capture_is_allowed(self, catalog_conn: sqlite3.Connection) -> None:
_asset(catalog_conn, capture_id=None, kind="color_chart")
def test_asset_requires_existing_kind(self, catalog_conn: sqlite3.Connection) -> None:
_subject(catalog_conn)
_capture(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_asset(catalog_conn, kind="no_such_kind")
def test_annotation_requires_existing_capture(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, capture_id="C999999")
def test_annotation_requires_existing_asset(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, capture_id=None, sha256=SHA_B)
def test_derivation_links_require_existing_rows(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
cur = catalog_conn.execute(
"INSERT INTO derivations (tool, tool_version, mode) VALUES"
" ('DeepTIAS', 'unknown', 'Image&Calib')"
)
derivation_id = cur.lastrowid
catalog_conn.execute(
"INSERT INTO derivation_outputs (derivation_id, sha256, role) VALUES (?, ?, 'cropped')",
(derivation_id, SHA_A),
)
for table in ("derivation_inputs", "derivation_outputs"):
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
f"INSERT INTO {table} (derivation_id, sha256, role) VALUES (?, ?, 'x')",
(derivation_id, SHA_B),
)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
f"INSERT INTO {table} (derivation_id, sha256, role) VALUES (9999, ?, 'x')",
(SHA_A,),
)
def test_dataset_version_requires_existing_dataset(
self, catalog_conn: sqlite3.Connection
) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_versions (name, version, definition, hub_commit,"
" manifest_sha256, created_at) VALUES ('nope', 1, '', 'c', ?, ?)",
(SHA_M, TS),
)
def test_dataset_member_requires_existing_version_and_asset(
self, catalog_conn: sqlite3.Connection
) -> None:
_frozen_version(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_members (name, version, sha256) VALUES ('tongue_color', 2, ?)",
(SHA_A,),
)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_members (name, version, sha256) VALUES ('tongue_color', 1, ?)",
(SHA_B,),
)
def test_member_annotation_requires_member_and_annotation(
self, catalog_conn: sqlite3.Connection
) -> None:
annotation_id = _frozen_version(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_member_annotations (name, version, sha256, annotation_id)"
" VALUES ('tongue_color', 1, ?, ?)",
(SHA_B, annotation_id),
)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_member_annotations (name, version, sha256, annotation_id)"
" VALUES ('tongue_color', 1, ?, 9999)",
(SHA_A,),
)
def test_delivery_requires_existing_version(self, catalog_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO deliveries (name, version, server, path, delivered_at, verified)"
" VALUES ('nope', 1, 'newton', 'p', ?, 0)",
(TS,),
)
def test_subject_with_captures_cannot_be_deleted(
self, catalog_conn: sqlite3.Connection
) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute("DELETE FROM subjects WHERE subject_id = 'S000001'")
# ---------------------------------------------------------------------------
# 種類の分類の整合
# ---------------------------------------------------------------------------
class TestKindCategory:
def test_asset_with_annotation_kind_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
_subject(catalog_conn)
_capture(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_asset(catalog_conn, kind="tongue_color")
def test_asset_kind_cannot_be_changed_to_annotation_kind(
self, catalog_conn: sqlite3.Connection
) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute("UPDATE assets SET kind = 'tongue_color'")
def test_annotation_with_asset_kind_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, kind="tongue_crop_masked")
# ---------------------------------------------------------------------------
# 所見（撮影かファイルのどちらか一方・書き換えない・訂正は superseded_by）
# ---------------------------------------------------------------------------
class TestAnnotations:
def test_annotation_on_capture_is_accepted(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
_annotation(catalog_conn, capture_id="C000001")
def test_annotation_on_asset_is_accepted(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
_annotation(catalog_conn, capture_id=None, sha256=SHA_A)
def test_annotation_on_both_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, capture_id="C000001", sha256=SHA_A)
def test_annotation_on_neither_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, capture_id=None, sha256=None)
def test_annotation_value_must_be_json(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, value="{label: red")
def test_annotation_created_at_format(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
_annotation(catalog_conn, created_at="2026-10-06 00:00:00")
@pytest.mark.parametrize(
"assignment",
[
'value = \'{"label": "pale", "severity": "mild"}\'',
"annotator = 'A'",
"source = 'other'",
"created_at = '2026-10-07T00:00:00Z'",
"kind = 'tongue_color'",
"capture_id = NULL, sha256 = '" + SHA_A + "'",
],
ids=["value", "annotator", "source", "created_at", "same_kind", "target"],
)
def test_annotation_cannot_be_updated(
self, catalog_conn: sqlite3.Connection, assignment: str
) -> None:
_base(catalog_conn)
annotation_id = _annotation(catalog_conn)
catalog_conn.commit()
before = catalog_conn.execute("SELECT * FROM annotations").fetchall()
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
f"UPDATE annotations SET {assignment} WHERE annotation_id = ?", (annotation_id,)
)
assert catalog_conn.execute("SELECT * FROM annotations").fetchall() == before
def test_annotation_cannot_be_deleted(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
_annotation(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute("DELETE FROM annotations")
assert catalog_conn.execute("SELECT COUNT(*) FROM annotations").fetchone()[0] == 1
def test_correction_sets_superseded_by_once(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
old_id = _annotation(catalog_conn)
new_value = json.dumps({"label": "pale", "severity": "strong"})
new_id = _annotation(catalog_conn, value=new_value)
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?", (new_id, old_id)
)
old = catalog_conn.execute(
"SELECT value, superseded_by FROM annotations WHERE annotation_id = ?", (old_id,)
).fetchone()
# 古い行は値を保ったまま残る
assert old == (TONGUE_COLOR, new_id)
# 2 度目（別の行への付け替え・NULL に戻す）は拒否する
third_id = _annotation(catalog_conn, value=new_value)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?",
(third_id, old_id),
)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = NULL WHERE annotation_id = ?", (old_id,)
)
def test_superseded_by_must_point_to_later_row(self, catalog_conn: sqlite3.Connection) -> None:
_base(catalog_conn)
first = _annotation(catalog_conn)
second = _annotation(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?",
(first, second),
)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?",
(second, second),
)
def test_superseded_by_must_point_to_existing_row(
self, catalog_conn: sqlite3.Connection
) -> None:
_base(catalog_conn)
annotation_id = _annotation(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = 9999 WHERE annotation_id = ?",
(annotation_id,),
)
def test_correction_and_value_change_together_is_rejected(
self, catalog_conn: sqlite3.Connection
) -> None:
_base(catalog_conn)
old_id = _annotation(catalog_conn)
new_id = _annotation(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ?, value = '{}' WHERE annotation_id = ?",
(new_id, old_id),
)
# ---------------------------------------------------------------------------
# 凍結した版
# ---------------------------------------------------------------------------
class TestFrozenVersions:
@pytest.mark.parametrize(
"statement",
[
"UPDATE dataset_versions SET note = 'changed'",
"UPDATE dataset_versions SET manifest_sha256 = '" + "d" * 64 + "'",
"DELETE FROM dataset_versions",
"UPDATE dataset_members SET sha256 = '" + SHA_B + "'",
"DELETE FROM dataset_members",
"UPDATE dataset_member_annotations SET annotation_id = annotation_id",
"DELETE FROM dataset_member_annotations",
],
)
def test_frozen_rows_cannot_be_changed(
self, catalog_conn: sqlite3.Connection, statement: str
) -> None:
_frozen_version(catalog_conn)
tables = ("dataset_versions", "dataset_members", "dataset_member_annotations")
before = {t: catalog_conn.execute(f"SELECT * FROM {t}").fetchall() for t in tables}
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(statement)
after = {t: catalog_conn.execute(f"SELECT * FROM {t}").fetchall() for t in tables}
assert after == before
def test_new_version_can_be_added(self, catalog_conn: sqlite3.Connection) -> None:
_frozen_version(catalog_conn)
catalog_conn.execute(
"INSERT INTO dataset_versions (name, version, definition, hub_commit, manifest_sha256,"
" created_at) VALUES ('tongue_color', 2, 'def', 'abc124', ?, ?)",
(SHA_M, TS),
)
def test_duplicate_version_is_rejected(self, catalog_conn: sqlite3.Connection) -> None:
_frozen_version(catalog_conn)
with pytest.raises(sqlite3.IntegrityError):
catalog_conn.execute(
"INSERT INTO dataset_versions (name, version, definition, hub_commit,"
" manifest_sha256, created_at) VALUES ('tongue_color', 1, 'x', 'y', ?, ?)",
(SHA_M, TS),
)
def test_annotation_used_by_version_survives_correction(
self, catalog_conn: sqlite3.Connection
) -> None:
# 訂正しても過去の版が古い行（古い値）を参照し続けられる
old_id = _frozen_version(catalog_conn)
new_id = _annotation(
catalog_conn, value=json.dumps({"label": "pale", "severity": "strong"})
)
catalog_conn.execute(
"UPDATE annotations SET superseded_by = ? WHERE annotation_id = ?", (new_id, old_id)
)
row = catalog_conn.execute(
"SELECT a.value FROM dataset_member_annotations m"
" JOIN annotations a ON a.annotation_id = m.annotation_id"
" WHERE m.name = 'tongue_color' AND m.version = 1"
).fetchone()
assert row == (TONGUE_COLOR,)
# ---------------------------------------------------------------------------
# 対応表
# ---------------------------------------------------------------------------
class TestIdentityMap:
def _insert(self, conn: sqlite3.Connection, patient_key: str, subject_id: str) -> None:
conn.execute(
"INSERT INTO identity_map (patient_key, subject_id, created_at) VALUES (?, ?, ?)",
(patient_key, subject_id, TS),
)
def test_mapping_is_accepted(self, identity_conn: sqlite3.Connection) -> None:
self._insert(identity_conn, "DUMMY-P1", "S000001")
self._insert(identity_conn, "DUMMY-P2", "S000002")
def test_same_patient_twice_is_rejected(self, identity_conn: sqlite3.Connection) -> None:
self._insert(identity_conn, "DUMMY-P1", "S000001")
with pytest.raises(sqlite3.IntegrityError):
self._insert(identity_conn, "DUMMY-P1", "S000002")
def test_subject_shared_by_two_patients_is_rejected(
self, identity_conn: sqlite3.Connection
) -> None:
self._insert(identity_conn, "DUMMY-P1", "S000001")
with pytest.raises(sqlite3.IntegrityError):
self._insert(identity_conn, "DUMMY-P2", "S000001")
@pytest.mark.parametrize("subject_id", ["S1", "C000001", "s000001"])
def test_invalid_subject_id_is_rejected(
self, identity_conn: sqlite3.Connection, subject_id: str
) -> None:
with pytest.raises(sqlite3.IntegrityError):
self._insert(identity_conn, "DUMMY-P1", subject_id)
def test_empty_patient_key_is_rejected(self, identity_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
self._insert(identity_conn, "", "S000001")
def test_invalid_created_at_is_rejected(self, identity_conn: sqlite3.Connection) -> None:
with pytest.raises(sqlite3.IntegrityError):
identity_conn.execute(
"INSERT INTO identity_map (patient_key, subject_id, created_at)"
" VALUES ('DUMMY-P1', 'S000001', '2026-10-06')"
)
