weichen
5618cfc8cf
Layer advisory Cursor hooks and skills on the PI executor, shrink the test-report rule to the validation contract, and fix MCP relative paths.
353 lines
12 KiB
Python
353 lines
12 KiB
Python
"""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()
|