#677 + #635 hooks: -p headless short-circuit + Codex bridge + portable hook test runner. #677: session_boot.py detects -p in extra_args and execvp's claude directly — headless one-shots never register in ~/.claude/sessions, so tmux/session-lookup/live-attach logic was wrong for them (DPLAN-0226 fine-tune debt; +5 tests, 39 pass). #635: new bridges/codex.py mirrors claude.py with Codex protocol normalization (stdin input->tool_input, stdout hookSpecificOutput envelope, permissionDecision+permissionDecisionReason — fixes DPLAN-0205 bugs) + new hook_test module: drone @hooks test fires every hook from .aipass/hooks.json with per-event mock data, reports fired/blocked/disabled/crashed with timing (23 tests). codex.py bypass entries exactly mirror shipped claude.py set. Built by @hooks; devpulse verified: 62/62 tests re-run against real repo, seedgo 31/31 all three files.
This commit is contained in:
@@ -41,6 +41,36 @@
|
||||
"standard": "imports",
|
||||
"reason": "Bridge imports engine module by design \u2014 sole purpose."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "dead_code",
|
||||
"reason": "Bridge called externally by Codex hook settings subprocess \u2014 no internal import. Wired in .codex/hooks.json."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "unused_function",
|
||||
"reason": "main() called as subprocess entry point from Codex hook settings \u2014 never statically imported."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "handlers",
|
||||
"reason": "Bridges import engine module by design \u2014 that is their entire purpose."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "json_structure",
|
||||
"reason": "Thin entry point using stdlib json for Codex protocol envelope \u2014 no JSON file ops needing json_handler."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "architecture",
|
||||
"reason": "Bridge importing engine module is its architectural purpose."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/bridges/codex.py",
|
||||
"standard": "imports",
|
||||
"reason": "Bridge imports engine module by design \u2014 sole purpose."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/prompt/identity.py",
|
||||
"standard": "dead_code",
|
||||
@@ -396,6 +426,11 @@
|
||||
"standard": "modules",
|
||||
"reason": "dispatch() is the engine's core purpose \u2014 it IS the module's primary function, not a handler that belongs elsewhere. The engine exists to dispatch; moving dispatch to handlers/ would leave an empty module."
|
||||
},
|
||||
{
|
||||
"file": "apps/modules/hook_test.py",
|
||||
"standard": "json_structure",
|
||||
"reason": "Uses stdlib json.dumps to serialize mock event data for engine dispatch — no JSON file ops needing json_handler."
|
||||
},
|
||||
{
|
||||
"file": "apps/modules/hooksound.py",
|
||||
"standard": "json_structure",
|
||||
|
||||
@@ -50,6 +50,7 @@ src/aipass/hooks/
|
||||
│ ├── sound.py # Shared sound utilities (speak, play, mute)
|
||||
│ ├── modules/
|
||||
│ │ ├── cadence.py # Prompt injection cadence (every-Nth-turn gating)
|
||||
│ │ ├── hook_test.py # Portable test runner (drone @hooks test)
|
||||
│ │ ├── cc_sessions.py # CC-native session file reader (~/.claude/sessions/<pid>.json)
|
||||
│ │ ├── engine.py # Core dispatch — routes events to handlers
|
||||
│ │ ├── hooksound.py # Sound control (drone @hooks hooksound on/off)
|
||||
@@ -59,7 +60,8 @@ src/aipass/hooks/
|
||||
│ │ └── wire_verify.py # Wire verification — provider ↔ project hook wiring checker
|
||||
│ ├── handlers/
|
||||
│ │ ├── bridges/ # One per provider (thin normalization)
|
||||
│ │ │ └── claude.py # Claude Code bridge
|
||||
│ │ │ ├── claude.py # Claude Code bridge
|
||||
│ │ │ └── codex.py # Codex bridge (normalizes stdin/stdout envelope)
|
||||
│ │ ├── prompt/ # Prompt injection hooks
|
||||
│ │ │ ├── branch_loader.py # Injects aipass_local_prompt.md
|
||||
│ │ │ ├── tier0_kernel.py # Injects tier0 kernel prompt (every turn)
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: codex.py
|
||||
# Version: 1.0.0
|
||||
# Description: Codex bridge — entry point for provider hook settings
|
||||
# Branch: hooks
|
||||
# Layer: apps/handlers/bridges
|
||||
# Created: 2026-07-10
|
||||
# Modified: 2026-07-10
|
||||
# =============================================
|
||||
|
||||
"""Codex bridge.
|
||||
|
||||
Thin entry point called from .codex/hooks.json hook entries.
|
||||
Normalizes Codex's stdin/stdout format and calls the engine.
|
||||
|
||||
Codex protocol differences from Claude Code:
|
||||
- stdin: uses 'input' instead of 'tool_input' for tool parameters
|
||||
- stdout: wraps output in hookSpecificOutput envelope
|
||||
- blocking: permissionDecision + permissionDecisionReason (not exit code 2)
|
||||
|
||||
Supports two forms:
|
||||
codex.py EventType — dispatch ALL enabled hooks for that event
|
||||
codex.py EventType:hook_name — dispatch ONLY that one hook (separate output)
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
|
||||
from aipass.hooks.apps.modules.engine import dispatch
|
||||
from aipass.hooks.apps.handlers.config.loader import find_project_config
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
|
||||
|
||||
def _normalize_stdin(stdin_data: str) -> str:
|
||||
"""Remap Codex field names to engine-expected names."""
|
||||
if not stdin_data.strip():
|
||||
return stdin_data
|
||||
try:
|
||||
parsed = json.loads(stdin_data)
|
||||
if "input" in parsed and "tool_input" not in parsed:
|
||||
parsed["tool_input"] = parsed.pop("input")
|
||||
return json.dumps(parsed)
|
||||
except (json.JSONDecodeError, TypeError) as exc:
|
||||
logger.info("[HOOKS:codex] stdin normalization failed: %s", exc)
|
||||
return stdin_data
|
||||
|
||||
|
||||
def _wrap_output(event_type: str, output: str, exit_code: int) -> str:
|
||||
"""Wrap engine output into Codex hookSpecificOutput envelope."""
|
||||
if exit_code == 2:
|
||||
try:
|
||||
decision = json.loads(output)
|
||||
if decision.get("decision") == "block":
|
||||
reason = decision.get("reason", "Blocked by AIPass hook")
|
||||
return json.dumps(
|
||||
{
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": event_type,
|
||||
"permissionDecision": "deny",
|
||||
"permissionDecisionReason": reason,
|
||||
},
|
||||
"systemMessage": reason,
|
||||
}
|
||||
)
|
||||
except (json.JSONDecodeError, TypeError, AttributeError) as exc:
|
||||
logger.info("[HOOKS:codex] block output parse failed: %s", exc)
|
||||
|
||||
if not output:
|
||||
return json.dumps({})
|
||||
|
||||
return json.dumps(
|
||||
{
|
||||
"hookSpecificOutput": {
|
||||
"hookEventName": event_type,
|
||||
"additionalContext": output,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Entry point — receive event type from Codex, dispatch via engine."""
|
||||
if len(sys.argv) < 2:
|
||||
sys.stderr.write("Usage: codex.py <EventType> or codex.py <EventType:hook_name>\n")
|
||||
sys.exit(1)
|
||||
|
||||
arg = sys.argv[1]
|
||||
hook_filter = None
|
||||
if ":" in arg:
|
||||
event_type, hook_filter = arg.split(":", 1)
|
||||
else:
|
||||
event_type = arg
|
||||
|
||||
stdin_data = ""
|
||||
if not sys.stdin.isatty():
|
||||
stdin_data = sys.stdin.read()
|
||||
|
||||
normalized = _normalize_stdin(stdin_data)
|
||||
|
||||
config = find_project_config()
|
||||
if config is None:
|
||||
config = {"hooks_enabled": True}
|
||||
logger.info("[HOOKS:codex] no project config found, using defaults")
|
||||
|
||||
if hook_filter:
|
||||
full_config: dict = config
|
||||
hook_def = full_config.get(event_type, {}).get(hook_filter, {})
|
||||
config = {"hooks_enabled": True, event_type: {hook_filter: hook_def}}
|
||||
|
||||
output, exit_code = dispatch(event_type, normalized, config)
|
||||
|
||||
wrapped = _wrap_output(event_type, output, exit_code)
|
||||
sys.stdout.write(wrapped)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -144,6 +144,13 @@ def boot(cwd: str | None = None, extra_args: list[str] | None = None) -> dict:
|
||||
|
||||
defaults = _DEFAULT_ARGS if not (extra_args and "--permission-mode" in extra_args) else []
|
||||
|
||||
if extra_args and "-p" in extra_args:
|
||||
logger.info("[SESSION_BOOT] Headless mode (-p) — running claude directly, no tmux")
|
||||
claude_cmd = [claude_bin] + defaults
|
||||
claude_cmd.extend(extra_args)
|
||||
os.execvp(claude_bin, claude_cmd)
|
||||
return {"exit_code": 0, "action": "direct", "reason": "headless -p mode"}
|
||||
|
||||
if os.environ.get("TMUX"):
|
||||
logger.info("[SESSION_BOOT] Already inside tmux — running claude directly")
|
||||
claude_cmd = [claude_bin] + defaults
|
||||
|
||||
@@ -0,0 +1,218 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: hook_test.py
|
||||
# Version: 1.0.0
|
||||
# Description: Portable hook test runner — fires every hook with mock data
|
||||
# Branch: hooks
|
||||
# Layer: apps/modules
|
||||
# Created: 2026-07-10
|
||||
# Modified: 2026-07-10
|
||||
# =============================================
|
||||
|
||||
"""Portable hook test runner.
|
||||
|
||||
Fires every hook from a project's .aipass/hooks.json with mock data
|
||||
and reports what fired, what blocked, and what crashed. Runnable from
|
||||
any project directory.
|
||||
|
||||
Usage:
|
||||
drone @hooks test [--verbose]
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
from aipass.hooks.apps.modules.engine import dispatch
|
||||
from aipass.hooks.apps.handlers.config.loader import find_project_config
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
from aipass.cli.apps.modules import err_console
|
||||
|
||||
CONSOLE = err_console
|
||||
|
||||
HELP_COMMANDS = [
|
||||
("test [--verbose]", "Fire every hook with mock data and report results"),
|
||||
]
|
||||
|
||||
_SYNTHETIC_PATH = os.path.join(tempfile.gettempdir(), "hook_test_synthetic.txt")
|
||||
|
||||
MOCK_EVENTS = {
|
||||
"UserPromptSubmit": {
|
||||
"type": "UserPromptSubmit",
|
||||
"prompt": "[hook test] synthetic prompt for test runner",
|
||||
},
|
||||
"PreToolUse": {
|
||||
"tool_name": "Read",
|
||||
"tool_input": {"file_path": _SYNTHETIC_PATH},
|
||||
},
|
||||
"PostToolUse": {
|
||||
"tool_name": "Read",
|
||||
"tool_input": {"file_path": _SYNTHETIC_PATH},
|
||||
"tool_output": "synthetic output",
|
||||
},
|
||||
"SubagentStop": {
|
||||
"agent_type": "general-purpose",
|
||||
"type": "SubagentStop",
|
||||
},
|
||||
"Stop": {
|
||||
"type": "Stop",
|
||||
},
|
||||
"Notification": {
|
||||
"type": "Notification",
|
||||
"message": "[hook test] synthetic notification",
|
||||
},
|
||||
"PreCompact": {
|
||||
"compact_type": "PreCompact",
|
||||
},
|
||||
"SessionStart": {
|
||||
"type": "SessionStart",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _test_single_hook(event_type: str, hook_name: str, hook_def: dict, stdin_data: str, verbose: bool) -> dict:
|
||||
"""Dispatch one hook and return its result dict."""
|
||||
single_config = {"hooks_enabled": True, event_type: {hook_name: hook_def}}
|
||||
enabled = hook_def.get("enabled", True)
|
||||
|
||||
start = time.monotonic()
|
||||
try:
|
||||
output, exit_code = dispatch(event_type, stdin_data, single_config)
|
||||
elapsed_ms = round((time.monotonic() - start) * 1000, 1)
|
||||
|
||||
if not enabled:
|
||||
status = "disabled"
|
||||
elif exit_code == 2:
|
||||
status = "blocked"
|
||||
elif output:
|
||||
status = "fired"
|
||||
else:
|
||||
status = "fired (empty output)"
|
||||
|
||||
return {
|
||||
"hook": hook_name,
|
||||
"status": status,
|
||||
"elapsed_ms": elapsed_ms,
|
||||
"exit_code": exit_code,
|
||||
"output_len": len(output),
|
||||
"output_preview": output[:200] if verbose else "",
|
||||
}
|
||||
except Exception as exc:
|
||||
elapsed_ms = round((time.monotonic() - start) * 1000, 1)
|
||||
logger.error("[HOOKS:test] %s.%s crashed: %s", event_type, hook_name, exc)
|
||||
return {
|
||||
"hook": hook_name,
|
||||
"status": "crashed",
|
||||
"elapsed_ms": elapsed_ms,
|
||||
"error": str(exc)[:200],
|
||||
}
|
||||
|
||||
|
||||
def run_test(verbose: bool = False) -> dict:
|
||||
"""Fire every hook with mock data, return results summary."""
|
||||
config = find_project_config()
|
||||
if config is None:
|
||||
return {"error": "No .aipass/hooks.json found — run from an AIPass project directory."}
|
||||
|
||||
if not config.get("hooks_enabled", True):
|
||||
return {"error": "hooks_enabled is false in project config."}
|
||||
|
||||
results = {}
|
||||
|
||||
for event_type, event_hooks in config.items():
|
||||
if event_type in ("hooks_enabled", "_comment"):
|
||||
continue
|
||||
if not isinstance(event_hooks, dict):
|
||||
continue
|
||||
|
||||
mock_data = MOCK_EVENTS.get(event_type, {"type": event_type})
|
||||
stdin_data = json.dumps(mock_data)
|
||||
|
||||
event_results = []
|
||||
for hook_name, hook_def in event_hooks.items():
|
||||
if not isinstance(hook_def, dict):
|
||||
continue
|
||||
if not hook_def.get("handler", "") and not hook_def.get("command", ""):
|
||||
continue
|
||||
result = _test_single_hook(event_type, hook_name, hook_def, stdin_data, verbose)
|
||||
event_results.append(result)
|
||||
|
||||
if event_results:
|
||||
results[event_type] = event_results
|
||||
|
||||
return results
|
||||
|
||||
|
||||
_STATUS_ICONS = {
|
||||
"fired": "[green]✓[/green]",
|
||||
"fired (empty output)": "[green]✓[/green]",
|
||||
"blocked": "[yellow]⊘[/yellow]",
|
||||
"disabled": "[dim]○[/dim]",
|
||||
"crashed": "[red]✗[/red]",
|
||||
}
|
||||
|
||||
_STATUS_COUNTS = {"fired", "fired (empty output)", "blocked", "disabled", "crashed"}
|
||||
|
||||
|
||||
def print_results(results: dict, verbose: bool = False) -> None:
|
||||
"""Render test results to console."""
|
||||
if "error" in results:
|
||||
CONSOLE.print(f"[red]{results['error']}[/red]")
|
||||
return
|
||||
|
||||
counts = {"fired": 0, "blocked": 0, "disabled": 0, "crashed": 0}
|
||||
|
||||
for event_type, hooks in results.items():
|
||||
CONSOLE.print(f"\n[bold cyan]{event_type}[/bold cyan]")
|
||||
for h in hooks:
|
||||
status = h["status"]
|
||||
name = h["hook"]
|
||||
ms = h.get("elapsed_ms", 0)
|
||||
icon = _STATUS_ICONS.get(status, "[dim]?[/dim]")
|
||||
|
||||
if status in ("fired", "fired (empty output)"):
|
||||
counts["fired"] += 1
|
||||
elif status in counts:
|
||||
counts[status] += 1
|
||||
|
||||
CONSOLE.print(f" {icon} {name:30} {status:20} {ms:>6.0f}ms")
|
||||
if verbose and h.get("output_preview"):
|
||||
CONSOLE.print(f" [dim]{h['output_preview']}[/dim]")
|
||||
if h.get("error"):
|
||||
CONSOLE.print(f" [red]{h['error']}[/red]")
|
||||
|
||||
CONSOLE.print()
|
||||
CONSOLE.print(
|
||||
f"[bold]Summary:[/bold] {counts['fired']} fired, "
|
||||
f"{counts['blocked']} blocked, {counts['disabled']} disabled, "
|
||||
f"{counts['crashed']} crashed"
|
||||
)
|
||||
|
||||
|
||||
def print_introspection() -> None:
|
||||
"""Print module introspection for drone discovery."""
|
||||
CONSOLE.print("[cyan]hook_test[/cyan] — Portable hook test runner")
|
||||
CONSOLE.print(" Fire every hook with mock data and report what fired.")
|
||||
|
||||
|
||||
def handle_command(command: str, args: list) -> bool:
|
||||
"""Route 'test' command."""
|
||||
if command != "test":
|
||||
return False
|
||||
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
|
||||
if args[0] in ("--help", "-h"):
|
||||
print_introspection()
|
||||
return True
|
||||
|
||||
verbose = "--verbose" in args or "-v" in args
|
||||
|
||||
CONSOLE.print("[bold cyan]HOOKS Test Runner[/bold cyan]")
|
||||
CONSOLE.print("[dim]Firing every hook with mock data...[/dim]")
|
||||
|
||||
results = run_test(verbose=verbose)
|
||||
print_results(results, verbose=verbose)
|
||||
return True
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Tests for the Codex bridge (handlers/bridges/codex.py)."""
|
||||
|
||||
import json
|
||||
|
||||
from aipass.hooks.apps.handlers.bridges.codex import _normalize_stdin, _wrap_output
|
||||
|
||||
|
||||
class TestNormalizeStdin:
|
||||
def test_remaps_input_to_tool_input(self):
|
||||
stdin = json.dumps({"tool_name": "Edit", "input": {"file_path": "/tmp/x.py"}})
|
||||
result = json.loads(_normalize_stdin(stdin))
|
||||
assert "tool_input" in result
|
||||
assert result["tool_input"] == {"file_path": "/tmp/x.py"}
|
||||
assert "input" not in result
|
||||
|
||||
def test_preserves_existing_tool_input(self):
|
||||
stdin = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/x.py"}})
|
||||
result = json.loads(_normalize_stdin(stdin))
|
||||
assert result["tool_input"] == {"file_path": "/tmp/x.py"}
|
||||
|
||||
def test_no_clobber_when_both_present(self):
|
||||
stdin = json.dumps({"tool_input": {"a": 1}, "input": {"b": 2}})
|
||||
result = json.loads(_normalize_stdin(stdin))
|
||||
assert result["tool_input"] == {"a": 1}
|
||||
assert "input" in result
|
||||
|
||||
def test_empty_string_passthrough(self):
|
||||
assert _normalize_stdin("") == ""
|
||||
assert _normalize_stdin(" ") == " "
|
||||
|
||||
def test_invalid_json_passthrough(self):
|
||||
assert _normalize_stdin("not json") == "not json"
|
||||
|
||||
def test_non_dict_json_passthrough(self):
|
||||
result = _normalize_stdin("[1, 2, 3]")
|
||||
assert json.loads(result) == [1, 2, 3]
|
||||
|
||||
|
||||
class TestWrapOutput:
|
||||
def test_block_wraps_as_deny(self):
|
||||
block_json = json.dumps({"decision": "block", "reason": "git write blocked"})
|
||||
result = json.loads(_wrap_output("PreToolUse", block_json, 2))
|
||||
hook_output = result["hookSpecificOutput"]
|
||||
assert hook_output["hookEventName"] == "PreToolUse"
|
||||
assert hook_output["permissionDecision"] == "deny"
|
||||
assert hook_output["permissionDecisionReason"] == "git write blocked"
|
||||
assert result["systemMessage"] == "git write blocked"
|
||||
|
||||
def test_block_with_no_reason_uses_default(self):
|
||||
block_json = json.dumps({"decision": "block"})
|
||||
result = json.loads(_wrap_output("PreToolUse", block_json, 2))
|
||||
assert result["hookSpecificOutput"]["permissionDecisionReason"] == "Blocked by AIPass hook"
|
||||
|
||||
def test_context_injection(self):
|
||||
result = json.loads(_wrap_output("UserPromptSubmit", "# Identity\nYou are hooks.", 0))
|
||||
hook_output = result["hookSpecificOutput"]
|
||||
assert hook_output["hookEventName"] == "UserPromptSubmit"
|
||||
assert hook_output["additionalContext"] == "# Identity\nYou are hooks."
|
||||
assert "permissionDecision" not in hook_output
|
||||
|
||||
def test_empty_output_returns_empty_object(self):
|
||||
result = json.loads(_wrap_output("PreToolUse", "", 0))
|
||||
assert result == {}
|
||||
|
||||
def test_exit_2_non_block_json_falls_through(self):
|
||||
result = json.loads(_wrap_output("PreToolUse", "crash output", 2))
|
||||
assert result["hookSpecificOutput"]["additionalContext"] == "crash output"
|
||||
|
||||
def test_exit_2_non_decision_json_falls_through(self):
|
||||
non_block = json.dumps({"something": "else"})
|
||||
result = json.loads(_wrap_output("PreToolUse", non_block, 2))
|
||||
assert "additionalContext" in result["hookSpecificOutput"]
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Tests for the portable hook test runner (modules/hook_test.py)."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from aipass.hooks.apps.modules import hook_test
|
||||
|
||||
_MOD = "aipass.hooks.apps.modules.hook_test"
|
||||
|
||||
|
||||
class TestRunTest:
|
||||
def test_no_config_returns_error(self):
|
||||
with patch(f"{_MOD}.find_project_config", return_value=None):
|
||||
result = hook_test.run_test()
|
||||
assert "error" in result
|
||||
assert "hooks.json" in result["error"]
|
||||
|
||||
def test_hooks_disabled_returns_error(self):
|
||||
with patch(f"{_MOD}.find_project_config", return_value={"hooks_enabled": False}):
|
||||
result = hook_test.run_test()
|
||||
assert "error" in result
|
||||
assert "hooks_enabled" in result["error"]
|
||||
|
||||
def test_fires_enabled_hooks(self):
|
||||
config = {
|
||||
"hooks_enabled": True,
|
||||
"PreToolUse": {
|
||||
"test_hook": {
|
||||
"enabled": True,
|
||||
"handler": "aipass.hooks.apps.handlers.security.git_gate.handle",
|
||||
"matcher": "Bash|Edit",
|
||||
},
|
||||
},
|
||||
}
|
||||
with patch(f"{_MOD}.find_project_config", return_value=config):
|
||||
result = hook_test.run_test()
|
||||
assert "PreToolUse" in result
|
||||
assert len(result["PreToolUse"]) == 1
|
||||
assert result["PreToolUse"][0]["hook"] == "test_hook"
|
||||
|
||||
def test_skips_non_dict_entries(self):
|
||||
config = {
|
||||
"hooks_enabled": True,
|
||||
"_comment": "template config",
|
||||
}
|
||||
with patch(f"{_MOD}.find_project_config", return_value=config):
|
||||
result = hook_test.run_test()
|
||||
assert result == {}
|
||||
|
||||
def test_skips_hooks_without_handler(self):
|
||||
config = {
|
||||
"hooks_enabled": True,
|
||||
"PreToolUse": {
|
||||
"empty_hook": {"enabled": True},
|
||||
},
|
||||
}
|
||||
with patch(f"{_MOD}.find_project_config", return_value=config):
|
||||
result = hook_test.run_test()
|
||||
assert result == {}
|
||||
|
||||
def test_reports_crashed_hooks(self):
|
||||
config = {
|
||||
"hooks_enabled": True,
|
||||
"PreToolUse": {
|
||||
"bad_hook": {
|
||||
"enabled": True,
|
||||
"handler": "nonexistent.module.handle",
|
||||
"matcher": "",
|
||||
},
|
||||
},
|
||||
}
|
||||
with patch(f"{_MOD}.find_project_config", return_value=config):
|
||||
result = hook_test.run_test()
|
||||
assert "PreToolUse" in result
|
||||
assert result["PreToolUse"][0]["status"] in ("crashed", "fired (empty output)")
|
||||
|
||||
|
||||
class TestPrintResults:
|
||||
def test_error_result_prints(self):
|
||||
with patch.object(hook_test.CONSOLE, "print") as mock_print:
|
||||
hook_test.print_results({"error": "No config found"})
|
||||
mock_print.assert_called_once()
|
||||
|
||||
def test_normal_results_print_summary(self):
|
||||
results = {
|
||||
"PreToolUse": [
|
||||
{"hook": "git_gate", "status": "fired", "elapsed_ms": 5.0},
|
||||
{"hook": "rm_gate", "status": "blocked", "elapsed_ms": 3.0},
|
||||
],
|
||||
}
|
||||
calls = []
|
||||
with patch.object(hook_test.CONSOLE, "print", side_effect=lambda x="": calls.append(x)):
|
||||
hook_test.print_results(results)
|
||||
summary = [c for c in calls if "Summary" in str(c)]
|
||||
assert len(summary) == 1
|
||||
assert "1 fired" in summary[0]
|
||||
assert "1 blocked" in summary[0]
|
||||
|
||||
|
||||
class TestHandleCommand:
|
||||
def test_rejects_non_test_command(self):
|
||||
assert hook_test.handle_command("status", []) is False
|
||||
|
||||
def test_no_args_shows_introspection(self):
|
||||
with patch.object(hook_test, "print_introspection") as mock_intro:
|
||||
result = hook_test.handle_command("test", [])
|
||||
assert result is True
|
||||
mock_intro.assert_called_once()
|
||||
|
||||
def test_runs_test_with_run_arg(self):
|
||||
with (
|
||||
patch.object(hook_test, "run_test", return_value={}) as mock_run,
|
||||
patch.object(hook_test, "print_results"),
|
||||
patch.object(hook_test.CONSOLE, "print"),
|
||||
):
|
||||
result = hook_test.handle_command("test", ["run"])
|
||||
assert result is True
|
||||
mock_run.assert_called_once()
|
||||
@@ -245,6 +245,65 @@ class TestMain:
|
||||
mock_boot.assert_called_once_with(extra_args=None)
|
||||
|
||||
|
||||
class TestHeadlessBypass:
|
||||
def test_p_flag_skips_tmux_and_runs_directly(self, tmp_path):
|
||||
with (
|
||||
patch.dict("os.environ", {}, clear=True),
|
||||
patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
|
||||
patch(f"{_MOD}.os.execvp") as mock_exec,
|
||||
):
|
||||
session_boot.boot(cwd=str(tmp_path), extra_args=["-p", "do something"])
|
||||
mock_exec.assert_called_once()
|
||||
args = mock_exec.call_args[0][1]
|
||||
assert args[0] == "/usr/local/bin/claude"
|
||||
assert "-p" in args
|
||||
assert "do something" in args
|
||||
|
||||
def test_p_flag_does_not_look_for_live_sessions(self, tmp_path):
|
||||
with (
|
||||
patch.dict("os.environ", {}, clear=True),
|
||||
patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
|
||||
patch.object(session_boot, "_find_live_sessions") as mock_live,
|
||||
patch(f"{_MOD}.os.execvp"),
|
||||
):
|
||||
session_boot.boot(cwd=str(tmp_path), extra_args=["-p", "query"])
|
||||
mock_live.assert_not_called()
|
||||
|
||||
def test_p_flag_does_not_require_tmux(self, tmp_path):
|
||||
with (
|
||||
patch.dict("os.environ", {}, clear=True),
|
||||
patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
|
||||
patch.object(session_boot, "_find_tmux", return_value=None),
|
||||
patch(f"{_MOD}.os.execvp") as mock_exec,
|
||||
):
|
||||
result = session_boot.boot(cwd=str(tmp_path), extra_args=["-p", "query"])
|
||||
assert result["action"] == "direct"
|
||||
assert result["reason"] == "headless -p mode"
|
||||
mock_exec.assert_called_once()
|
||||
|
||||
def test_p_flag_still_gets_permission_mode_default(self, tmp_path):
|
||||
with (
|
||||
patch.dict("os.environ", {}, clear=True),
|
||||
patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
|
||||
patch(f"{_MOD}.os.execvp") as mock_exec,
|
||||
):
|
||||
session_boot.boot(cwd=str(tmp_path), extra_args=["-p", "query"])
|
||||
cmd = mock_exec.call_args[0][1]
|
||||
assert "--permission-mode" in cmd
|
||||
assert "bypassPermissions" in cmd
|
||||
|
||||
def test_p_flag_respects_custom_permission_mode(self, tmp_path):
|
||||
with (
|
||||
patch.dict("os.environ", {}, clear=True),
|
||||
patch.object(session_boot, "_resolve_claude_binary", return_value="/usr/local/bin/claude"),
|
||||
patch(f"{_MOD}.os.execvp") as mock_exec,
|
||||
):
|
||||
session_boot.boot(cwd=str(tmp_path), extra_args=["-p", "query", "--permission-mode", "default"])
|
||||
cmd = mock_exec.call_args[0][1]
|
||||
assert cmd.count("--permission-mode") == 1
|
||||
assert "default" in cmd
|
||||
|
||||
|
||||
class TestPermissionModeDedupe:
|
||||
def test_no_extra_args_includes_default(self, tmp_path):
|
||||
with (
|
||||
|
||||
Reference in New Issue
Block a user