"""テスト共通の fixture（ダミーデータのみ．臨床データは使わない）."""
import sqlite3
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
import pytest
from PIL import Image
from tiasdata import catalog
# 舌色データセットの 9 フォルダ（フォルダ名 → label・severity）．
# TIASJudgeModel の TEST_01「フォルダ構成とラベル対応」と同じ．フォルダの直下に <番号>.png を置く
TONGUE_COLOR_FOLDERS: dict[str, tuple[str, str]] = {
"淡紅（正常）": ("light_red", "normal"),
"暗赤紅＿あり": ("dark_red", "strong"),
"暗赤紅＿ややあり": ("dark_red", "mild"),
"紫＿ややあり": ("purple", "mild"),
"紅＿あり": ("red", "strong"),
"紅＿ややあり": ("red", "mild"),
"紫＿あり": ("purple", "strong"),
"淡白＿あり": ("pale", "strong"),
"淡白＿ややあり": ("pale", "mild"),
}
def write_dummy_png(path: Path, seed: int, size: tuple[int, int] = (8, 6)) -> Path:
"""`seed` ごとに中身の違う小さい PNG を書く（同じ `seed` なら同じバイト列）．"""
path.parent.mkdir(parents=True, exist_ok=True)
color = (seed * 37 % 256, seed * 59 % 256, seed * 83 % 256)
image = Image.new("RGB", size, color)
# 色だけだと seed が 256 周期で重なるため，1 画素に seed を埋めて中身を一意にする
image.putpixel((0, 0), (seed % 256, seed // 256 % 256, seed // 65536 % 256))
image.save(path, format="PNG")
return path
@dataclass(frozen=True)
class DummyImage:
"""ダミーの取り込み元の画像 1 枚．"""
number: int
folder: str
label: str
severity: str
path: Path
@dataclass(frozen=True)
class DummyTongueColorSource:
"""ダミーの舌色データセット（9 フォルダ × 数枚）．"""
root: Path
images: list[DummyImage]
@pytest.fixture
def make_png():
"""小さい PNG を書く関数（`make_png(path, seed)`）．"""
return write_dummy_png
@pytest.fixture
def store_dir(tmp_path: Path) -> Path:
"""空のストアのフォルダ（`tiasdata init` 後と同じ状態）．"""
directory = tmp_path / "tiasdata" / "store"
directory.mkdir(parents=True)
return directory
@pytest.fixture
def tongue_color_source(tmp_path: Path) -> DummyTongueColorSource:
"""舌色データセットと同じ構成のダミーの取り込み元（各フォルダ 2 枚・番号は 1 から連番）．
対応表（xlsx）は `tongue_color_patient_map` で作る．
"""
root = tmp_path / "source" / "TIASJudgeModel"
images: list[DummyImage] = []
number = 1
for folder, (label, severity) in TONGUE_COLOR_FOLDERS.items():
for _ in range(2):
path = write_dummy_png(root / folder / f"{number}.png", seed=number)
images.append(DummyImage(number, folder, label, severity, path))
number += 1
return DummyTongueColorSource(root=root, images=images)
def dummy_patient_key(number: int) -> str:
"""ダミーの取り込み元の画像の番号 → 架空の患者 ID．
7 人に割り振るので，18 枚のダミーでは同じ患者に 2〜3 枚が当たる（ラベルの違う画像も混ざる）．
値は架空（99 で始まる 6 桁）．
"""
return str(990001 + (number - 1) % 7)
def write_patient_map(
path: Path,
rows: list[tuple[object, ...]],
*,
header: tuple[object, ...] = ("番号", "何枚目", "患者ID"),
) -> Path:
"""対応表の xlsx を書く．
1 行目は見出し，以降は `rows`（列は 1: 番号・2: 何枚目・3: 患者 ID）．
"""
from openpyxl import Workbook
path.parent.mkdir(parents=True, exist_ok=True)
workbook = Workbook()
sheet = workbook.active
sheet.append(list(header))
for row in rows:
sheet.append(list(row))
workbook.save(path)
return path
@dataclass(frozen=True)
class DummyPatientMap:
"""ダミーの対応表（xlsx）．"""
path: Path
# 番号 → 架空の患者 ID（`dummy_patient_key` と同じ）
patient_keys: dict[int, str]
@pytest.fixture
def make_patient_map():
"""対応表の xlsx を書く関数（`make_patient_map(path, rows, header=...)`）．"""
return write_patient_map
@pytest.fixture
def tongue_color_patient_map(
tmp_path: Path, tongue_color_source: DummyTongueColorSource
) -> DummyPatientMap:
"""`tongue_color_source` の全画像の対応表（架空の患者 ID．同じ患者に複数枚）．
実データと同じく，患者 ID のセルは数値の行と文字列の行を混ぜる（偶数の番号は文字列）．
"""
rows: list[tuple[object, ...]] = []
patient_keys: dict[int, str] = {}
seen: dict[str, int] = {}
for image in tongue_color_source.images:
key = dummy_patient_key(image.number)
patient_keys[image.number] = key
seen[key] = seen.get(key, 0) + 1
cell: object = key if image.number % 2 == 0 else int(key)
rows.append((image.number, seen[key], cell))
path = write_patient_map(tmp_path / "source" / "patient_map.xlsx", rows)
return DummyPatientMap(path=path, patient_keys=patient_keys)
def write_reviews(reviews_dir: Path, source_name: str, numbers: list[int]) -> Path:
"""要確認リスト `<reviews_dir>/<source_name>.toml` を書く．"""
reviews_dir.mkdir(parents=True, exist_ok=True)
entries = "".join(
f'[[review]]\nnumber = {number}\nreason = "テスト用の要確認"\n'
f'resolve = "テスト用"\nsince = "2026-10-06"\n\n'
for number in numbers
)
path = reviews_dir / f"{source_name}.toml"
path.write_text(entries, encoding="utf-8")
return path
@pytest.fixture
def reviews_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
"""一時フォルダの要確認リストのフォルダ（リポジトリの `reviews/` の代わりに使わせる）．
`tiasdata.importers.reviews.default_reviews_dir` を差し替えるので，同じプロセスで呼ぶ
`run_import`・`cli.main` はこのフォルダを読む（subprocess で起動した CLI には効かない）．
要確認リストは `write_reviews(reviews_dir, <取り込み元の名前>, [番号, ...])` で書く
（要確認リストのファイルが無い取り込み元の取り込みは停止する．要確認が無いなら `[]` で書く）．
"""
from tiasdata.importers import reviews
directory = tmp_path / "reviews"
directory.mkdir()
monkeypatch.setattr(reviews, "default_reviews_dir", lambda: directory)
return directory
@dataclass(frozen=True)
class ImportEnv:
"""取り込みを試すための一時の設定ファイル・取り込み元・対応表一式．"""
config_path: Path
source_name: str
source: DummyTongueColorSource
patient_map: DummyPatientMap
reviews_dir: Path
catalog_root: Path
identity_path: Path
@pytest.fixture
def import_env(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
tongue_color_source: DummyTongueColorSource,
tongue_color_patient_map: DummyPatientMap,
reviews_dir: Path,
) -> ImportEnv:
"""ダミーの取り込み元 `dummy_tongue` を設定した設定ファイル（`TIASDATA_CONFIG` で指す）．
`tiasdata init` はまだ実行していない．要確認リストは空（`[[review]]` が 0 件のファイル．
要確認リストのファイルが無いと取り込みは停止する）．
"""
source_name = "dummy_tongue"
write_reviews(reviews_dir, source_name, [])
catalog_root = tmp_path / "data" / "tiasdata"
identity_path = tmp_path / "data" / "identity" / "identity.sqlite"
config_path = tmp_path / "conf" / "config.toml"
config_path.parent.mkdir(parents=True, exist_ok=True)
config_path.write_text(
"[catalog]\n"
f"root = '{catalog_root.as_posix()}'\n"
f"identity = '{identity_path.as_posix()}'\n\n"
f"[sources.{source_name}]\n"
f"path = '{tongue_color_source.root.as_posix()}'\n"
'importer = "tongue_color_folders"\n'
f"patient_map = '{tongue_color_patient_map.path.as_posix()}'\n",
encoding="utf-8",
)
monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
return ImportEnv(
config_path=config_path,
source_name=source_name,
source=tongue_color_source,
patient_map=tongue_color_patient_map,
reviews_dir=reviews_dir,
catalog_root=catalog_root,
identity_path=identity_path,
)
@pytest.fixture(autouse=True)
def _isolate_config_env(monkeypatch: pytest.MonkeyPatch) -> None:
# 開発者の環境変数が混ざらないようにする（テストごとに必要なら設定する）
monkeypatch.delenv("TIASDATA_CONFIG", raising=False)
@pytest.fixture
def catalog_conn(tmp_path: Path) -> Iterator[sqlite3.Connection]:
"""マイグレーション適用済みの目録（tmp_path に作る）．"""
conn = catalog.connect(tmp_path / "catalog.sqlite", create=True)
catalog.apply_migrations(conn, "catalog")
try:
yield conn
finally:
conn.close()
@pytest.fixture
def identity_conn(tmp_path: Path) -> Iterator[sqlite3.Connection]:
"""マイグレーション適用済みの対応表（tmp_path に作る）．"""
conn = catalog.connect(tmp_path / "identity.sqlite", create=True)
catalog.apply_migrations(conn, "identity")
try:
yield conn
finally:
conn.close()
@dataclass
class DatasetsEnv:
"""データセットの定義のフォルダと，差し替えた本リポジトリの git の状態．
`repo_state` を書き換えると，以後の `tiasdata dataset freeze` が読む状態が変わる．
"""
datasets_dir: Path
repo_state: object
@pytest.fixture
def datasets_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DatasetsEnv:
"""一時フォルダの `datasets/`（リポジトリの `datasets/tongue_color.toml` を複製）と，
未コミットの変更が無い git の状態を差し替えた環境．
`tiasdata.datasets.default_datasets_dir`・`tiasdata.repo.read_repo_state` を差し替えるので，
同じプロセスで呼ぶ `cli.main` に効く（subprocess で起動した CLI には効かない）．
"""
from tiasdata import datasets, repo
directory = tmp_path / "datasets"
directory.mkdir()
source = Path(__file__).resolve().parents[1] / "datasets" / "tongue_color.toml"
(directory / "tongue_color.toml").write_bytes(source.read_bytes())
env = DatasetsEnv(
datasets_dir=directory,
repo_state=repo.RepoState(commit="0123456789abcdef0123456789abcdef01234567", uncommitted=0),
)
monkeypatch.setattr(datasets, "default_datasets_dir", lambda: directory)
monkeypatch.setattr(repo, "read_repo_state", lambda root=None: env.repo_state)
return env
