Newer
Older
TIASDataHub / tests / test_config.py
"""設定ファイルの読み込み・検証(SPEC_01「設定ファイル」,testing.md 必須対象 7)."""

from pathlib import Path

import pytest
from pydantic import ValidationError

from tiasdata.config import default_config_path, load_config


def _write(path: Path, text: str) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")
    return path


def _minimal(tmp_path: Path, extra: str = "") -> str:
    root = (tmp_path / "data" / "tiasdata").as_posix()
    identity = (tmp_path / "data" / "identity" / "identity.sqlite").as_posix()
    return f'[catalog]\nroot = "{root}"\nidentity = "{identity}"\n{extra}'


FULL_EXTRA = """
[sources.tiasjudgemodel]
path = "src_a"
importer = "tongue_color_folders"
patient_map = "map.xlsx"

[servers.newton]
host = "user@example.invalid"
key = "keys/id_ed25519"
root = "~/data/tiasdata"
"""


class TestLoadValid:
    def test_loads_minimal_config(self, tmp_path: Path) -> None:
        config = load_config(_write(tmp_path / "config.toml", _minimal(tmp_path)))
        assert config.catalog.root == tmp_path / "data" / "tiasdata"
        assert config.catalog.identity == tmp_path / "data" / "identity" / "identity.sqlite"
        assert config.sources == {}
        assert config.servers == {}

    def test_loads_sources_and_servers_by_name(self, tmp_path: Path) -> None:
        config = load_config(_write(tmp_path / "config.toml", _minimal(tmp_path, FULL_EXTRA)))
        assert set(config.sources) == {"tiasjudgemodel"}
        assert config.sources["tiasjudgemodel"].importer == "tongue_color_folders"
        assert set(config.servers) == {"newton"}
        assert config.servers["newton"].host == "user@example.invalid"

    def test_patient_map_is_optional(self, tmp_path: Path) -> None:
        extra = '[sources.a]\npath = "x"\nimporter = "tongue_color_folders"\n'
        config = load_config(_write(tmp_path / "config.toml", _minimal(tmp_path, extra)))
        assert config.sources["a"].patient_map is None

    def test_catalog_layout_under_root(self, tmp_path: Path) -> None:
        catalog = load_config(_write(tmp_path / "config.toml", _minimal(tmp_path))).catalog
        assert catalog.catalog_path == catalog.root / "catalog.sqlite"
        assert catalog.store_dir == catalog.root / "store"
        assert catalog.bundles_dir == catalog.root / "bundles"
        assert catalog.backups_dir == catalog.root / "backups"


class TestPathResolution:
    def test_relative_paths_are_relative_to_config_folder(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        conf_dir = tmp_path / "conf"
        text = '[catalog]\nroot = "data/tiasdata"\nidentity = "identity/identity.sqlite"\n'
        config_path = _write(conf_dir / "config.toml", text + FULL_EXTRA)
        # 作業フォルダに依存しないこと
        elsewhere = tmp_path / "elsewhere"
        elsewhere.mkdir()
        monkeypatch.chdir(elsewhere)

        config = load_config(config_path)

        base = conf_dir.resolve()
        assert config.catalog.root == base / "data" / "tiasdata"
        assert config.catalog.identity == base / "identity" / "identity.sqlite"
        assert config.sources["tiasjudgemodel"].path == base / "src_a"
        assert config.sources["tiasjudgemodel"].patient_map == base / "map.xlsx"
        assert config.servers["newton"].key == base / "keys" / "id_ed25519"

    def test_tilde_is_expanded(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        home = tmp_path / "home"
        home.mkdir()
        monkeypatch.setenv("HOME", str(home))
        monkeypatch.setenv("USERPROFILE", str(home))
        text = (
            '[catalog]\nroot = "~/tiasdata"\nidentity = "~/identity/identity.sqlite"\n'
            '[sources.a]\npath = "~/src"\nimporter = "tongue_color_folders"\n'
            'patient_map = "~/map.xlsx"\n'
            '[servers.s]\nhost = "u@h"\nkey = "~/.ssh/key"\nroot = "~/data/tiasdata"\n'
        )
        config = load_config(_write(tmp_path / "conf" / "config.toml", text))
        assert config.catalog.root == home / "tiasdata"
        assert config.catalog.identity == home / "identity" / "identity.sqlite"
        assert config.sources["a"].path == home / "src"
        assert config.sources["a"].patient_map == home / "map.xlsx"
        assert config.servers["s"].key == home / ".ssh" / "key"

    def test_server_root_is_not_expanded(self, tmp_path: Path) -> None:
        # サーバ上のパスはサーバ側で解釈するため,手元で展開・解決しない
        config = load_config(_write(tmp_path / "config.toml", _minimal(tmp_path, FULL_EXTRA)))
        assert config.servers["newton"].root == "~/data/tiasdata"

    def test_resolved_paths_are_absolute(self, tmp_path: Path) -> None:
        text = '[catalog]\nroot = "r"\nidentity = "i/identity.sqlite"\n'
        config = load_config(_write(tmp_path / "config.toml", text))
        assert config.catalog.root.is_absolute()
        assert config.catalog.identity.is_absolute()


class TestValidationErrors:
    @pytest.mark.parametrize(
        "extra",
        [
            "unknown_top = 1\n",
            "[unknown_section]\nx = 1\n",
            '[sources.a]\npath = "x"\nimporter = "i"\ntypo = "y"\n',
            '[servers.s]\nhost = "h"\nkey = "k"\nroot = "r"\nport = 22\n',
        ],
        ids=["top_level", "section", "source", "server"],
    )
    def test_unknown_key_is_rejected(self, tmp_path: Path, extra: str) -> None:
        # 未知のキーはトップレベルに置くと [catalog] の中に入ってしまうので,
        # トップレベル用は [catalog] より前に書く
        if extra.startswith("unknown_top"):
            text = extra + _minimal(tmp_path)
        else:
            text = _minimal(tmp_path, extra)
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", text))

    def test_unknown_key_in_catalog_is_rejected(self, tmp_path: Path) -> None:
        text = _minimal(tmp_path) + 'stroe = "typo"\n'
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", text))

    def test_missing_catalog_section_is_rejected(self, tmp_path: Path) -> None:
        text = '[sources.a]\npath = "x"\nimporter = "i"\n'
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", text))

    @pytest.mark.parametrize("missing", ["root", "identity"])
    def test_missing_catalog_key_is_rejected(self, tmp_path: Path, missing: str) -> None:
        values = {"root": '"r"', "identity": '"i/identity.sqlite"'}
        lines = [f"{k} = {v}" for k, v in values.items() if k != missing]
        text = "[catalog]\n" + "\n".join(lines) + "\n"
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", text))

    @pytest.mark.parametrize("missing", ["path", "importer"])
    def test_missing_source_key_is_rejected(self, tmp_path: Path, missing: str) -> None:
        values = {"path": '"x"', "importer": '"tongue_color_folders"'}
        lines = [f"{k} = {v}" for k, v in values.items() if k != missing]
        extra = "[sources.a]\n" + "\n".join(lines) + "\n"
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", _minimal(tmp_path, extra)))

    @pytest.mark.parametrize("missing", ["host", "key", "root"])
    def test_missing_server_key_is_rejected(self, tmp_path: Path, missing: str) -> None:
        values = {"host": '"u@h"', "key": '"k"', "root": '"~/data"'}
        lines = [f"{k} = {v}" for k, v in values.items() if k != missing]
        extra = "[servers.s]\n" + "\n".join(lines) + "\n"
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", _minimal(tmp_path, extra)))

    @pytest.mark.parametrize(
        "identity",
        ["data/tiasdata/identity.sqlite", "data/tiasdata/sub/identity.sqlite"],
        ids=["directly_in_root", "under_root"],
    )
    def test_identity_inside_catalog_root_is_rejected(self, tmp_path: Path, identity: str) -> None:
        # 対応表は目録とは別のフォルダに置く(SPEC_01「構成要素」)
        text = f'[catalog]\nroot = "data/tiasdata"\nidentity = "{identity}"\n'
        with pytest.raises(ValidationError):
            load_config(_write(tmp_path / "config.toml", text))


class TestFileErrors:
    def test_missing_file_raises_value_error(self, tmp_path: Path) -> None:
        with pytest.raises(ValueError, match="設定ファイル"):
            load_config(tmp_path / "nope.toml")

    def test_missing_file_is_not_validation_error(self, tmp_path: Path) -> None:
        with pytest.raises(ValueError) as exc:
            load_config(tmp_path / "nope.toml")
        assert not isinstance(exc.value, ValidationError)

    def test_broken_toml_raises_value_error(self, tmp_path: Path) -> None:
        path = _write(tmp_path / "config.toml", "[catalog\nroot = \n")
        with pytest.raises(ValueError) as exc:
            load_config(path)
        assert not isinstance(exc.value, ValidationError)

    def test_non_utf8_file_raises_value_error(self, tmp_path: Path) -> None:
        path = tmp_path / "config.toml"
        path.write_bytes('[catalog]\nroot = "日本語"\n'.encode("cp932"))
        with pytest.raises(ValueError):
            load_config(path)


class TestLocation:
    def test_env_var_takes_precedence(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        config_path = _write(tmp_path / "custom" / "my.toml", _minimal(tmp_path))
        monkeypatch.setenv("TIASDATA_CONFIG", str(config_path))
        assert default_config_path() == config_path
        assert load_config().catalog.root == tmp_path / "data" / "tiasdata"

    def test_default_location_without_env_var(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        home = tmp_path / "home"
        monkeypatch.setenv("HOME", str(home))
        monkeypatch.setenv("USERPROFILE", str(home))
        assert default_config_path() == home / ".config" / "tiasdata" / "config.toml"

    def test_load_without_path_uses_default_location(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        home = tmp_path / "home"
        monkeypatch.setenv("HOME", str(home))
        monkeypatch.setenv("USERPROFILE", str(home))
        _write(home / ".config" / "tiasdata" / "config.toml", _minimal(tmp_path))
        assert load_config().catalog.root == tmp_path / "data" / "tiasdata"

    def test_env_var_with_tilde_is_expanded(
        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
    ) -> None:
        home = tmp_path / "home"
        monkeypatch.setenv("HOME", str(home))
        monkeypatch.setenv("USERPROFILE", str(home))
        monkeypatch.setenv("TIASDATA_CONFIG", "~/conf/config.toml")
        assert default_config_path() == home / "conf" / "config.toml"