Agent-System/scripts/tests/test_cursor_hooks.py

"""Unit tests for project Cursor hooks (advisory gates) and hooks.json structure."""
from __future__ import annotations

import json
import os
import shutil
import stat
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

SOURCE = Path(__file__).resolve().parents[1]
REPO = SOURCE.parent
HOOKS = REPO / ".cursor" / "hooks"
HOOKS_JSON = REPO / ".cursor" / "hooks.json"
RULES = REPO / ".cursor" / "rules"


def _run_hook(script: Path, payload: dict | str, *, env: dict | None = None, cwd: Path | None = None) -> tuple[int, dict | None, str]:
    data = payload if isinstance(payload, str) else json.dumps(payload)
    e = os.environ.copy()
    e["CURSOR_HOOKS_NO_INVENTORY"] = "1"
    if env:
        e.update(env)
    # Isolate audit/probe under temp repo root via CURSOR_PROJECT_DIR
    r = subprocess.run(
        [str(script)],
        input=data,
        text=True,
        capture_output=True,
        cwd=str(cwd or REPO),
        env=e,
        check=False,
    )
    out = (r.stdout or "").strip()
    parsed = None
    if out:
        try:
            parsed = json.loads(out)
        except json.JSONDecodeError:
            parsed = None
    return r.returncode, parsed, r.stderr or ""


class CursorHooksStructureTest(unittest.TestCase):
    def test_hooks_json_parseable_and_scripts_executable(self):
        cfg = json.loads(HOOKS_JSON.read_text(encoding="utf-8"))
        self.assertEqual(cfg.get("version"), 1)
        hooks = cfg.get("hooks") or {}
        self.assertTrue(hooks)
        for event, entries in hooks.items():
            self.assertIsInstance(entries, list, event)
            for entry in entries:
                cmd = entry.get("command")
                self.assertTrue(cmd, f"{event} missing command")
                path = REPO / cmd
                self.assertTrue(path.is_file(), f"missing {path}")
                mode = path.stat().st_mode
                self.assertTrue(mode & stat.S_IXUSR, f"not executable: {path}")
                matcher = entry.get("matcher")
                if matcher:
                    # JavaScript RegExp must compile in Node
                    r = subprocess.run(
                        ["node", "-e", f"new RegExp({json.dumps(matcher)})"],
                        capture_output=True,
                        text=True,
                        check=False,
                    )
                    self.assertEqual(r.returncode, 0, f"bad matcher {matcher!r}: {r.stderr}")

    def test_rules_no_drift_markers(self):
        text = (RULES / "test-reports-standard.mdc").read_text(encoding="utf-8")
        self.assertNotIn("MiniMax-M2.7-highspeed", text)
        self.assertNotIn("PROJECT.md) §8", text)
        self.assertNotIn("PROJECT.md](../../PROJECT.md) §8", text)
        # No unconditional full go test requirement as acceptance
        self.assertNotIn("go test -v ./...", text)
        self.assertIn("验证契约", text)

    def test_mcp_relative_paths_for_agentctl_jev(self):
        mcp = json.loads((REPO / ".cursor" / "mcp.json").read_text(encoding="utf-8"))
        for name in ("agentctl", "jev", "pi-subagents"):
            args = mcp["mcpServers"][name]["args"]
            self.assertTrue(any(a.startswith("scripts/") for a in args), name)
            self.assertFalse(any("${workspaceFolder}" in a for a in args), name)


class GuardGitTest(unittest.TestCase):
    SCRIPT = HOOKS / "guard-git.sh"

    def _perm(self, command: str) -> str:
        code, parsed, err = _run_hook(self.SCRIPT, {"command": command})
        self.assertEqual(code, 0, err)
        self.assertIsInstance(parsed, dict)
        return parsed.get("permission")

    def test_allow_status(self):
        self.assertEqual(self._perm("git status"), "allow")

    def test_ask_push(self):
        self.assertEqual(self._perm("git push origin dev"), "ask")

    def test_ask_force(self):
        self.assertEqual(self._perm("git push --force origin main"), "ask")

    def test_ask_reset_hard(self):
        self.assertEqual(self._perm("git reset --hard HEAD~1"), "ask")

    def test_ask_clean_f(self):
        self.assertEqual(self._perm("git clean -fd"), "ask")

    def test_ask_branch_D(self):
        self.assertEqual(self._perm("git branch -D feature/x"), "ask")


class GuardPiDispatchTest(unittest.TestCase):
    SCRIPT = HOOKS / "guard-pi-dispatch.sh"

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        (self.root / ".cursor").mkdir()
        self.env = {"CURSOR_PROJECT_DIR": str(self.root), "CURSOR_HOOKS_NO_INVENTORY": "1"}

    def tearDown(self):
        self.tmp.cleanup()

    def _run(self, payload: dict) -> dict:
        code, parsed, err = _run_hook(self.SCRIPT, payload, env=self.env, cwd=self.root)
        self.assertEqual(code, 0, err)
        self.assertIsInstance(parsed, dict)
        return parsed

    def _valid_run(self, **overrides) -> dict:
        req = {
            "role_id": "research",
            "request_id": "research-001",
            "goal": "read PROJECT.md",
            "authorization_ref": "task auth",
            "source_sha": "cfc80b34a40e8de78e77a44839571bb684d405d5",
            "input_files": ["PROJECT.md"],
            "timeout_seconds": 120,
        }
        req.update(overrides)
        return {
            "mcp_server_name": "pi-subagents",
            "tool_name": "pi_subagent_run",
            "tool_input": {"worktree": "/tmp/wt", "task_id": "t1", "request": req},
        }

    def test_allow_valid_run(self):
        self.assertEqual(self._run(self._valid_run()).get("permission"), "allow")

    def test_allow_roles(self):
        p = {
            "mcp_server_name": "pi-subagents",
            "tool_name": "pi_subagent_roles",
            "tool_input": {"worktree": "/tmp/wt", "task_id": "t1"},
        }
        self.assertEqual(self._run(p).get("permission"), "allow")

    def test_allow_result(self):
        p = {
            "tool_name": "pi_subagent_result",
            "arguments": {"worktree": "/tmp/wt", "task_id": "t1", "request_id": "r1"},
        }
        self.assertEqual(self._run(p).get("permission"), "allow")

    def test_ask_source_sha_head(self):
        self.assertEqual(self._run(self._valid_run(source_sha="HEAD")).get("permission"), "ask")

    def test_ask_source_sha_short(self):
        self.assertEqual(self._run(self._valid_run(source_sha="cfc80b34")).get("permission"), "ask")

    def test_ask_test_missing_commands(self):
        p = self._valid_run(role_id="test_run")
        self.assertEqual(self._run(p).get("permission"), "ask")

    def test_ask_test_bad_digest(self):
        p = self._valid_run(
            role_id="acceptance_check",
            approved_commands=[["true"]],
            plan_digest="not-a-digest",
        )
        self.assertEqual(self._run(p).get("permission"), "ask")

    def test_allow_test_with_commands(self):
        digest = "a" * 64
        p = self._valid_run(
            role_id="test_run",
            approved_commands=[["python3", "-c", "print(1)"]],
            plan_digest=digest,
        )
        self.assertEqual(self._run(p).get("permission"), "allow")

    def test_ask_model_override(self):
        p = self._valid_run(model_override={"provider": "x", "model": "y"})
        self.assertEqual(self._run(p).get("permission"), "ask")

    def test_fail_open_empty_json(self):
        code, parsed, err = _run_hook(self.SCRIPT, "", env=self.env, cwd=self.root)
        self.assertEqual(code, 0, err)
        self.assertEqual(parsed.get("permission"), "allow")
        audit = (self.root / ".cursor" / "hook-audit.jsonl").read_text(encoding="utf-8")
        self.assertIn("UNRECOGNIZED_INPUT", audit)

    def test_alt_field_names_arguments(self):
        # tool_input via "arguments"
        payload = {
            "name": "pi_subagent_run",
            "arguments": {
                "request": {
                    "role_id": "planning",
                    "request_id": "p-1",
                    "goal": "g",
                    "authorization_ref": "a",
                    "source_sha": "cfc80b34a40e8de78e77a44839571bb684d405d5",
                    "input_files": [],
                    "timeout_seconds": 30,
                }
            },
        }
        self.assertEqual(self._run(payload).get("permission"), "allow")

    def test_ask_mcp_prefixed_tool_name_bad_sha(self):
        # IDE postToolUse/beforeMCP observed tool_name="MCP:pi_subagent_*"
        payload = {
            "mcp_server_name": "pi-subagents",
            "tool_name": "MCP:pi_subagent_run",
            "tool_input": {
                "worktree": "/tmp/wt",
                "task_id": "t1",
                "request": {
                    "role_id": "research",
                    "request_id": "research-001",
                    "goal": "g",
                    "authorization_ref": "a",
                    "source_sha": "HEAD",
                    "input_files": [],
                    "timeout_seconds": 30,
                },
            },
        }
        self.assertEqual(self._run(payload).get("permission"), "ask")

    def test_allow_mcp_prefixed_roles(self):
        payload = {
            "tool_name": "MCP:pi_subagent_roles",
            "tool_input": {"worktree": "/tmp/wt", "task_id": "t1"},
        }
        self.assertEqual(self._run(payload).get("permission"), "allow")


class GuardPiShellTest(unittest.TestCase):
    SCRIPT = HOOKS / "guard-pi-shell.sh"

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        (self.root / ".cursor").mkdir()
        self.env = {"CURSOR_PROJECT_DIR": str(self.root)}

    def tearDown(self):
        self.tmp.cleanup()

    def _perm(self, command: str) -> str:
        code, parsed, err = _run_hook(
            self.SCRIPT, {"command": command}, env=self.env, cwd=self.root
        )
        self.assertEqual(code, 0, err)
        return parsed.get("permission")

    def test_allow_roles(self):
        self.assertEqual(self._perm("python3 -B scripts/pi_subagents.py roles"), "allow")

    def test_ask_bare_run(self):
        self.assertEqual(
            self._perm("python3 -B scripts/pi_subagents.py run --request /tmp/r.json"),
            "ask",
        )

    def test_allow_leased_run(self):
        cmd = (
            "python3 scripts/agentctl.py with --task t1 "
            "--resource path:.doc/test-reports/assets/pi-subagents/t1/r1 -- "
            "python3 -B scripts/pi_subagents.py run --request /tmp/r.json"
        )
        self.assertEqual(self._perm(cmd), "allow")

    def test_ask_pkill_f(self):
        self.assertEqual(self._perm("pkill -f pi_subagents"), "ask")


class PiReceiptContextTest(unittest.TestCase):
    SCRIPT = HOOKS / "pi-receipt-context.sh"

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        (self.root / ".cursor").mkdir()
        self.env = {"CURSOR_PROJECT_DIR": str(self.root)}

    def tearDown(self):
        self.tmp.cleanup()

    def test_injects_for_pi_tool(self):
        code, parsed, err = _run_hook(
            self.SCRIPT,
            {"tool_name": "pi_subagent_result", "mcp_server_name": "pi-subagents"},
            env=self.env,
            cwd=self.root,
        )
        self.assertEqual(code, 0, err)
        self.assertIn("additional_context", parsed)
        self.assertIn("run_status", parsed["additional_context"])

    def test_injects_for_mcp_prefixed_tool_name(self):
        code, parsed, err = _run_hook(
            self.SCRIPT,
            {"tool_name": "MCP:pi_subagent_roles", "mcp_server_name": "pi-subagents"},
            env=self.env,
            cwd=self.root,
        )
        self.assertEqual(code, 0, err)
        self.assertIn("additional_context", parsed)

    def test_noop_for_unrelated(self):
        code, parsed, err = _run_hook(
            self.SCRIPT,
            {"tool_name": "Read", "tool_input": {"path": "x"}},
            env=self.env,
            cwd=self.root,
        )
        self.assertEqual(code, 0, err)
        self.assertEqual(parsed, {})


class AuditSessionTest(unittest.TestCase):
    SCRIPT = HOOKS / "audit-session.sh"

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        (self.root / ".cursor").mkdir()
        # Minimal fake repo so inventory finds no agentctl and skips cleanly
        self.env = {
            "CURSOR_PROJECT_DIR": str(self.root),
            "CURSOR_HOOKS_NO_INVENTORY": "1",
        }

    def tearDown(self):
        self.tmp.cleanup()

    def test_emits_empty_object(self):
        code, parsed, err = _run_hook(
            self.SCRIPT,
            {"hook_event_name": "sessionStart"},
            env=self.env,
            cwd=self.root,
        )
        self.assertEqual(code, 0, err)
        self.assertEqual(parsed, {})
        audit = (self.root / ".cursor" / "hook-audit.jsonl").read_text(encoding="utf-8")
        self.assertIn("SESSION_AUDIT", audit)

    def test_probe_writes_when_enabled(self):
        env = dict(self.env)
        env["CURSOR_HOOKS_PROBE"] = "1"
        code, _, err = _run_hook(
            self.SCRIPT, {"event": "stop"}, env=env, cwd=self.root
        )
        self.assertEqual(code, 0, err)
        probe = self.root / ".cursor" / "hook-probe.jsonl"
        self.assertTrue(probe.is_file())
        self.assertIn("audit-session", probe.read_text(encoding="utf-8"))


class SkillsPresentTest(unittest.TestCase):
    def test_skills_exist(self):
        for name in ("pi-subagent-dispatch", "subagent-stage-record"):
            p = REPO / ".cursor" / "skills" / name / "SKILL.md"
            self.assertTrue(p.is_file(), p)
            text = p.read_text(encoding="utf-8")
            self.assertIn("name:", text)


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