feat(tooling): add Cursor governance hooks, skills, and rule pointers
Layer advisory Cursor hooks and skills on the PI executor, shrink the test-report rule to the validation contract, and fix MCP relative paths.
This commit is contained in:
weichen
@@ -0,0 +1,352 @@
|
||||
"""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")
|
||||
|
||||
|
||||
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_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()
|
||||
Reference in New Issue
Block a user