"""回収ホワイトリストの判定のテスト（SPEC_01「回収ホワイトリスト」・SPEC_03「pull」）.

最重要: 名前が一致するファイルだけが回収対象になり，分割結果・個票の予測・画像・manifest 等は
回収されないこと．testing ルール必須対象 4．
"""

import io
import tarfile

import pytest

from tias_judge_model.remote.whitelist import (
    check_pull_members,
    is_pull_allowed,
    parse_find_listing,
    plan_pull,
    validate_run_id,
)

RUN_ID = "20260927-120000_newton_abc1234"

ROOT_FILES = ["config.yaml", "history.csv", "metrics.json", "confusion_matrix.csv", "log.txt"]

# 列挙外（回収してはならない）ファイル．患者 ID・画像名を含みうる名前も混ぜる
NOT_ALLOWED = [
    "split.json",
    "predictions.csv",
    "per_image_predictions.csv",
    "manifest.csv",
    "confusion_matrix.png",  # figures/ 以外にある *.png
    "P0001_tongue.png",
    "P0001_tongue.jpg",
    "images/pale_1/P0001.png",
    "misclassified/P0001.png",
    "data/manifest.csv",
    "config.yml",
    "config.yaml.bak",
    "metrics.json.tmp",
    "sub/config.yaml",
    "figures/config.yaml",
    "figures/cm.jpg",
    "figures/sub/cm.png",
    "checkpoints/best.pth",
    "checkpoints/sub/best.pt",
    "best.pt",
]


class TestIsPullAllowed:
    @pytest.mark.parametrize("name", ROOT_FILES)
    def test_root_files_are_allowed(self, name: str) -> None:
        assert is_pull_allowed(name, with_checkpoints=False)

    @pytest.mark.parametrize("name", ["figures/confusion_matrix.png", "figures/learning_curve.png"])
    def test_figures_png_is_allowed(self, name: str) -> None:
        assert is_pull_allowed(name, with_checkpoints=False)

    @pytest.mark.parametrize("name", NOT_ALLOWED)
    def test_names_outside_whitelist_are_not_allowed(self, name: str) -> None:
        assert not is_pull_allowed(name, with_checkpoints=False)
        # --with-checkpoints は checkpoints/*.pt 以外の判定を変えない
        assert not is_pull_allowed(name, with_checkpoints=True)

    def test_checkpoints_only_with_flag(self) -> None:
        assert not is_pull_allowed("checkpoints/best.pt", with_checkpoints=False)
        assert is_pull_allowed("checkpoints/best.pt", with_checkpoints=True)

    @pytest.mark.parametrize(
        "name",
        [
            "../config.yaml",
            "figures/../config.yaml",
            "/config.yaml",
            "C:config.yaml",
            "C:/config.yaml",
            "figures\\cm.png",
            "..\\config.yaml",
        ],
    )
    def test_unsafe_paths_are_not_allowed(self, name: str) -> None:
        assert not is_pull_allowed(name, with_checkpoints=True)


class TestValidateRunId:
    def test_accepts_normal_run_id(self) -> None:
        assert validate_run_id(RUN_ID) == RUN_ID

    @pytest.mark.parametrize(
        "run_id", ["a/b", "a\\b", "..", "../x", "x/..", "a..b", "/abs", "runs/x", ""]
    )
    def test_rejects_run_id_with_separator_or_dotdot(self, run_id: str) -> None:
        with pytest.raises(ValueError):
            validate_run_id(run_id)


class TestPlanPull:
    def _listing(self, names: list[str], kind: str = "f") -> list[tuple[str, str]]:
        return [(kind, f"{RUN_ID}/{name}") for name in names]

    def test_selects_only_whitelisted_files(self) -> None:
        allowed = [*ROOT_FILES, "figures/cm.png"]
        listing = self._listing(allowed + NOT_ALLOWED + ["checkpoints/best.pt"])
        plan = plan_pull(listing, RUN_ID, with_checkpoints=False)
        assert sorted(plan.files) == sorted(f"{RUN_ID}/{name}" for name in allowed)
        assert plan.skipped == len(NOT_ALLOWED) + 1

    def test_with_checkpoints_includes_pt(self) -> None:
        listing = self._listing(["config.yaml", "checkpoints/best.pt", "split.json"])
        plan = plan_pull(listing, RUN_ID, with_checkpoints=True)
        assert sorted(plan.files) == [f"{RUN_ID}/checkpoints/best.pt", f"{RUN_ID}/config.yaml"]
        assert plan.skipped == 1

    def test_links_are_not_selected_even_with_whitelisted_name(self) -> None:
        listing = self._listing(["config.yaml"], kind="l")
        plan = plan_pull(listing, RUN_ID, with_checkpoints=False)
        assert plan.files == ()
        assert plan.skipped == 1

    def test_files_of_other_run_are_not_selected(self) -> None:
        plan = plan_pull([("f", "other_run/config.yaml")], RUN_ID, with_checkpoints=False)
        assert plan.files == ()

    def test_parse_find_listing_round_trip(self) -> None:
        data = f"f {RUN_ID}/config.yaml\0f {RUN_ID}/figures/a b.png\0l {RUN_ID}/x\0".encode()
        assert parse_find_listing(data) == [
            ("f", f"{RUN_ID}/config.yaml"),
            ("f", f"{RUN_ID}/figures/a b.png"),
            ("l", f"{RUN_ID}/x"),
        ]


def _regular(name: str) -> tarfile.TarInfo:
    info = tarfile.TarInfo(name)
    info.size = 0
    return info


def _link(name: str, target: str, kind: bytes) -> tarfile.TarInfo:
    info = tarfile.TarInfo(name)
    info.type = kind
    info.linkname = target
    return info


class TestCheckPullMembers:
    def test_accepts_only_whitelisted_regular_files(self) -> None:
        members = [_regular(f"{RUN_ID}/{name}") for name in [*ROOT_FILES, "figures/cm.png"]]
        assert check_pull_members(members, RUN_ID, with_checkpoints=False) == len(members)

    @pytest.mark.parametrize(
        "bad",
        [
            *(_regular(f"{RUN_ID}/{name}") for name in NOT_ALLOWED),
            _regular(f"{RUN_ID}/checkpoints/best.pt"),
            _regular("other_run/config.yaml"),
            _regular("config.yaml"),
            _regular(f"{RUN_ID}/../config.yaml"),
            _regular(f"/{RUN_ID}/config.yaml"),
            _regular(f"C:/{RUN_ID}/config.yaml"),
            _regular(f"{RUN_ID}\\config.yaml"),
            _link(f"{RUN_ID}/config.yaml", "/etc/passwd", tarfile.SYMTYPE),
            _link(f"{RUN_ID}/metrics.json", f"{RUN_ID}/config.yaml", tarfile.LNKTYPE),
        ],
    )
    def test_rejects_whole_archive_if_any_member_is_bad(self, bad: tarfile.TarInfo) -> None:
        members = [_regular(f"{RUN_ID}/config.yaml"), bad, _regular(f"{RUN_ID}/log.txt")]
        with pytest.raises(ValueError):
            check_pull_members(members, RUN_ID, with_checkpoints=False)

    def test_error_message_does_not_contain_member_names(self) -> None:
        secret = "P0001_yamada_tongue"
        members = [
            _regular(f"{RUN_ID}/{secret}.png"),
            _link(f"{RUN_ID}/{secret}_link", "/etc/passwd", tarfile.SYMTYPE),
            _regular(f"../{secret}.csv"),
        ]
        with pytest.raises(ValueError) as excinfo:
            check_pull_members(members, RUN_ID, with_checkpoints=False)
        assert secret not in str(excinfo.value)

    def test_checkpoints_accepted_only_with_flag(self) -> None:
        members = [_regular(f"{RUN_ID}/checkpoints/best.pt")]
        assert check_pull_members(members, RUN_ID, with_checkpoints=True) == 1
        with pytest.raises(ValueError):
            check_pull_members(members, RUN_ID, with_checkpoints=False)

    def test_members_from_real_tar_file(self) -> None:
        buffer = io.BytesIO()
        with tarfile.open(fileobj=buffer, mode="w") as tar:
            tar.addfile(_regular(f"{RUN_ID}/config.yaml"), io.BytesIO(b""))
            tar.addfile(_link(f"{RUN_ID}/log.txt", "/etc/shadow", tarfile.SYMTYPE))
        buffer.seek(0)
        with tarfile.open(fileobj=buffer, mode="r") as tar, pytest.raises(ValueError):
            check_pull_members(tar.getmembers(), RUN_ID, with_checkpoints=False)