"""`.claude/hooks/` のフック起動まわりのテスト（Issue #35: Python フックの起動ラッパー）.

実行（リポジトリルートで）:
    python -B -m unittest discover -s .claude/hooks/tests -v

外部依存なし（標準ライブラリの unittest のみ）．bash が必要（Windows では Git for Windows の bash）．
notify.py は NTFY_TOPIC 未設定，またはローカルの一時 HTTP サーバー宛てでのみ実行し，ntfy.sh には送信しない．
"""

from __future__ import annotations

import ast
import contextlib
import http.server
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import unittest
from unittest import mock

HOOKS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
REPO_ROOT = os.path.dirname(os.path.dirname(HOOKS_DIR))
WRAPPER = os.path.join(HOOKS_DIR, "run_python.sh")
SETTINGS = os.path.join(REPO_ROOT, ".claude", "settings.json")
RESTRICT = os.path.join(HOOKS_DIR, "restrict_repo_access.py")
NOTIFY = os.path.join(HOOKS_DIR, "notify.py")
CHECK_SYNC = os.path.join(HOOKS_DIR, "check_sync.sh")

TIMEOUT = 60


def find_bash() -> str:
    """テストに使う bash を返す（Windows では WSL の System32\\bash.exe を避ける）."""
    env_bash = os.environ.get("HOOK_TEST_BASH")
    if env_bash:
        return env_bash
    found = shutil.which("bash")
    if os.name == "nt":
        if found is None or "system32" in found.lower():
            for candidate in (
                r"C:\Program Files\Git\usr\bin\bash.exe",
                r"C:\Program Files\Git\bin\bash.exe",
            ):
                if os.path.exists(candidate):
                    return candidate
    if found is None:
        raise unittest.SkipTest("bash が見つからない")
    return found


BASH = find_bash()
REAL_PYTHON = sys.executable.replace("\\", "/")


def bash_bin_dirs() -> list[str]:
    """git・timeout・dirname 等の基本コマンドがあるディレクトリ（Python は含まない）."""
    dirs = [os.path.dirname(BASH)]
    for cmd in ("git", "timeout", "dirname"):
        found = shutil.which(cmd)
        if found and os.path.dirname(found) not in dirs:
            dirs.append(os.path.dirname(found))
    if os.name != "nt":
        for d in ("/usr/bin", "/bin"):
            if d not in dirs:
                dirs.append(d)
    return dirs


class FakeBin:
    """PATH の先頭に置く偽コマンド用の一時ディレクトリ."""

    def __init__(self) -> None:
        self.dir = tempfile.mkdtemp(prefix="fakebin_")

    def cleanup(self) -> None:
        shutil.rmtree(self.dir, ignore_errors=True)

    def _write(self, name: str, body: str) -> None:
        path = os.path.join(self.dir, name)
        with open(path, "w", newline="\n") as f:
            f.write("#!/bin/sh\n" + body)
        os.chmod(path, 0o755)

    def dummy(self, name: str) -> None:
        """存在するが起動できない（常に exit 1）ダミー（Microsoft Store エイリアス相当）."""
        self._write(name, 'echo "DUMMY_CALLED:' + name + '" >&2\nexit 1\n')

    def marker(self, name: str) -> None:
        """`-c ''` は成功し，本実行時は自分の名前と引数を出力する偽 Python."""
        self._write(
            name,
            'if [ "$1" = "-c" ]; then exit 0; fi\n'
            'echo "SELECTED:' + name + '"\nexit 0\n',
        )

    def real(self, name: str) -> None:
        """テスト実行中の本物の Python に委譲する."""
        self._write(name, 'exec "' + REAL_PYTHON + '" "$@"\n')


def make_env(path_dirs: list[str], extra: dict | None = None, drop: tuple = ()) -> dict:
    env = dict(os.environ)
    for key in drop:
        env.pop(key, None)
    env["PATH"] = os.pathsep.join(path_dirs)
    if extra:
        env.update(extra)
    return env


def run_wrapper(args: list[str], env: dict, stdin: bytes = b"", cwd: str | None = None):
    return subprocess.run(
        [BASH, WRAPPER] + args,
        input=stdin,
        capture_output=True,
        env=env,
        cwd=cwd or REPO_ROOT,
        timeout=TIMEOUT,
    )


class TempDirMixin:
    def setUp(self) -> None:  # noqa: D401
        self.fake = FakeBin()
        self.work = tempfile.mkdtemp(prefix="hooktest_")

    def tearDown(self) -> None:
        self.fake.cleanup()
        shutil.rmtree(self.work, ignore_errors=True)

    def write_script(self, name: str, source: str) -> str:
        path = os.path.join(self.work, name)
        with open(path, "w", encoding="utf-8", newline="\n") as f:
            f.write(source)
        return path


# ---------------------------------------------------------------------------
# run_python.sh: Python コマンドの選択順
# ---------------------------------------------------------------------------
class TestWrapperSelection(TempDirMixin, unittest.TestCase):
    def selected(self, *wrapper_args: str) -> subprocess.CompletedProcess:
        return run_wrapper(list(wrapper_args) or ["script.py"], make_env([self.fake.dir]))

    def assertSelected(self, result, name: str) -> None:
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.decode().strip(), "SELECTED:" + name)

    def test_prefers_python3_when_all_available(self):
        for n in ("python3", "python", "py"):
            self.fake.marker(n)
        self.assertSelected(self.selected(), "python3")

    def test_falls_back_to_python_when_no_python3(self):
        self.fake.marker("python")
        self.fake.marker("py")
        self.assertSelected(self.selected(), "python")

    def test_falls_back_to_py_when_only_py(self):
        self.fake.marker("py")
        self.assertSelected(self.selected(), "py")

    def test_skips_dummy_python3(self):
        self.fake.dummy("python3")
        self.fake.marker("python")
        self.fake.marker("py")
        self.assertSelected(self.selected(), "python")

    def test_skips_dummy_python3_and_python(self):
        self.fake.dummy("python3")
        self.fake.dummy("python")
        self.fake.marker("py")
        self.assertSelected(self.selected(), "py")

    def test_fail_closed_flag_does_not_change_selection(self):
        self.fake.dummy("python3")
        self.fake.marker("python")
        self.assertSelected(self.selected("--fail-closed", "script.py"), "python")

    def test_dummy_is_not_used_for_actual_run(self):
        """ダミーは起動確認で弾かれ，本実行には使われない."""
        self.fake.dummy("python3")
        self.fake.marker("python")
        result = self.selected()
        self.assertSelected(result, "python")
        # ダミーが呼ばれるのは起動確認（-c ''）のみで，その stderr は捨てられている
        self.assertNotIn(b"DUMMY_CALLED", result.stderr)


# ---------------------------------------------------------------------------
# run_python.sh: Python 未検出時
# ---------------------------------------------------------------------------
class TestWrapperNotFound(TempDirMixin, unittest.TestCase):
    def test_no_python_exit_1_without_flag(self):
        result = run_wrapper(["script.py"], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 1)
        self.assertTrue(result.stderr.strip(), "stderr に理由が出力されること")

    def test_no_python_exit_2_with_fail_closed(self):
        result = run_wrapper(["--fail-closed", "script.py"], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 2)
        self.assertTrue(result.stderr.strip(), "stderr に理由が出力されること")

    def test_only_dummies_exit_1_without_flag(self):
        for n in ("python3", "python", "py"):
            self.fake.dummy(n)
        result = run_wrapper(["script.py"], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 1)
        self.assertTrue(result.stderr.strip())

    def test_only_dummies_exit_2_with_fail_closed(self):
        for n in ("python3", "python", "py"):
            self.fake.dummy(n)
        result = run_wrapper(["--fail-closed", "script.py"], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 2)
        self.assertTrue(result.stderr.strip())

    def test_not_found_with_no_script_args(self):
        """引数なしでも異常終了（set -u 的なクラッシュ）せず規定の終了コードを返す."""
        result = run_wrapper([], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 1)
        result = run_wrapper(["--fail-closed"], make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 2)


# ---------------------------------------------------------------------------
# run_python.sh: 終了コード・stdin・引数の伝搬（本物の Python に委譲）
# ---------------------------------------------------------------------------
class TestWrapperPassthrough(TempDirMixin, unittest.TestCase):
    def setUp(self) -> None:
        super().setUp()
        self.fake.dummy("python3")  # Windows の Store エイリアス相当
        self.fake.real("python")
        self.env = make_env([self.fake.dir])

    def test_exit_codes_propagate(self):
        script = self.write_script("exit.py", "import sys\nsys.exit(int(sys.argv[1]))\n")
        for code in (0, 1, 2, 3, 42):
            for flag in ([], ["--fail-closed"]):
                with self.subTest(code=code, flag=flag):
                    result = run_wrapper(flag + [script, str(code)], self.env)
                    self.assertEqual(result.returncode, code, result.stderr)

    def test_args_passed_verbatim(self):
        script = self.write_script(
            "argv.py", "import json, sys\nprint(json.dumps(sys.argv[1:]))\n"
        )
        args = ["a b", "", "--fail-closed", "x=y", "日本語", "--flag"]
        for flag in ([], ["--fail-closed"]):
            with self.subTest(flag=flag):
                result = run_wrapper(
                    flag + [script] + args,
                    dict(self.env, PYTHONIOENCODING="utf-8"),
                )
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(json.loads(result.stdout.decode("utf-8")), args)

    def test_fail_closed_flag_not_passed_to_script(self):
        script = self.write_script(
            "argv.py", "import json, sys\nprint(json.dumps(sys.argv[1:]))\n"
        )
        result = run_wrapper(["--fail-closed", script, "one"], self.env)
        self.assertEqual(json.loads(result.stdout.decode()), ["one"])

    def test_stdin_passed_through(self):
        script = self.write_script(
            "stdin.py",
            "import sys\nsys.stdout.buffer.write(sys.stdin.buffer.read())\n",
        )
        payload = json.dumps({"tool_name": "Read", "x": "日本語"}, ensure_ascii=False).encode("utf-8")
        payload = payload * 200  # ある程度の大きさでも欠けないこと
        for flag in ([], ["--fail-closed"]):
            with self.subTest(flag=flag):
                result = run_wrapper(flag + [script], self.env, stdin=payload)
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(result.stdout, payload)

    def test_stdin_not_consumed_by_probe(self):
        """起動確認（-c ''）が stdin を読み捨てないこと（ダミー・本物ともに）."""
        self.fake.real("python3")  # python3 も本物にして probe → 本実行の経路を通す
        script = self.write_script(
            "stdin.py", "import sys\nsys.stdout.write(sys.stdin.read())\n"
        )
        result = run_wrapper([script], self.env, stdin=b"hello-stdin")
        self.assertEqual(result.stdout, b"hello-stdin")

    def test_stdin_script_dash(self):
        """check_sync.sh と同じ `-` 形式（ヒアドキュメントのスクリプト）が動く."""
        result = run_wrapper(
            ["-", "arg1"],
            self.env,
            stdin=b"import sys\nprint('ok:' + sys.argv[1])\n",
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout.decode().strip(), "ok:arg1")


# ---------------------------------------------------------------------------
# settings.json の配線
# ---------------------------------------------------------------------------
class TestSettingsJson(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        with open(SETTINGS, encoding="utf-8") as f:
            cls.raw = f.read()
        cls.settings = json.loads(cls.raw)
        cls.commands = []
        for event, groups in cls.settings.get("hooks", {}).items():
            for group in groups:
                for hook in group.get("hooks", []):
                    if hook.get("type") == "command":
                        cls.commands.append((event, group.get("matcher"), hook))

    def test_restrict_repo_access_uses_wrapper_with_fail_closed(self):
        cmds = [h["command"] for _, _, h in self.commands if "restrict_repo_access.py" in h["command"]]
        self.assertEqual(
            cmds,
            ["bash .claude/hooks/run_python.sh --fail-closed .claude/hooks/restrict_repo_access.py"],
        )

    def test_restrict_repo_access_is_pretooluse_for_file_tools(self):
        entries = [(e, m) for e, m, h in self.commands if "restrict_repo_access.py" in h["command"]]
        self.assertEqual(len(entries), 1)
        event, matcher = entries[0]
        self.assertEqual(event, "PreToolUse")
        for tool in ("Read", "Write", "Edit", "Glob", "Grep", "Bash"):
            self.assertIn(tool, matcher.split("|"))

    def test_notify_three_hooks_use_wrapper_without_fail_closed(self):
        cmds = [h["command"] for _, _, h in self.commands if "notify.py" in h["command"]]
        self.assertEqual(len(cmds), 3)
        for cmd in cmds:
            self.assertEqual(cmd, "bash .claude/hooks/run_python.sh .claude/hooks/notify.py")
            self.assertNotIn("--fail-closed", cmd)

    def test_notify_events(self):
        events = sorted(e for e, _, h in self.commands if "notify.py" in h["command"])
        self.assertEqual(events, ["Notification", "PreToolUse", "Stop"])

    def test_no_direct_python_invocation_left(self):
        self.assertIsNone(
            re.search(r"\b(python3?|py)\s+\.claude/hooks/", self.raw),
            "python .claude/hooks/ の直接呼び出しが残っている",
        )
        for _, _, hook in self.commands:
            if ".py" in hook["command"]:
                self.assertTrue(
                    hook["command"].startswith("bash .claude/hooks/run_python.sh "),
                    hook["command"],
                )

    def test_referenced_scripts_exist(self):
        for _, _, hook in self.commands:
            for token in hook["command"].split():
                if token.startswith(".claude/hooks/"):
                    self.assertTrue(os.path.isfile(os.path.join(REPO_ROOT, token)), token)


# ---------------------------------------------------------------------------
# restrict_repo_access.py をラッパー経由で（settings.json と同じコマンドで）起動
# ---------------------------------------------------------------------------
class TestRestrictViaWrapper(TempDirMixin, unittest.TestCase):
    OUTSIDE = os.path.join(os.path.dirname(REPO_ROOT), "outside_of_repo_dummy.txt")
    INSIDE = os.path.join(REPO_ROOT, "CLAUDE.md")

    def run_hook(self, data: dict, env: dict) -> subprocess.CompletedProcess:
        # settings.json の command と同じ相対パス・cwd で起動する
        return subprocess.run(
            [BASH, ".claude/hooks/run_python.sh", "--fail-closed", ".claude/hooks/restrict_repo_access.py"],
            input=json.dumps(data).encode("utf-8"),
            capture_output=True,
            env=env,
            cwd=REPO_ROOT,
            timeout=TIMEOUT,
        )

    @staticmethod
    def is_blocked(result) -> bool:
        if result.returncode == 2:
            return True
        if result.returncode != 0:
            return False
        out = result.stdout.decode("utf-8").strip()
        if not out:
            return False
        decision = json.loads(out).get("hookSpecificOutput", {}).get("permissionDecision")
        return decision == "deny"

    def envs(self):
        # 1) 実環境の PATH（このマシンでは python3 が Store エイリアスのダミー）
        yield "real PATH", dict(os.environ)
        # 2) ダミー python3 + 本物 python のみ
        self.fake.dummy("python3")
        self.fake.real("python")
        yield "dummy python3 + python", make_env([self.fake.dir])

    def test_blocks_edit_outside_repo(self):
        # Issue #49: 読み取りと新規作成はリポジトリ外も許可し，既存ファイルの書き換え（Edit）はブロックする
        data = {"tool_name": "Edit", "tool_input": {"file_path": self.OUTSIDE, "old_string": "a", "new_string": "b"}, "cwd": REPO_ROOT}
        for label, env in self.envs():
            with self.subTest(env=label):
                result = self.run_hook(data, env)
                self.assertTrue(self.is_blocked(result), (result.returncode, result.stdout, result.stderr))

    def test_blocks_destructive_bash_outside_repo(self):
        outside_dir = os.path.dirname(REPO_ROOT).replace("\\", "/")
        data = {"tool_name": "Bash", "tool_input": {"command": f'rm -rf "{outside_dir}/foo"'}, "cwd": REPO_ROOT}
        for label, env in self.envs():
            with self.subTest(env=label):
                result = self.run_hook(data, env)
                self.assertTrue(self.is_blocked(result), (result.returncode, result.stdout, result.stderr))

    def test_allows_inside_repo(self):
        cases = [
            {"tool_name": "Read", "tool_input": {"file_path": self.INSIDE}, "cwd": REPO_ROOT},
            {"tool_name": "Edit", "tool_input": {"file_path": os.path.join(REPO_ROOT, "docs", "x.md")}, "cwd": REPO_ROOT},
            {"tool_name": "Grep", "tool_input": {"pattern": "x"}, "cwd": REPO_ROOT},
            {"tool_name": "Bash", "tool_input": {"command": "rm -rf ./build"}, "cwd": REPO_ROOT},
        ]
        for label, env in self.envs():
            for data in cases:
                with self.subTest(env=label, tool=data["tool_name"]):
                    result = self.run_hook(data, env)
                    self.assertEqual(result.returncode, 0, result.stderr)
                    self.assertFalse(self.is_blocked(result), result.stdout)

    def test_blocks_everything_when_python_missing(self):
        """Python が無いと fail-closed で exit 2（リポジトリ内でもブロック）."""
        data = {"tool_name": "Read", "tool_input": {"file_path": self.INSIDE}, "cwd": REPO_ROOT}
        self.fake.dummy("python3")
        result = self.run_hook(data, make_env([self.fake.dir]))
        self.assertEqual(result.returncode, 2)
        self.assertTrue(result.stderr.strip())


# ---------------------------------------------------------------------------
# notify.py をラッパー経由で起動（ntfy.sh には送らない）
# ---------------------------------------------------------------------------
class _Capture(http.server.BaseHTTPRequestHandler):
    received: list = []

    def do_POST(self):  # noqa: N802
        length = int(self.headers.get("Content-Length", 0))
        _Capture.received.append(self.rfile.read(length))
        self.send_response(200)
        self.end_headers()

    def log_message(self, *args):  # 出力を抑制
        pass


class TestNotifyViaWrapper(TempDirMixin, unittest.TestCase):
    def run_notify(self, data: dict, env: dict) -> subprocess.CompletedProcess:
        return subprocess.run(
            [BASH, ".claude/hooks/run_python.sh", ".claude/hooks/notify.py"],
            input=json.dumps(data).encode("utf-8"),
            capture_output=True,
            env=env,
            cwd=REPO_ROOT,
            timeout=TIMEOUT,
        )

    def test_no_topic_does_nothing(self):
        env = dict(os.environ)
        env.pop("NTFY_TOPIC", None)
        env["NTFY_SERVER"] = "http://127.0.0.1:9"  # 万一送信しても外部に出ない
        result = self.run_notify({"hook_event_name": "Stop", "cwd": REPO_ROOT}, env)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(result.stdout, b"")
        self.assertEqual(result.stderr, b"")

    def test_python_missing_is_non_blocking(self):
        self.fake.dummy("python3")
        env = make_env([self.fake.dir])
        env.pop("NTFY_TOPIC", None)
        result = self.run_notify({"hook_event_name": "Stop"}, env)
        self.assertEqual(result.returncode, 1)
        self.assertNotEqual(result.returncode, 2)

    def test_sends_to_local_server_with_repo_name(self):
        """ローカル HTTP サーバー宛てに送信し，Python 3.7 互換化後も内容が正しいこと."""
        _Capture.received = []
        server = http.server.HTTPServer(("127.0.0.1", 0), _Capture)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            repo = os.path.join(self.work, "folder-name")
            os.makedirs(repo)
            subprocess.run(["git", "init", "-q", repo], check=True, capture_output=True)
            subprocess.run(
                ["git", "-C", repo, "remote", "add", "origin", "git@github.com:owner/my-repo.git"],
                check=True,
                capture_output=True,
            )
            self.fake.dummy("python3")
            self.fake.real("python")
            env = make_env(
                [self.fake.dir] + bash_bin_dirs(),
                extra={
                    "NTFY_TOPIC": "test-topic",
                    "NTFY_SERVER": f"http://127.0.0.1:{server.server_address[1]}",
                },
            )
            result = self.run_notify({"hook_event_name": "Stop", "cwd": repo}, env)
            self.assertEqual(result.returncode, 0, result.stderr)
        finally:
            server.shutdown()
            server.server_close()
        self.assertEqual(len(_Capture.received), 1)
        payload = json.loads(_Capture.received[0].decode("utf-8"))
        self.assertEqual(payload["topic"], "test-topic")
        self.assertEqual(payload["title"], "Claude Code (my-repo)")


# ---------------------------------------------------------------------------
# notify.project_name（removesuffix 置換後の挙動）
# ---------------------------------------------------------------------------
class TestNotifyProjectName(TempDirMixin, unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        sys.path.insert(0, HOOKS_DIR)
        try:
            import notify  # noqa: WPS433
        finally:
            sys.path.pop(0)
        cls.notify = notify

    def repo_with_origin(self, url: str | None) -> str:
        repo = os.path.join(self.work, "local-folder")
        os.makedirs(repo, exist_ok=True)
        subprocess.run(["git", "init", "-q", repo], check=True, capture_output=True)
        if url is not None:
            subprocess.run(["git", "-C", repo, "remote", "add", "origin", url], check=True, capture_output=True)
        return repo

    def test_cases(self):
        cases = [
            ("https://github.com/owner/repo.git", "repo"),
            ("https://github.com/owner/repo", "repo"),
            ("https://github.com/owner/repo/", "repo"),
            ("git@github.com:owner/repo.git", "repo"),
            ("git@github.com:owner/repo", "repo"),
            ("https://github.com/owner/my.git.repo.git", "my.git.repo"),
            ("https://github.com/owner/gitrepo", "gitrepo"),
        ]
        for url, expected in cases:
            with self.subTest(url=url):
                shutil.rmtree(os.path.join(self.work, "local-folder"), ignore_errors=True)
                repo = self.repo_with_origin(url)
                self.assertEqual(self.notify.project_name(repo), expected)

    def test_no_origin_returns_folder(self):
        repo = self.repo_with_origin(None)
        self.assertEqual(self.notify.project_name(repo), "local-folder")

    def test_only_dotgit_falls_back_to_folder(self):
        repo = self.repo_with_origin("https://example.com/.git")
        self.assertEqual(self.notify.project_name(repo), "local-folder")


# ---------------------------------------------------------------------------
# notify.build_notification（どのイベントで通知するか）
# ---------------------------------------------------------------------------
def _subagent_task(agent_type: str = "coder") -> dict:
    return {"id": "t1", "type": "subagent", "status": "running", "description": "...", "agent_type": agent_type}


class TestNotifyBuildNotification(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        sys.path.insert(0, HOOKS_DIR)
        try:
            import notify  # noqa: WPS433
        finally:
            sys.path.pop(0)
        cls.notify = notify

    def build(self, data: dict):
        return self.notify.build_notification(data)

    # Stop: 依頼したエージェントが動いている間は通知しない
    def test_stop_without_background_tasks_notifies(self):
        self.assertIsNotNone(self.build({"hook_event_name": "Stop"}))

    def test_stop_with_empty_background_tasks_notifies(self):
        self.assertIsNotNone(self.build({"hook_event_name": "Stop", "background_tasks": []}))

    def test_stop_with_running_subagent_is_silent(self):
        self.assertIsNone(self.build({"hook_event_name": "Stop", "background_tasks": [_subagent_task()]}))

    def test_stop_with_running_workflow_is_silent(self):
        task = {"id": "w1", "type": "workflow", "status": "running", "description": "...", "name": "wf"}
        self.assertIsNone(self.build({"hook_event_name": "Stop", "background_tasks": [task]}))

    def test_stop_with_shell_task_still_notifies(self):
        """開発サーバー等の常駐 shell / monitor で通知が止まらないこと."""
        for kind in ("shell", "monitor"):
            with self.subTest(kind=kind):
                task = {"id": "s1", "type": kind, "status": "running", "description": "..."}
                self.assertIsNotNone(self.build({"hook_event_name": "Stop", "background_tasks": [task]}))

    def test_stop_with_mixed_tasks_is_silent(self):
        tasks = [{"id": "s1", "type": "shell", "status": "running", "description": "..."}, _subagent_task()]
        self.assertIsNone(self.build({"hook_event_name": "Stop", "background_tasks": tasks}))

    def test_stop_with_malformed_background_tasks_notifies(self):
        """想定外の形（古いバージョン・壊れた入力）でも通知を落とさない."""
        for tasks in ("not-a-list", {"type": "subagent"}, [None, "x", 1], [{}]):
            with self.subTest(tasks=tasks):
                self.assertIsNotNone(self.build({"hook_event_name": "Stop", "background_tasks": tasks}))

    # 待ち状態の通知はエージェント名を添える
    def test_permission_prompt_from_subagent_names_agent(self):
        data = {"hook_event_name": "Notification", "notification_type": "permission_prompt",
                "agent_id": "a1", "agent_type": "coder"}
        self.assertIn("coder", self.build(data)["message"])

    def test_permission_prompt_from_main_has_no_agent_label(self):
        data = {"hook_event_name": "Notification", "notification_type": "permission_prompt"}
        self.assertNotIn("エージェント", self.build(data)["message"])

    def test_question_from_subagent_names_agent(self):
        data = {"hook_event_name": "PreToolUse", "tool_name": "AskUserQuestion", "agent_type": "tester"}
        self.assertIn("tester", self.build(data)["message"])

    def test_agent_name_is_trimmed_and_capped(self):
        data = {"hook_event_name": "Notification", "notification_type": "permission_prompt",
                "agent_type": "  a" + chr(10) + "b  " + "x" * 200}
        message = self.build(data)["message"]
        self.assertNotIn(chr(10), message)
        self.assertIn("a b", message)
        self.assertLess(len(message), 80)

    def test_blank_agent_type_is_ignored(self):
        for value in ("", "   ", None, 123):
            with self.subTest(value=value):
                data = {"hook_event_name": "Notification", "notification_type": "idle_prompt", "agent_type": value}
                self.assertNotIn("エージェント", self.build(data)["message"])

    def test_unknown_event_is_silent(self):
        self.assertIsNone(self.build({"hook_event_name": "SubagentStop"}))
        self.assertIsNone(self.build({"hook_event_name": "PreToolUse", "tool_name": "Bash"}))


# ---------------------------------------------------------------------------
# Python 3.7 互換（静的チェック）
# ---------------------------------------------------------------------------
PY39_ONLY_ATTRS = {"removesuffix", "removeprefix", "is_relative_to", "with_stem", "readlink", "randbytes", "lcm", "cache"}
PY38_PLUS_MODULES = {"zoneinfo", "graphlib", "tomllib", "importlib.metadata"}
GENERIC_BUILTINS = {"list", "dict", "tuple", "set", "frozenset", "type"}


class TestPython37Compat(unittest.TestCase):
    def check_file(self, path: str) -> None:
        with open(path, encoding="utf-8") as f:
            source = f.read()
        # 3.8+ 専用構文（walrus・位置専用引数等）を検出する
        tree = ast.parse(source, filename=path, feature_version=(3, 7))

        # from __future__ import annotations（docstring の直後）
        body = tree.body
        idx = 1 if body and isinstance(body[0], ast.Expr) and isinstance(getattr(body[0], "value", None), ast.Constant) else 0
        first = body[idx]
        self.assertIsInstance(first, ast.ImportFrom, "先頭に from __future__ import annotations が必要")
        self.assertEqual(first.module, "__future__")
        self.assertIn("annotations", [a.name for a in first.names])

        # アノテーション以外（実行時に評価される箇所）での list[...] / X | None を検出
        annotation_nodes = set()
        for node in ast.walk(tree):
            anns = []
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                anns.append(node.returns)
                a = node.args
                for arg in a.args + a.kwonlyargs + getattr(a, "posonlyargs", []) + [a.vararg, a.kwarg]:
                    if arg is not None:
                        anns.append(arg.annotation)
            elif isinstance(node, ast.AnnAssign):
                anns.append(node.annotation)
            for ann in anns:
                if ann is not None:
                    annotation_nodes.update(id(n) for n in ast.walk(ann))

        problems = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Attribute) and node.attr in PY39_ONLY_ATTRS:
                problems.append(f"{node.lineno}: .{node.attr}")
            if isinstance(node, (ast.Import, ast.ImportFrom)):
                names = [node.module] if isinstance(node, ast.ImportFrom) else [a.name for a in node.names]
                for name in names:
                    if name in PY38_PLUS_MODULES:
                        problems.append(f"{node.lineno}: import {name}")
            if id(node) in annotation_nodes:
                continue
            if isinstance(node, ast.Subscript) and isinstance(node.value, ast.Name) and node.value.id in GENERIC_BUILTINS:
                problems.append(f"{node.lineno}: 実行時の {node.value.id}[...]")
            if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr):
                if any(isinstance(s, ast.Constant) and s.value is None for s in (node.left, node.right)):
                    problems.append(f"{node.lineno}: 実行時の X | None")
            if hasattr(ast, "Match") and isinstance(node, ast.Match):
                problems.append(f"{node.lineno}: match 文")
        self.assertEqual(problems, [], f"{os.path.basename(path)} に 3.7 非互換の記述")

    def test_notify_py37_compatible(self):
        self.check_file(NOTIFY)

    def test_restrict_repo_access_py37_compatible(self):
        self.check_file(RESTRICT)


# ---------------------------------------------------------------------------
# check_sync.sh: python3 決め打ちではなくラッパー経由
# ---------------------------------------------------------------------------
# check_sync.sh は team 層のファイルで，solo モードのプロジェクトには配置されない．
# 一方このテスト自体は共通層として全プロジェクトに同期されるため，存在ガードで skip する．
@unittest.skipUnless(os.path.exists(CHECK_SYNC), "check_sync.sh は team 層のため solo には無い")
class TestCheckSync(TempDirMixin, unittest.TestCase):
    def test_no_hardcoded_python3(self):
        with open(CHECK_SYNC, encoding="utf-8") as f:
            lines = [l for l in f.read().splitlines() if not l.lstrip().startswith("#")]
        code = "\n".join(lines)
        self.assertIsNone(re.search(r"(^|[\s;|&(])(python3?|py)\s", code), "python コマンドの直接呼び出しが残っている")
        self.assertIn("run_python.sh", code)

    def test_emits_json_with_dummy_python3(self):
        """ダミー python3 しか無い環境（Windows 相当）でも JSON を出力する."""
        # origin が存在しないパスなので fetch が失敗し emit が呼ばれる
        repo = os.path.join(self.work, "repo")
        os.makedirs(repo)
        subprocess.run(["git", "init", "-q", repo], check=True, capture_output=True)
        subprocess.run(
            ["git", "-C", repo, "remote", "add", "origin", os.path.join(self.work, "no-such-remote")],
            check=True,
            capture_output=True,
        )
        self.fake.dummy("python3")
        self.fake.real("python")
        env = make_env([self.fake.dir] + bash_bin_dirs(), drop=("PYTHONIOENCODING",))
        result = subprocess.run(
            [BASH, CHECK_SYNC],
            capture_output=True,
            env=env,
            cwd=repo,
            timeout=TIMEOUT,
        )
        self.assertEqual(result.returncode, 0, result.stderr)
        out = json.loads(result.stdout.decode("utf-8"))
        self.assertIn("[sync-check]", out["systemMessage"])
        self.assertEqual(out["hookSpecificOutput"]["hookEventName"], "SessionStart")
        self.assertEqual(out["hookSpecificOutput"]["additionalContext"], out["systemMessage"])

    def test_python_missing_does_not_block_session(self):
        repo = os.path.join(self.work, "repo")
        os.makedirs(repo)
        subprocess.run(["git", "init", "-q", repo], check=True, capture_output=True)
        subprocess.run(
            ["git", "-C", repo, "remote", "add", "origin", os.path.join(self.work, "no-such-remote")],
            check=True,
            capture_output=True,
        )
        self.fake.dummy("python3")
        env = make_env([self.fake.dir] + bash_bin_dirs())
        # bash_bin_dirs に python が無い前提（あればこのテストは意味をなさないのでスキップ）
        for d in bash_bin_dirs():
            for n in ("python3", "python", "py"):
                if os.path.exists(os.path.join(d, n)) or os.path.exists(os.path.join(d, n + ".exe")):
                    self.skipTest(f"{d} に {n} がある")
        result = subprocess.run([BASH, CHECK_SYNC], capture_output=True, env=env, cwd=repo, timeout=TIMEOUT)
        self.assertEqual(result.returncode, 0, result.stderr)


# ---------------------------------------------------------------------------
# restrict_repo_access.py: 判定ロジック（Issue #49 の仕様）
# ---------------------------------------------------------------------------
# 「リポジトリ外かつ一時ディレクトリ外の既存ファイル」が必要なため，このテストのディレクトリ配下に
# 使い捨てのサンドボックスを作り，その中に「仮のリポジトリ」「リポジトリ外」「仮のホーム」を
# 置いて判定関数に repo_root を直接渡す（ユーザーの実ファイルは作成・変更・削除しない）．
# テストは判定関数を呼ぶだけで，実際の削除・上書きは行わない．
TESTS_DIR = os.path.dirname(os.path.abspath(__file__))


def _load_restrict():
    sys.path.insert(0, HOOKS_DIR)
    try:
        import restrict_repo_access  # noqa: WPS433
    finally:
        sys.path.pop(0)
    return restrict_repo_access


def _fwd(path: str) -> str:
    """Bash のコマンド文字列に埋め込む形（区切りを / に）."""
    return path.replace("\\", "/")


def _gitbash(path: str) -> str:
    """Windows の C:\\x を Git Bash 形式 /c/x にする."""
    p = _fwd(path)
    if re.match(r"^[A-Za-z]:/", p):
        return "/" + p[0].lower() + p[2:]
    return p


class RestrictSandboxMixin:
    """sandbox/
         repo/            仮のリポジトリ（repo_root）．sub/・file.txt・.claude/
         outside/         リポジトリ外（一時ディレクトリ外）
           existing.txt   既存ファイル
           dir/file.txt   既存ディレクトリ内の既存ファイル
           other/data/    削除対象のディレクトリ
           archive_dest/  既存の展開先ディレクトリ
         home/            仮のホーム（~・$HOME）．.bashrc・.ssh/id_rsa・source/other/
    """

    @classmethod
    def setUpClass(cls) -> None:
        cls.rra = _load_restrict()

    def setUp(self) -> None:
        self.base = tempfile.mkdtemp(prefix="_restrict_sandbox_", dir=TESTS_DIR)
        if self.rra.is_temp_path(self.base):
            shutil.rmtree(self.base, ignore_errors=True)
            self.skipTest("リポジトリが一時ディレクトリ配下にあり，一時ディレクトリ外のリポジトリ外パスを用意できない")
        self.repo = os.path.join(self.base, "repo")
        self.sub = os.path.join(self.repo, "sub")
        self.out = os.path.join(self.base, "outside")
        self.home = os.path.join(self.base, "home")
        for d in (self.sub, os.path.join(self.repo, ".claude"), os.path.join(self.out, "dir"),
                  os.path.join(self.out, "other", "data"), os.path.join(self.out, "archive_dest"),
                  os.path.join(self.home, ".ssh"), os.path.join(self.home, "source", "other")):
            os.makedirs(d)
        for f in (os.path.join(self.repo, "file.txt"), os.path.join(self.out, "existing.txt"),
                  os.path.join(self.out, "dir", "file.txt"), os.path.join(self.home, ".bashrc"),
                  os.path.join(self.home, ".ssh", "id_rsa"), os.path.join(self.out, "other", "data", "a.txt")):
            with open(f, "w", encoding="utf-8") as fh:
                fh.write("x")
        self.existing = os.path.join(self.out, "existing.txt")
        self.missing = os.path.join(self.out, "new_file.txt")
        self.deny = []

    def tearDown(self) -> None:
        shutil.rmtree(self.base, ignore_errors=True)

    # --- 判定ヘルパー ---
    def bash(self, command: str, cwd: str | None = None):
        return self.rra.check_bash_command(command, cwd or self.repo, repo_root=self.repo, deny_read=self.deny)

    def ps(self, command: str, cwd: str | None = None):
        return self.rra.check_powershell_command(command, cwd or self.repo, repo_root=self.repo, deny_read=self.deny)

    def tool(self, name: str, tool_input: dict, cwd: str | None = None):
        policy = self.rra.Policy(self.repo, self.deny)
        return self.rra.check_file_tool(name, tool_input, cwd or self.repo, policy)

    def fake_home(self):
        return mock.patch.dict(os.environ, {"HOME": self.home, "USERPROFILE": self.home})

    def run_main(self, data: dict, env: dict) -> subprocess.CompletedProcess:
        return subprocess.run([sys.executable, "-B", RESTRICT], input=json.dumps(data).encode("utf-8"),
                              capture_output=True, env=env, cwd=self.repo, timeout=TIMEOUT)

    @staticmethod
    def decision(result):
        out = result.stdout.decode("utf-8").strip()
        if not out:
            return None
        return json.loads(out)["hookSpecificOutput"]["permissionDecision"]

    def assertDenied(self, reason, label=""):
        self.assertIsNotNone(reason, f"拒否されるべき: {label}")

    def assertAllowed(self, reason, label=""):
        self.assertIsNone(reason, f"許可されるべき: {label} / 理由: {reason}")


class TestRestrictIssueExamples(RestrictSandboxMixin, unittest.TestCase):
    """Issue #49 の取りこぼし例がすべて拒否されること."""

    def test_powershell_remove_item_recurse(self):
        target = os.path.join(self.out, "other")
        self.assertDenied(self.ps(f"Remove-Item {target} -Recurse"))
        self.assertDenied(self.ps(f"Remove-Item {target} -Recurse -Force"))

    def test_cd_outside_then_rm_relative(self):
        self.assertDenied(self.bash(f'cd "{_fwd(os.path.join(self.out, "other"))}" && rm -rf data'))

    def test_rm_rf_home_subdir(self):
        with self.fake_home():
            self.assertDenied(self.bash("rm -rf ~/source/other"))
        # 実際のホームでも（削除は存在に関係なく拒否）
        self.assertDenied(self.bash("rm -rf ~/source/__no_such_dir_for_test__"))

    def test_overwrite_existing_file_variants(self):
        ex = _fwd(self.existing)
        cases = {
            "sed -i": f"sed -i 's/a/b/' \"{ex}\"",
            ">": f"echo x > \"{ex}\"",
            "tee": f"echo x | tee \"{ex}\"",
            "truncate": f"truncate -s 0 \"{ex}\"",
        }
        for label, cmd in cases.items():
            with self.subTest(label):
                self.assertDenied(self.bash(cmd), label)

    def test_python_inline_rmtree(self):
        target = _fwd(os.path.join(self.out, "other"))
        self.assertDenied(self.bash(f"python -c \"import shutil; shutil.rmtree('{target}')\""))


class TestRestrictBaseAndZones(RestrictSandboxMixin, unittest.TestCase):
    """基準（リポジトリのルート・cwd 起点の相対パス）と許可ゾーン."""

    def test_relative_from_subfolder_inside_repo_is_allowed(self):
        self.assertAllowed(self.bash("rm ../file.txt", cwd=self.sub))
        self.assertAllowed(self.ps("Remove-Item ..\\file.txt", cwd=self.sub))

    def test_relative_from_subfolder_escaping_repo_is_denied(self):
        self.assertDenied(self.bash("rm ../../outside/existing.txt", cwd=self.sub))
        self.assertDenied(self.ps("Remove-Item ..\\..\\outside\\existing.txt", cwd=self.sub))

    def test_inside_repo_delete_and_overwrite_allowed(self):
        f = _fwd(os.path.join(self.repo, "file.txt"))
        for cmd in (f'rm -rf "{f}"', f'echo x > "{f}"', f'sed -i s/a/b/ "{f}"', "rm -rf build"):
            with self.subTest(cmd):
                self.assertAllowed(self.bash(cmd))
        self.assertAllowed(self.tool("Edit", {"file_path": os.path.join(self.repo, "file.txt")}))
        self.assertAllowed(self.tool("Write", {"file_path": os.path.join(self.repo, "file.txt")}))

    def test_temp_dir_allowed(self):
        tmp = tempfile.mkdtemp(prefix="hooktest_zone_")
        try:
            existing = os.path.join(tmp, "e.txt")
            with open(existing, "w") as fh:
                fh.write("x")
            for cmd in (f'rm -rf "{_fwd(tmp)}"', f'echo x > "{_fwd(existing)}"', "rm -rf /tmp/some-dir"):
                with self.subTest(cmd):
                    self.assertAllowed(self.bash(cmd))
            self.assertAllowed(self.ps(f"Remove-Item {tmp} -Recurse -Force"))
            self.assertAllowed(self.tool("Edit", {"file_path": existing}))
            self.assertAllowed(self.tool("Write", {"file_path": existing}))
        finally:
            shutil.rmtree(tmp, ignore_errors=True)

    def test_special_paths_allowed(self):
        for cmd in ("echo x > /dev/null", "ls 2> /dev/null", "cp file.txt /dev/null", "echo x &> /dev/null"):
            with self.subTest(cmd):
                self.assertAllowed(self.bash(cmd))
        for cmd in ("Get-Item x > $null", "echo x | Out-File NUL", "echo x | Out-File $null", "echo x > NUL"):
            with self.subTest(cmd):
                self.assertAllowed(self.ps(cmd))

    def test_main_uses_claude_project_dir_and_cwd(self):
        """CLAUDE_PROJECT_DIR を基準に，相対パスは入力の cwd 起点で解決する（main 経由）."""
        env = dict(os.environ, CLAUDE_PROJECT_DIR=self.repo)
        allow = self.run_main({"tool_name": "Bash", "tool_input": {"command": "rm ../file.txt"}, "cwd": self.sub}, env)
        self.assertEqual(allow.returncode, 0, allow.stderr)
        self.assertIsNone(self.decision(allow))
        deny = self.run_main({"tool_name": "Bash", "tool_input": {"command": "rm ../../outside/existing.txt"}, "cwd": self.sub}, env)
        self.assertEqual(self.decision(deny), "deny")

    def test_main_falls_back_to_cwd_without_project_dir(self):
        env = dict(os.environ)
        env.pop("CLAUDE_PROJECT_DIR", None)
        allow = self.run_main({"tool_name": "Bash", "tool_input": {"command": "rm -rf build"}, "cwd": self.repo}, env)
        self.assertIsNone(self.decision(allow))
        deny = self.run_main({"tool_name": "Bash", "tool_input": {"command": "rm -rf ../outside/other"}, "cwd": self.repo}, env)
        self.assertEqual(self.decision(deny), "deny")

    def test_main_powershell_and_notebook_tools(self):
        env = dict(os.environ, CLAUDE_PROJECT_DIR=self.repo)
        cases = [
            {"tool_name": "PowerShell", "tool_input": {"command": f"Remove-Item {self.out} -Recurse"}},
            {"tool_name": "NotebookEdit", "tool_input": {"notebook_path": os.path.join(self.out, "n.ipynb"), "new_source": "x"}},
        ]
        for data in cases:
            with self.subTest(data["tool_name"]):
                data["cwd"] = self.repo
                self.assertEqual(self.decision(self.run_main(data, env)), "deny")


class TestRestrictFileTools(RestrictSandboxMixin, unittest.TestCase):
    def test_read_glob_grep_outside_allowed(self):
        self.assertAllowed(self.tool("Read", {"file_path": self.existing}))
        self.assertAllowed(self.tool("Glob", {"pattern": "**/*", "path": self.out}))
        self.assertAllowed(self.tool("Glob", {"pattern": _fwd(self.out) + "/**/*.txt"}))
        self.assertAllowed(self.tool("Grep", {"pattern": "x", "path": self.out}))

    def test_write_outside_new_allowed_existing_denied(self):
        self.assertAllowed(self.tool("Write", {"file_path": self.missing, "content": "x"}))
        self.assertDenied(self.tool("Write", {"file_path": self.existing, "content": "x"}))

    def test_edit_and_notebook_edit_outside_always_denied(self):
        for path in (self.existing, self.missing):
            with self.subTest(path=path):
                self.assertDenied(self.tool("Edit", {"file_path": path, "old_string": "a", "new_string": "b"}))
                self.assertDenied(self.tool("NotebookEdit", {"notebook_path": path + ".ipynb", "new_source": "x"}))

    def test_relative_tool_path_resolved_from_cwd(self):
        self.assertDenied(self.tool("Write", {"file_path": "../../outside/existing.txt"}, cwd=self.sub))
        self.assertAllowed(self.tool("Edit", {"file_path": "../file.txt"}, cwd=self.sub))

    def test_settings_matcher_covers_all_target_tools(self):
        with open(SETTINGS, encoding="utf-8") as f:
            settings = json.load(f)
        matchers = [g.get("matcher", "") for g in settings["hooks"]["PreToolUse"]
                    if any("restrict_repo_access.py" in h.get("command", "") for h in g.get("hooks", []))]
        self.assertEqual(len(matchers), 1)
        for tool in ("Read", "Write", "Edit", "NotebookEdit", "Glob", "Grep", "Bash", "PowerShell"):
            self.assertIn(tool, matchers[0].split("|"))


class TestRestrictDenyRead(RestrictSandboxMixin, unittest.TestCase):
    def write_config(self, text: str) -> None:
        with open(os.path.join(self.repo, ".claude", "repo-access.json"), "w", encoding="utf-8") as fh:
            fh.write(text)

    def test_load_absolute_home_and_relative_entries(self):
        abs_entry = _fwd(os.path.join(self.out, "dir"))
        self.write_config(json.dumps({"deny_read": [abs_entry, "~/.ssh", "secrets"]}))
        with self.fake_home():
            entries = self.rra.load_deny_read(self.repo)
        normed = [os.path.normcase(os.path.normpath(e)) for e in entries]
        for expected in (os.path.join(self.out, "dir"), os.path.join(self.home, ".ssh"), os.path.join(self.repo, "secrets")):
            self.assertIn(os.path.normcase(os.path.normpath(expected)), normed)

    @unittest.skipUnless(os.name == "nt", "Git Bash 形式のパスは Windows のみ")
    def test_load_gitbash_form_entry(self):
        self.write_config(json.dumps({"deny_read": [_gitbash(os.path.join(self.out, "dir"))]}))
        entries = self.rra.load_deny_read(self.repo)
        self.assertEqual([os.path.normcase(e) for e in entries], [os.path.normcase(os.path.join(self.out, "dir"))])

    def test_missing_config_is_empty(self):
        self.assertEqual(self.rra.load_deny_read(self.repo), [])

    def test_broken_config_is_empty_with_warning(self):
        for text in ("{not json", json.dumps({"deny_read": "~/.ssh"}), json.dumps({"deny_read": [1, 2]}), json.dumps([1])):
            with self.subTest(text=text):
                self.write_config(text)
                err = io.StringIO()
                with contextlib.redirect_stderr(err):
                    self.assertEqual(self.rra.load_deny_read(self.repo), [])
                self.assertTrue(err.getvalue().strip(), "stderr に警告が出ること")

    def test_broken_config_via_main_does_not_block(self):
        self.write_config("{broken")
        env = dict(os.environ, CLAUDE_PROJECT_DIR=self.repo)
        result = self.run_main({"tool_name": "Read", "tool_input": {"file_path": self.existing}, "cwd": self.repo}, env)
        self.assertEqual(result.returncode, 0)
        self.assertIsNone(self.decision(result))
        self.assertTrue(result.stderr.strip())

    def test_config_via_main_denies_read(self):
        self.write_config(json.dumps({"deny_read": [_fwd(os.path.join(self.out, "dir"))]}))
        env = dict(os.environ, CLAUDE_PROJECT_DIR=self.repo)
        data = {"tool_name": "Read", "tool_input": {"file_path": os.path.join(self.out, "dir", "file.txt")}, "cwd": self.repo}
        self.assertEqual(self.decision(self.run_main(data, env)), "deny")

    def test_denied_path_read_and_write_rejected(self):
        denied_dir = os.path.join(self.out, "dir")
        self.deny = [denied_dir]
        f = os.path.join(denied_dir, "file.txt")
        self.assertDenied(self.tool("Read", {"file_path": f}))
        self.assertDenied(self.tool("Write", {"file_path": os.path.join(denied_dir, "new.txt")}), "新規作成でも拒否")
        self.assertDenied(self.bash(f'cat "{_fwd(f)}"'))
        self.assertDenied(self.bash(f'echo x >> "{_fwd(os.path.join(denied_dir, "new.txt"))}"'))
        self.assertDenied(self.ps(f"Get-Content {f}"))
        # 禁止されていない兄弟は読める
        self.assertAllowed(self.tool("Read", {"file_path": self.existing}))
        self.assertAllowed(self.bash(f'cat "{_fwd(self.existing)}"'))

    def test_denied_path_inside_repo(self):
        self.deny = [os.path.join(self.repo, "secrets")]
        self.assertDenied(self.tool("Read", {"file_path": os.path.join(self.repo, "secrets", "k.txt")}))
        self.assertDenied(self.bash("cat secrets/k.txt"))

    def test_search_root_containing_denied_path_rejected(self):
        self.deny = [os.path.join(self.out, "dir")]
        self.assertDenied(self.tool("Grep", {"pattern": "x", "path": self.out}))
        self.assertDenied(self.tool("Glob", {"pattern": "**/*", "path": self.out}))
        self.assertDenied(self.tool("Glob", {"pattern": _fwd(self.out) + "/**/*.txt"}))
        self.assertDenied(self.bash(f'grep -r x "{_fwd(self.out)}"'))
        self.assertDenied(self.bash(f'find "{_fwd(self.out)}" -name "*.txt"'))
        self.assertDenied(self.ps(f"Get-ChildItem {self.out} -Recurse"))
        # 起点が禁止パスを含まない場所なら許可
        self.assertAllowed(self.tool("Grep", {"pattern": "x", "path": os.path.join(self.out, "other")}))

    def test_home_entry_denies_tilde_access(self):
        with self.fake_home():
            self.deny = [os.path.join(self.home, ".ssh")]
            self.assertDenied(self.bash("cat ~/.ssh/id_rsa"))
            self.assertDenied(self.bash("cat $HOME/.ssh/id_rsa"))
            self.assertDenied(self.tool("Read", {"file_path": "~/.ssh/id_rsa"}))


class TestRestrictCreateAllowed(RestrictSandboxMixin, unittest.TestCase):
    """新規作成・ダウンロード・追記・権限変更はリポジトリ外でも許可."""

    def test_bash_create_append_chmod(self):
        ex, new = _fwd(self.existing), _fwd(self.missing)
        newdir = _fwd(os.path.join(self.out, "newdir", "a"))
        cases = [
            f'mkdir -p "{newdir}"',
            f'touch "{new}"',
            f'touch "{ex}"',
            f'echo x >> "{ex}"',
            f'echo x | tee -a "{ex}"',
            f'curl -o "{new}" https://example.com/x',
            f'curl -fsSLo "{new}" https://example.com/x',
            f'wget -O "{new}" https://example.com/x',
            f'chmod 600 "{ex}"',
            f'echo x > "{new}"',
            f'cp file.txt "{new}"',
            f'cp -n file.txt "{ex}"',
            f'cat "{ex}"',
        ]
        for cmd in cases:
            with self.subTest(cmd):
                self.assertAllowed(self.bash(cmd), cmd)

    def test_powershell_create_append_acl(self):
        ex, new = self.existing, self.missing
        cases = [
            f"New-Item -ItemType Directory {os.path.join(self.out, 'newdir')}",
            f"New-Item -ItemType Directory -Force {os.path.join(self.out, 'dir')}",
            f"New-Item {new}",
            f"Add-Content {ex} 'x'",
            f"Add-Content -Path {ex} -Value 'x'",
            f"'x' | Out-File {ex} -Append",
            f"'x' | Out-File -FilePath {ex} -Append",
            f"Invoke-WebRequest https://example.com/x -OutFile {new}",
            f"Set-Acl {ex} $acl",
            f"Get-Content {ex}",
            f"'x' > {new}",
            f"'x' >> {ex}",
            f"Set-Content {new} 'x'",
            f"Copy-Item file.txt {new}",
        ]
        for cmd in cases:
            with self.subTest(cmd):
                self.assertAllowed(self.ps(cmd), cmd)


class TestRestrictDelete(RestrictSandboxMixin, unittest.TestCase):
    """削除は（存在に関係なく）拒否."""

    def test_bash_delete_commands(self):
        ex = _fwd(self.existing)
        d = _fwd(os.path.join(self.out, "other"))
        cases = [
            f'rm "{ex}"', f'rm -f "{_fwd(self.missing)}"', f'rmdir "{d}"', f'unlink "{ex}"',
            f'truncate -s 0 "{ex}"', f'find "{d}" -delete', f'find "{d}" -name "*.txt" -exec rm {{}} \\;',
            f'mv "{ex}" ./moved.txt', f'sudo rm -rf "{d}"',
        ]
        for cmd in cases:
            with self.subTest(cmd):
                self.assertDenied(self.bash(cmd), cmd)

    def test_powershell_delete_commands(self):
        ex = self.existing
        d = os.path.join(self.out, "other")
        cases = [
            f"Remove-Item {ex}", f"Remove-Item -Path {d} -Recurse -Force", f"Remove-Item -LiteralPath {ex}",
            f"del {ex}", f"rd {d} -Recurse", f"ri {ex}", f"rm {ex}", f"erase {ex}", f"rmdir {d}",
            f"Clear-Content {ex}", f"Move-Item {ex} .\\moved.txt", f"Move-Item -Path {ex} -Destination .\\m.txt",
            f"Rename-Item {ex} renamed.txt", f"[IO.File]::Delete('{ex}')", f"[System.IO.File]::Delete(\"{ex}\")",
            f"cmd /c del {ex}", f"cmd /c rd /s /q {d}",
        ]
        for cmd in cases:
            with self.subTest(cmd):
                self.assertDenied(self.ps(cmd), cmd)

    def test_inline_delete_calls(self):
        d = _fwd(os.path.join(self.out, "other"))
        ex = _fwd(self.existing)
        cases = [
            f"python -c \"import shutil; shutil.rmtree('{d}')\"",
            f"python3 -c \"import os; os.remove('{ex}')\"",
            f"py -c \"import os; os.unlink('{ex}')\"",
            f"node -e \"require('fs').rmSync('{d}', {{recursive: true}})\"",
        ]
        for cmd in cases:
            with self.subTest(cmd):
                self.assertDenied(self.bash(cmd), cmd)
        self.assertDenied(self.ps(f"python -c \"import shutil; shutil.rmtree('{d}')\""))

    def test_inline_without_delete_or_inside_repo_allowed(self):
        ex = _fwd(self.existing)
        self.assertAllowed(self.bash(f"python -c \"print(open('{ex}').read())\""))
        self.assertAllowed(self.bash("python -c \"import shutil; shutil.rmtree('build')\""))


class TestRestrictOverwrite(RestrictSandboxMixin, unittest.TestCase):
    """既存なら拒否，存在しなければ許可."""

    BASH_TEMPLATES = [
        "echo x > \"{t}\"",
        "echo x >| \"{t}\"",
        "cp file.txt \"{t}\"",
        "mv file.txt \"{t}\"",
        "sed -i 's/a/b/' \"{t}\"",
        "sed -i.bak -e 's/a/b/' \"{t}\"",
        "echo x | tee \"{t}\"",
        "dd if=file.txt of=\"{t}\"",
        "curl -o \"{t}\" https://example.com/x",
        "curl --output \"{t}\" https://example.com/x",
        "wget -O \"{t}\" https://example.com/x",
        "python -c \"open('{t}', 'w').write('x')\"",
    ]
    PS_TEMPLATES = [
        "'x' > {t}",
        "Copy-Item file.txt {t}",
        "Copy-Item -Path file.txt -Destination {t}",
        "Move-Item file.txt {t}",
        "Set-Content {t} 'x'",
        "Set-Content -Path {t} -Value 'x'",
        "'x' | Out-File {t}",
        "'x' | Out-File -FilePath {t}",
        "New-Item {t} -Force",
        "Invoke-WebRequest https://example.com/x -OutFile {t}",
        "iwr https://example.com/x -OutFile {t}",
    ]

    def test_bash_overwrite(self):
        for tpl in self.BASH_TEMPLATES:
            with self.subTest(tpl, state="existing"):
                self.assertDenied(self.bash(tpl.format(t=_fwd(self.existing))), tpl)
            with self.subTest(tpl, state="missing"):
                self.assertAllowed(self.bash(tpl.format(t=_fwd(self.missing))), tpl)

    def test_powershell_overwrite(self):
        for tpl in self.PS_TEMPLATES:
            with self.subTest(tpl, state="existing"):
                self.assertDenied(self.ps(tpl.format(t=self.existing)), tpl)
            with self.subTest(tpl, state="missing"):
                self.assertAllowed(self.ps(tpl.format(t=self.missing)), tpl)

    def test_copy_into_existing_directory_uses_basename(self):
        d = os.path.join(self.out, "dir")  # dir/file.txt が既存
        with open(os.path.join(self.repo, "fresh.txt"), "w") as fh:
            fh.write("x")
        self.assertDenied(self.bash(f'cp file.txt "{_fwd(d)}"'))
        self.assertDenied(self.bash(f'cp file.txt "{_fwd(d)}/"'))
        self.assertAllowed(self.bash(f'cp fresh.txt "{_fwd(d)}"'))
        self.assertDenied(self.ps(f"Copy-Item file.txt {d}"))
        self.assertDenied(self.ps(f"Copy-Item -Path file.txt -Destination {d}"))
        self.assertAllowed(self.ps(f"Copy-Item fresh.txt {d}"))

    def test_cp_no_clobber_allowed(self):
        for opt in ("-n", "--no-clobber"):
            with self.subTest(opt):
                self.assertAllowed(self.bash(f'cp {opt} file.txt "{_fwd(self.existing)}"'))

    def test_tee_append_and_out_file_append_allowed(self):
        self.assertAllowed(self.bash(f'echo x | tee -a "{_fwd(self.existing)}"'))
        self.assertAllowed(self.ps(f"'x' | Out-File {self.existing} -Append"))

    def test_expand_archive_force(self):
        dest = os.path.join(self.out, "archive_dest")
        self.assertDenied(self.ps(f"Expand-Archive a.zip -DestinationPath {dest} -Force"))
        self.assertDenied(self.ps(f"Expand-Archive -Path a.zip -DestinationPath {dest} -Force"))
        self.assertAllowed(self.ps(f"Expand-Archive a.zip -DestinationPath {dest}"))
        self.assertAllowed(self.ps(f"Expand-Archive a.zip -DestinationPath {os.path.join(self.out, 'new_dest')} -Force"))

    def test_new_item_without_force_allowed(self):
        self.assertAllowed(self.ps(f"New-Item {self.existing}"))


class TestRestrictLocationTracking(RestrictSandboxMixin, unittest.TestCase):
    def test_bash_cd_variants(self):
        out = _fwd(self.out)
        repo = _fwd(self.repo)
        self.assertDenied(self.bash(f'cd "{out}"; echo x > existing.txt'))
        self.assertDenied(self.bash(f'cd "{out}" && rm -rf other'))
        self.assertAllowed(self.bash(f'cd "{out}" && cat existing.txt'))
        self.assertAllowed(self.bash(f'cd "{out}" && echo x > brand_new.txt'))
        self.assertAllowed(self.bash(f'cd "{out}" && cd "{repo}" && rm -rf build'))
        self.assertDenied(self.bash(f'pushd "{out}" && rm existing.txt'))
        self.assertAllowed(self.bash(f'pushd "{out}" && popd && rm -rf build'))
        self.assertDenied(self.bash("cd .. && rm -rf outside", cwd=self.repo))

    def test_command_substitution_cd_does_not_leak(self):
        out = _fwd(self.out)
        self.assertAllowed(self.bash(f'x=$(cd "{out}" && pwd); rm -rf build'))
        self.assertAllowed(self.bash(f'echo "$(cd "{out}" && pwd)" && rm -rf build'))
        self.assertAllowed(self.bash(f'(cd "{out}" && ls); rm -rf build'))
        # 内側の削除は検出する
        self.assertDenied(self.bash(f'x=$(cd "{out}" && rm -rf other)'))

    def test_powershell_location_variants(self):
        out = self.out
        for verb in ("Set-Location", "cd", "sl", "chdir"):
            with self.subTest(verb):
                self.assertDenied(self.ps(f"{verb} {out}; Remove-Item other -Recurse"))
        self.assertDenied(self.ps(f"Set-Location -Path {out}; Remove-Item existing.txt"))
        self.assertDenied(self.ps(f"Push-Location {out}; Remove-Item existing.txt"))
        self.assertAllowed(self.ps(f"Push-Location {out}; Pop-Location; Remove-Item build -Recurse"))
        self.assertAllowed(self.ps(f"Set-Location {out}; Get-Content existing.txt"))
        self.assertDenied(self.ps(f"Set-Location {out}; Set-Content existing.txt 'x'"))

    def test_home_expansion(self):
        with self.fake_home():
            for cmd in ("rm -rf ~/source/other", "rm -rf $HOME/source/other", "rm -rf ${HOME}/source/other",
                        "echo x > ~/.bashrc", "cd ~ && rm -rf source", "cd && rm .bashrc", "cd $HOME/source && rm -rf other"):
                with self.subTest(cmd):
                    self.assertDenied(self.bash(cmd), cmd)
            for cmd in ("echo x >> ~/.bashrc", "mkdir -p ~/newdir", "echo x > ~/new_file.txt", "cat ~/.bashrc"):
                with self.subTest(cmd):
                    self.assertAllowed(self.bash(cmd), cmd)
            for cmd in ("Remove-Item ~/source/other -Recurse", "Remove-Item $HOME\\source\\other -Recurse",
                        "Remove-Item $env:USERPROFILE\\source\\other -Recurse", "Set-Content ~/.bashrc 'x'"):
                with self.subTest(cmd):
                    self.assertDenied(self.ps(cmd), cmd)
            self.assertAllowed(self.ps("Add-Content ~/.bashrc 'x'"))

    def test_nested_shells(self):
        out = _fwd(self.out)
        self.assertDenied(self.bash(f"bash -c 'rm -rf \"{out}/other\"'"))
        self.assertDenied(self.bash(f"sh -c 'cd \"{out}\" && rm -rf other'"))
        self.assertDenied(self.bash(f"pwsh -Command \"Remove-Item '{self.out}' -Recurse\""))
        self.assertDenied(self.bash(f"powershell -Command \"Remove-Item '{self.out}' -Recurse\""))
        self.assertDenied(self.ps(f"bash -c 'rm -rf \"{out}/other\"'"))
        self.assertDenied(self.ps(f"pwsh -Command \"Remove-Item {self.out} -Recurse\""))
        self.assertAllowed(self.bash("bash -c 'rm -rf build'"))

    @unittest.skipUnless(os.name == "nt", "Git Bash 形式（/c/...）は Windows のみ")
    def test_gitbash_drive_form(self):
        out = _gitbash(self.out)
        self.assertDenied(self.bash(f'rm -rf "{out}/other"'))
        self.assertDenied(self.bash(f'cd "{out}" && rm -rf other'))
        self.assertDenied(self.bash(f'echo x > "{out}/existing.txt"'))
        self.assertAllowed(self.bash(f'echo x > "{out}/brand_new.txt"'))
        self.assertAllowed(self.bash(f'rm -rf "{_gitbash(self.repo)}/build"'))
        self.assertDenied(self.bash(f'rm -rf "/cygdrive{out}/other"'))


class TestRestrictReasons(RestrictSandboxMixin, unittest.TestCase):
    def assertWriteGuidance(self, reason):
        self.assertIsNotNone(reason)
        self.assertIn("迂回", reason)
        self.assertIn("! <コマンド>", reason)

    def test_delete_and_overwrite_reasons_have_guidance(self):
        self.assertWriteGuidance(self.bash(f'rm -rf "{_fwd(self.out)}"'))
        self.assertWriteGuidance(self.bash(f'echo x > "{_fwd(self.existing)}"'))
        self.assertWriteGuidance(self.ps(f"Remove-Item {self.out} -Recurse"))
        self.assertWriteGuidance(self.bash(f"python -c \"import shutil; shutil.rmtree('{_fwd(self.out)}')\""))
        self.assertWriteGuidance(self.tool("Write", {"file_path": self.existing}))
        self.assertWriteGuidance(self.tool("Edit", {"file_path": self.existing}))

    def test_deny_read_reason_has_no_bang_request(self):
        self.deny = [os.path.join(self.out, "dir")]
        reasons = [
            self.tool("Read", {"file_path": os.path.join(self.out, "dir", "file.txt")}),
            self.bash(f'cat "{_fwd(os.path.join(self.out, "dir", "file.txt"))}"'),
            self.tool("Grep", {"pattern": "x", "path": self.out}),
        ]
        for reason in reasons:
            with self.subTest(reason=reason):
                self.assertIsNotNone(reason)
                self.assertIn("迂回", reason)
                self.assertNotIn("! <コマンド>", reason)
                self.assertNotIn("`!", reason)


if __name__ == "__main__":
    unittest.main()