"""バンドルの読み込み・検証（SPEC_03「バンドル」「ライブラリの API」，SPEC_01「使う側からの利用」）.
必須対象 5（バンドル）の読み込み側: 目録なしで `open_bundle`・`load` で読めること，
manifest.csv の SHA-256 の不一致・ファイルの欠け・中身の改変を検出すること．
バンドルは実装の組み立て関数を使わず，仕様の形（dataset.json・manifest.csv・SHA256SUMS・
files/<sha256>.<ext>）どおりにテストの中で手で作る（読み込み側を仕様の形に対して確かめるため）．
書き出しからの流れは `test_export.py`・`test_cli.py` で確かめる．臨床データは使わない．
"""
import hashlib
import json
import os
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path
import pytest
import tiasdata
from tiasdata import bundle as bundle_module
from tiasdata.bundle import Bundle, BundleRow, load, open_bundle
NAME = "tongue_color"
CREATED_AT = "2026-10-06T09:00:00Z"
COMMIT = "0123456789abcdef0123456789abcdef01234567"
HEADER = "path,sha256,kind,subject_id,capture_id,tongue_color.label,tongue_color.severity"
# 架空の患者 ID（エラーメッセージに出ないことを確かめるため，バンドルの外のフォルダ名に仕込む）
LEAK_KEY = "990077"
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
@dataclass
class Item:
"""手で作るバンドルの 1 ファイル．"""
content: bytes
subject_id: str
capture_id: str
label: str = "red"
severity: str = "mild"
ext: str = "png"
path: str | None = None # manifest の path を差し替えるとき
sha256: str | None = None # manifest の sha256 を差し替えるとき
@property
def real_sha256(self) -> str:
return _sha(self.content)
@property
def manifest_sha256_cell(self) -> str:
return self.real_sha256 if self.sha256 is None else self.sha256
@property
def manifest_path(self) -> str:
return self.path or f"files/{self.real_sha256}.{self.ext}"
def default_items() -> list[Item]:
# subject_id・capture_id・sha256 の順に並ぶように並べてある
return [
Item(b"dummy-image-1", "S000001", "C000001", "red", "mild"),
Item(b"dummy-image-2", "S000001", "C000002", "light_red", "normal"),
Item(b"dummy-image-3", "S000002", "C000003", "pale", "strong"),
]
@dataclass
class Built:
root: Path
items: list[Item]
manifest: bytes
info: dict = field(default_factory=dict)
def build_manifest_bytes(items: list[Item], header: str = HEADER) -> bytes:
lines = [header]
for item in items:
lines.append(
",".join(
[
item.manifest_path,
item.manifest_sha256_cell,
"tongue_crop_masked",
item.subject_id,
item.capture_id,
item.label,
item.severity,
]
)
)
return ("\n".join(lines) + "\n").encode("utf-8")
def write_bundle(
root: Path,
items: list[Item] | None = None,
*,
name: str = NAME,
version: int = 1,
manifest: bytes | None = None,
info_overrides: dict | None = None,
write_files: bool = True,
) -> Built:
"""仕様の形どおりにバンドルを書く（改行は LF．bytes で書いて変換させない）．"""
items = default_items() if items is None else items
root.mkdir(parents=True, exist_ok=True)
data = build_manifest_bytes(items) if manifest is None else manifest
info = {
"format": 1,
"name": name,
"version": version,
"created_at": CREATED_AT,
"hub_commit": COMMIT,
"files": len(items),
"manifest_sha256": _sha(data),
}
info.update(info_overrides or {})
(root / "dataset.json").write_bytes(json.dumps(info, indent=2).encode("utf-8") + b"\n")
(root / "manifest.csv").write_bytes(data)
sums = "".join(
f"{item.real_sha256} files/{item.real_sha256}.{item.ext}\n"
for item in sorted(items, key=lambda i: f"files/{i.real_sha256}.{i.ext}")
)
(root / "SHA256SUMS").write_bytes(sums.encode("utf-8"))
if write_files:
(root / "files").mkdir(exist_ok=True)
for item in items:
(root / "files" / f"{item.real_sha256}.{item.ext}").write_bytes(item.content)
return Built(root=root, items=items, manifest=data, info=info)
@pytest.fixture
def bundles_root(tmp_path: Path) -> Path:
# パスに架空の患者 ID を含め，エラーメッセージにパスが出ればすぐ分かるようにする
return tmp_path / f"患者{LEAK_KEY}" / "bundles"
@pytest.fixture
def built(bundles_root: Path) -> Built:
return write_bundle(bundles_root / NAME / "v1")
def _assert_safe_message(message: str, tmp_path: Path) -> None:
# エラーメッセージにバンドルのパス・データの場所・患者 ID を出さない（SPEC_03「出力の規則」）
for form in (str(tmp_path), tmp_path.as_posix(), str(tmp_path.resolve())):
assert form not in message, message
assert LEAK_KEY not in message, message
assert "bundles" not in message, message
def _open_error(path: Path, tmp_path: Path) -> str:
with pytest.raises(ValueError) as exc:
open_bundle(path)
message = str(exc.value)
_assert_safe_message(message, tmp_path)
return message
def _verify_error(b: Bundle, tmp_path: Path) -> str:
with pytest.raises(ValueError) as exc:
b.verify()
message = str(exc.value)
_assert_safe_message(message, tmp_path)
return message
# ---------------------------------------------------------------------------
# 目録なしで読めること（SPEC_01「依存の向き」）
# ---------------------------------------------------------------------------
_NO_CATALOG_SCRIPT = """
import sys
import tiasdata
b = tiasdata.open_bundle(sys.argv[1])
b.verify()
b2 = tiasdata.load("tongue_color", version=1)
assert b2.fingerprint == b.fingerprint
for row in b.rows:
assert b.file_path(row).is_file()
heavy = sorted(
m for m in sys.modules
if m in ("sqlite3", "_sqlite3", "tiasdata.catalog", "tiasdata.store", "tiasdata.config",
"tiasdata.datasets", "tiasdata.export", "tiasdata.cli", "openpyxl")
or m.startswith(("tiasdata.catalog.", "tiasdata.importers"))
)
print("LOADED=" + ",".join(heavy))
print("ROWS=" + str(len(b.rows)))
"""
class TestWithoutCatalog:
def test_import_tiasdata_does_not_load_catalog_store_config_or_sqlite(self) -> None:
result = subprocess.run(
[
sys.executable,
"-c",
"import sys, tiasdata\n"
"names = ('sqlite3', '_sqlite3', 'tiasdata.catalog', 'tiasdata.store',"
" 'tiasdata.config', 'tiasdata.export', 'tiasdata.datasets')\n"
"print(','.join(n for n in names if n in sys.modules))\n"
"print(sorted(n for n in tiasdata.__all__))",
],
capture_output=True,
text=True,
check=False,
timeout=60,
)
assert result.returncode == 0, result.stderr
loaded, exported = result.stdout.splitlines()
assert loaded == ""
for api in ("load", "open_bundle", "Bundle", "BundleRow", "BundleInfo"):
assert api in exported
def test_open_verify_and_load_in_fresh_process_without_catalog(
self, built: Built, bundles_root: Path, tmp_path: Path
) -> None:
env = {k: v for k, v in os.environ.items() if k != "TIASDATA_CONFIG"}
env["TIASDATA_BUNDLES"] = str(bundles_root)
# 目録・設定ファイルが無い場所で動かす
result = subprocess.run(
[sys.executable, "-c", _NO_CATALOG_SCRIPT, str(built.root)],
capture_output=True,
text=True,
env=env,
cwd=tmp_path,
check=False,
timeout=60,
)
assert result.returncode == 0, result.stderr
assert result.stdout.splitlines() == ["LOADED=", "ROWS=3"]
def test_public_api_is_exported_from_package(self) -> None:
assert tiasdata.load is load
assert tiasdata.open_bundle is open_bundle
assert tiasdata.Bundle is Bundle
assert tiasdata.BundleRow is BundleRow
# ---------------------------------------------------------------------------
# open_bundle（正常系）
# ---------------------------------------------------------------------------
class TestOpenBundle:
def test_rows_follow_manifest(self, built: Built) -> None:
b = open_bundle(built.root)
assert len(b.rows) == 3
for row, item in zip(b.rows, built.items, strict=True):
assert isinstance(row, BundleRow)
assert row.path == f"files/{item.real_sha256}.png"
assert row.sha256 == item.real_sha256
assert row.kind == "tongue_crop_masked"
assert row.subject_id == item.subject_id
assert row.capture_id == item.capture_id
assert row.annotations == {
"tongue_color": {"label": item.label, "severity": item.severity}
}
def test_accepts_str_path(self, built: Built) -> None:
assert len(open_bundle(str(built.root)).rows) == 3
def test_info_and_fingerprint(self, built: Built) -> None:
b = open_bundle(built.root)
assert b.name == NAME
assert b.version == 1
assert b.manifest_sha256 == _sha(built.manifest)
assert b.info.created_at == CREATED_AT
assert b.info.hub_commit == COMMIT
assert b.info.files == 3
assert b.fingerprint == {
"name": NAME,
"version": 1,
"manifest_sha256": _sha(built.manifest),
}
def test_fingerprint_is_a_fresh_dict(self, built: Built) -> None:
b = open_bundle(built.root)
first = b.fingerprint
first["version"] = 99
assert b.fingerprint["version"] == 1
def test_file_path_points_to_file_in_bundle(self, built: Built) -> None:
b = open_bundle(built.root)
for row, item in zip(b.rows, built.items, strict=True):
path = b.file_path(row)
assert path.is_absolute()
assert path.read_bytes() == item.content
assert path.resolve().is_relative_to(built.root.resolve())
def test_rows_are_immutable(self, built: Built) -> None:
b = open_bundle(built.root)
with pytest.raises(TypeError):
b.rows[0] = b.rows[1] # type: ignore[index]
with pytest.raises(Exception): # noqa: B017,PT011 pydantic の frozen
b.rows[0].subject_id = "S999999" # type: ignore[misc]
def test_repr_has_no_path(self, built: Built, tmp_path: Path) -> None:
_assert_safe_message(repr(open_bundle(built.root)), tmp_path)
def test_empty_annotation_cell_is_none(self, bundles_root: Path) -> None:
items = default_items()
items[1].severity = ""
b = open_bundle(write_bundle(bundles_root / NAME / "v1", items).root)
assert b.rows[1].annotations == {"tongue_color": {"label": "light_red", "severity": None}}
assert b.rows[0].annotations["tongue_color"]["severity"] == "mild"
def test_annotation_values_are_strings(self, bundles_root: Path) -> None:
# 文字列以外の値も manifest の表記（文字列）のまま返す（ユーザー判断⑤）
items = default_items()
items[0].severity = "3"
b = open_bundle(write_bundle(bundles_root / NAME / "v1", items).root)
assert b.rows[0].annotations["tongue_color"]["severity"] == "3"
def test_annotations_grouped_by_kind(self, bundles_root: Path) -> None:
item = Item(b"x", "S000001", "C000001")
header = (
"path,sha256,kind,subject_id,capture_id,"
"tongue_color.label,tongue_coating.thickness,tongue_color.severity"
)
line = f"files/{item.real_sha256}.png,{item.real_sha256},k,S000001,C000001,red,,mild"
data = f"{header}\n{line}\n".encode()
b = open_bundle(write_bundle(bundles_root / NAME / "v1", [item], manifest=data).root)
assert b.rows[0].annotations == {
"tongue_color": {"label": "red", "severity": "mild"},
"tongue_coating": {"thickness": None},
}
def test_manifest_without_annotation_columns(self, bundles_root: Path) -> None:
item = Item(b"x", "S000001", "C000001")
data = (
"path,sha256,kind,subject_id,capture_id\n"
f"files/{item.real_sha256}.png,{item.real_sha256},k,S000001,C000001\n"
).encode()
b = open_bundle(write_bundle(bundles_root / NAME / "v1", [item], manifest=data).root)
assert b.rows[0].annotations == {}
def test_csv_quoted_values(self, bundles_root: Path) -> None:
# 値に , や " を含む所見（JSON の表記等）も CSV として読む
item = Item(b"x", "S000001", "C000001")
header = "path,sha256,kind,subject_id,capture_id,tongue_color.note"
line = f'files/{item.real_sha256}.png,{item.real_sha256},k,S000001,C000001,"[1,""a""]"'
data = f"{header}\n{line}\n".encode()
b = open_bundle(write_bundle(bundles_root / NAME / "v1", [item], manifest=data).root)
assert b.rows[0].annotations == {"tongue_color": {"note": '[1,"a"]'}}
def test_unknown_keys_in_dataset_json_are_ignored(self, bundles_root: Path) -> None:
built = write_bundle(bundles_root / NAME / "v1", info_overrides={"future_key": "x"})
assert open_bundle(built.root).version == 1
def test_zero_rows_bundle_opens(self, bundles_root: Path) -> None:
built = write_bundle(bundles_root / NAME / "v1", [], manifest=(HEADER + "\n").encode())
b = open_bundle(built.root)
assert b.rows == ()
b.verify()
def test_sha256sums_with_binary_marker_verifies(self, built: Built) -> None:
# `sha256sum -b` の形（`<sha256> *<パス>`）も受け付ける
text = (built.root / "SHA256SUMS").read_bytes().decode()
(built.root / "SHA256SUMS").write_bytes(text.replace(" files/", " *files/").encode())
open_bundle(built.root).verify()
# ---------------------------------------------------------------------------
# open_bundle（開けないもの）
# ---------------------------------------------------------------------------
class TestOpenBundleRejects:
def test_missing_folder(self, bundles_root: Path, tmp_path: Path) -> None:
_open_error(bundles_root / NAME / "v9", tmp_path)
def test_path_is_file(self, built: Built, tmp_path: Path) -> None:
_open_error(built.root / "manifest.csv", tmp_path)
@pytest.mark.parametrize("filename", ["dataset.json", "manifest.csv"])
def test_missing_metadata_file(self, built: Built, tmp_path: Path, filename: str) -> None:
(built.root / filename).unlink()
_open_error(built.root, tmp_path)
def test_metadata_file_is_directory(self, built: Built, tmp_path: Path) -> None:
(built.root / "manifest.csv").unlink()
(built.root / "manifest.csv").mkdir()
_open_error(built.root, tmp_path)
def test_tampered_manifest_stops(self, built: Built, tmp_path: Path) -> None:
# 所見の値を書き換えると SHA-256 が dataset.json と一致しない
data = built.manifest.replace(b",pale,strong", b",red,strong")
assert data != built.manifest
(built.root / "manifest.csv").write_bytes(data)
message = _open_error(built.root, tmp_path)
assert "SHA-256" in message
def test_manifest_with_crlf_stops(self, built: Built, tmp_path: Path) -> None:
# 改行を CRLF に変えたもの（Windows での変換等）も別の中身として止める
(built.root / "manifest.csv").write_bytes(built.manifest.replace(b"\n", b"\r\n"))
_open_error(built.root, tmp_path)
def test_manifest_sha256_in_dataset_json_wrong(
self, bundles_root: Path, tmp_path: Path
) -> None:
built = write_bundle(
bundles_root / NAME / "v1", info_overrides={"manifest_sha256": "0" * 64}
)
_open_error(built.root, tmp_path)
@pytest.mark.parametrize(
"overrides",
[
{"format": 2},
{"format": "1"},
{"files": 2},
{"files": 4},
{"version": 0},
{"version": "1"},
{"name": "Bad-Name"},
{"name": "../x"},
{"manifest_sha256": "XYZ"},
],
ids=lambda o: "-".join(f"{k}={v}" for k, v in o.items()),
)
def test_invalid_dataset_json_values(
self, bundles_root: Path, tmp_path: Path, overrides: dict
) -> None:
built = write_bundle(bundles_root / NAME / "v1", info_overrides=overrides)
_open_error(built.root, tmp_path)
@pytest.mark.parametrize(
"key", ["format", "name", "version", "created_at", "hub_commit", "files", "manifest_sha256"]
)
def test_missing_dataset_json_key(self, built: Built, tmp_path: Path, key: str) -> None:
info = dict(built.info)
del info[key]
(built.root / "dataset.json").write_bytes(json.dumps(info).encode())
_open_error(built.root, tmp_path)
@pytest.mark.parametrize("content", [b"{not json", b"[1, 2]", b"\xff\xfe", b""])
def test_dataset_json_not_object(self, built: Built, tmp_path: Path, content: bytes) -> None:
(built.root / "dataset.json").write_bytes(content)
_open_error(built.root, tmp_path)
@pytest.mark.parametrize(
"bad_path",
[
"../{sha}.png",
"files/../../{sha}.png",
"files/sub/{sha}.png",
"files\\{sha}.png",
"/files/{sha}.png",
"C:/files/{sha}.png",
"{sha}.png",
"files/{sha}",
"files/{sha}.PNG",
"files/{other}.png",
"other/{sha}.png",
],
)
def test_invalid_path_shape(self, bundles_root: Path, tmp_path: Path, bad_path: str) -> None:
items = default_items()
items[0].path = bad_path.format(sha=items[0].real_sha256, other=items[1].real_sha256)
built = write_bundle(bundles_root / NAME / "v1", items)
_open_error(built.root, tmp_path)
@pytest.mark.parametrize("bad_sha", ["", "abc", "A" * 64, "g" * 64])
def test_invalid_sha256_cell(self, bundles_root: Path, tmp_path: Path, bad_sha: str) -> None:
items = default_items()
items[0].sha256 = bad_sha
built = write_bundle(bundles_root / NAME / "v1", items)
_open_error(built.root, tmp_path)
def test_duplicate_sha256_rows(self, bundles_root: Path, tmp_path: Path) -> None:
items = default_items()
items.append(Item(items[0].content, "S000003", "C000009"))
built = write_bundle(bundles_root / NAME / "v1", items)
_open_error(built.root, tmp_path)
@pytest.mark.parametrize(
"header",
[
"sha256,path,kind,subject_id,capture_id,tongue_color.label,tongue_color.severity",
"path,sha256,kind,subject_id,tongue_color.label,tongue_color.severity",
"path,sha256,kind,subject_id,capture_id,label,tongue_color.severity",
"path,sha256,kind,subject_id,capture_id,tongue_color.label,tongue_color.label",
],
)
def test_invalid_header(self, bundles_root: Path, tmp_path: Path, header: str) -> None:
items = default_items()
data = build_manifest_bytes(items, header=header)
built = write_bundle(bundles_root / NAME / "v1", items, manifest=data)
_open_error(built.root, tmp_path)
def test_empty_manifest(self, bundles_root: Path, tmp_path: Path) -> None:
built = write_bundle(bundles_root / NAME / "v1", [], manifest=b"")
_open_error(built.root, tmp_path)
def test_row_with_wrong_column_count(self, bundles_root: Path, tmp_path: Path) -> None:
items = default_items()
data = build_manifest_bytes(items).replace(b",pale,strong", b",pale")
built = write_bundle(bundles_root / NAME / "v1", items, manifest=data)
_open_error(built.root, tmp_path)
def test_manifest_not_utf8(self, bundles_root: Path, tmp_path: Path) -> None:
items = default_items()
data = build_manifest_bytes(items).replace(b"red,mild", b"\x82\xa0,mild")
built = write_bundle(bundles_root / NAME / "v1", items, manifest=data)
_open_error(built.root, tmp_path)
# ---------------------------------------------------------------------------
# verify
# ---------------------------------------------------------------------------
class TestVerify:
def test_intact_bundle_returns_none(self, built: Built) -> None:
assert open_bundle(built.root).verify() is None
def test_detects_modified_file(self, built: Built, tmp_path: Path) -> None:
b = open_bundle(built.root)
target = b.file_path(b.rows[1])
target.write_bytes(target.read_bytes() + b"tampered")
message = _verify_error(b, tmp_path)
assert b.rows[1].sha256[:12] in message
def test_detects_missing_file(self, built: Built, tmp_path: Path) -> None:
b = open_bundle(built.root)
b.file_path(b.rows[0]).unlink()
message = _verify_error(b, tmp_path)
assert b.rows[0].sha256[:12] in message
def test_detects_missing_files_folder(self, built: Built, tmp_path: Path) -> None:
shutil.rmtree(built.root / "files")
_verify_error(open_bundle(built.root), tmp_path)
def test_detects_extra_file(self, built: Built, tmp_path: Path) -> None:
(built.root / "files" / f"{_sha(b'extra')}.png").write_bytes(b"extra")
_verify_error(open_bundle(built.root), tmp_path)
def test_detects_manifest_changed_after_open(self, built: Built, tmp_path: Path) -> None:
b = open_bundle(built.root)
(built.root / "manifest.csv").write_bytes(built.manifest.replace(b"pale", b"red"))
_verify_error(b, tmp_path)
def test_detects_missing_sha256sums(self, built: Built, tmp_path: Path) -> None:
(built.root / "SHA256SUMS").unlink()
_verify_error(open_bundle(built.root), tmp_path)
@pytest.mark.parametrize(
"change",
["drop_line", "duplicate_line", "extra_line", "wrong_hash", "bad_format"],
)
def test_detects_sha256sums_not_matching_manifest(
self, built: Built, tmp_path: Path, change: str
) -> None:
path = built.root / "SHA256SUMS"
lines = path.read_bytes().decode().splitlines(keepends=True)
if change == "drop_line":
lines = lines[1:]
elif change == "duplicate_line":
lines = [*lines, lines[0]]
elif change == "extra_line":
extra = _sha(b"extra")
lines = [*lines, f"{extra} files/{extra}.png\n"]
elif change == "wrong_hash":
sha = lines[0][:64]
lines[0] = lines[0].replace(sha, "0" * 64, 1)
else:
lines[0] = lines[0].replace(" ", "\t", 1)
path.write_bytes("".join(lines).encode())
_verify_error(open_bundle(built.root), tmp_path)
def test_sha256sums_order_does_not_matter(self, built: Built) -> None:
path = built.root / "SHA256SUMS"
lines = path.read_bytes().decode().splitlines(keepends=True)
path.write_bytes("".join(reversed(lines)).encode())
open_bundle(built.root).verify()
def test_verify_message_counts(self, built: Built, tmp_path: Path) -> None:
b = open_bundle(built.root)
b.file_path(b.rows[0]).unlink()
b.file_path(b.rows[1]).write_bytes(b"changed")
message = _verify_error(b, tmp_path)
assert "欠け 1 件" in message
assert "不一致 1 件" in message
@pytest.mark.skipif(shutil.which("sha256sum") is None, reason="sha256sum が無い")
def test_sha256sums_checks_with_sha256sum_command(self, built: Built) -> None:
# サーバで `sha256sum -c SHA256SUMS` が通る形であること（SPEC_03「受け渡し」手順 4）
result = subprocess.run(
["sha256sum", "-c", "--quiet", "SHA256SUMS"],
cwd=built.root,
capture_output=True,
check=False,
timeout=60,
)
assert result.returncode == 0, result.stdout + result.stderr
# ---------------------------------------------------------------------------
# load
# ---------------------------------------------------------------------------
class TestLoad:
def test_load_from_env_var(
self, built: Built, bundles_root: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
b = load(NAME, version=1)
assert b.fingerprint == open_bundle(built.root).fingerprint
assert b.root.resolve() == built.root.resolve()
assert len(b.rows) == 3
def test_load_default_is_home_data_tiasdata(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
write_bundle(home / "data" / "tiasdata" / NAME / "v2", version=2)
monkeypatch.delenv("TIASDATA_BUNDLES", raising=False)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
b = load(NAME, version=2)
assert b.version == 2
assert b.root.resolve() == (home / "data" / "tiasdata" / NAME / "v2").resolve()
def test_empty_env_var_falls_back_to_default(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
home = tmp_path / "home"
write_bundle(home / "data" / "tiasdata" / NAME / "v1")
monkeypatch.setenv("TIASDATA_BUNDLES", "")
monkeypatch.setenv("HOME", str(home))
monkeypatch.setenv("USERPROFILE", str(home))
assert load(NAME, version=1).version == 1
def test_version_is_keyword_only(
self, built: Built, bundles_root: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(TypeError):
load(NAME, 1) # type: ignore[misc]
with pytest.raises(TypeError):
load(NAME) # type: ignore[call-arg]
@pytest.mark.parametrize("version", [0, -1, True, 1.0, "1", None])
def test_invalid_version(
self,
built: Built,
bundles_root: Path,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
version: object,
) -> None:
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(ValueError) as exc:
load(NAME, version=version) # type: ignore[arg-type]
_assert_safe_message(str(exc.value), tmp_path)
@pytest.mark.parametrize("name", ["", "Tongue", "../tongue_color", "a/b", "a\\b", "1abc", "."])
def test_invalid_name(
self, bundles_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, name: str
) -> None:
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(ValueError) as exc:
load(name, version=1)
_assert_safe_message(str(exc.value), tmp_path)
def test_missing_bundle(
self, built: Built, bundles_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(ValueError) as exc:
load(NAME, version=2)
_assert_safe_message(str(exc.value), tmp_path)
with pytest.raises(ValueError):
load("other", version=1)
@pytest.mark.parametrize(("json_name", "json_version"), [("other", 1), (NAME, 2)])
def test_dataset_json_disagrees_with_request(
self,
bundles_root: Path,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
json_name: str,
json_version: int,
) -> None:
# 置き場所のフォルダ名と dataset.json の名前・版が違う（別の版のフォルダを写した等）
write_bundle(bundles_root / NAME / "v1", name=json_name, version=json_version)
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(ValueError) as exc:
load(NAME, version=1)
_assert_safe_message(str(exc.value), tmp_path)
def test_load_rejects_tampered_manifest(
self, built: Built, bundles_root: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
(built.root / "manifest.csv").write_bytes(built.manifest + b"\n")
monkeypatch.setenv("TIASDATA_BUNDLES", str(bundles_root))
with pytest.raises(ValueError) as exc:
load(NAME, version=1)
_assert_safe_message(str(exc.value), tmp_path)
# ---------------------------------------------------------------------------
# SHA256SUMS の読み書き（`sha256sum -c` の形）
# ---------------------------------------------------------------------------
class TestSha256sumsFormat:
def test_build_is_two_spaces_lf_sorted_by_path(self) -> None:
a, b = "a" * 64, "b" * 64
data = bundle_module.build_sha256sums([(b, f"files/{b}.png"), (a, f"files/{a}.jpg")])
assert data == f"{a} files/{a}.jpg\n{b} files/{b}.png\n".encode()
assert b"\r" not in data
def test_parse_roundtrip(self) -> None:
a = "a" * 64
entries = [(a, f"files/{a}.png")]
assert bundle_module.parse_sha256sums(bundle_module.build_sha256sums(entries)) == entries
