|
| 1 | +"""Mock Claude Code CLI for E2E hook testing. |
| 2 | +
|
| 3 | +Simulates the Claude Code hook protocol: |
| 4 | +- Feeds JSON input via stdin to hook scripts |
| 5 | +- Captures stdout/stderr output |
| 6 | +- Validates JSON responses match expected hook contract |
| 7 | +""" |
| 8 | +import json |
| 9 | +import os |
| 10 | +import subprocess |
| 11 | +import sys |
| 12 | +from dataclasses import dataclass, field |
| 13 | +from pathlib import Path |
| 14 | +from typing import Any, Dict, List, Optional |
| 15 | + |
| 16 | + |
| 17 | +# Resolve hooks directory relative to this project |
| 18 | +_PROJECT_ROOT = Path(__file__).resolve().parents[3] |
| 19 | +HOOKS_DIR = _PROJECT_ROOT / "packages" / "claude-code-plugin" / "hooks" |
| 20 | + |
| 21 | + |
| 22 | +@dataclass |
| 23 | +class HookResult: |
| 24 | + """Result of executing a hook script.""" |
| 25 | + |
| 26 | + exit_code: int |
| 27 | + stdout: str |
| 28 | + stderr: str |
| 29 | + json_output: Optional[Dict[str, Any]] = None |
| 30 | + |
| 31 | + @property |
| 32 | + def succeeded(self) -> bool: |
| 33 | + """Hook must always exit 0 (never block Claude Code).""" |
| 34 | + return self.exit_code == 0 |
| 35 | + |
| 36 | + @property |
| 37 | + def has_json(self) -> bool: |
| 38 | + return self.json_output is not None |
| 39 | + |
| 40 | + @property |
| 41 | + def additional_context(self) -> Optional[str]: |
| 42 | + """Extract additionalContext from hookSpecificOutput.""" |
| 43 | + if not self.json_output: |
| 44 | + return None |
| 45 | + hso = self.json_output.get("hookSpecificOutput", {}) |
| 46 | + return hso.get("additionalContext") |
| 47 | + |
| 48 | + @property |
| 49 | + def status_message(self) -> Optional[str]: |
| 50 | + """Extract statusMessage from hookSpecificOutput.""" |
| 51 | + if not self.json_output: |
| 52 | + return None |
| 53 | + hso = self.json_output.get("hookSpecificOutput", {}) |
| 54 | + return hso.get("statusMessage") |
| 55 | + |
| 56 | + @property |
| 57 | + def system_message(self) -> Optional[str]: |
| 58 | + """Extract systemMessage from top-level output.""" |
| 59 | + if not self.json_output: |
| 60 | + return None |
| 61 | + return self.json_output.get("systemMessage") |
| 62 | + |
| 63 | + |
| 64 | +@dataclass |
| 65 | +class MockEnvironment: |
| 66 | + """Isolated environment for hook execution.""" |
| 67 | + |
| 68 | + home_dir: str |
| 69 | + project_dir: str |
| 70 | + env_vars: Dict[str, str] = field(default_factory=dict) |
| 71 | + |
| 72 | + def build_env(self) -> Dict[str, str]: |
| 73 | + """Build environment variables for hook subprocess.""" |
| 74 | + env = os.environ.copy() |
| 75 | + env["HOME"] = self.home_dir |
| 76 | + env["CLAUDE_PROJECT_DIR"] = self.project_dir |
| 77 | + env["CLAUDE_CWD"] = self.project_dir |
| 78 | + env["CLAUDE_PLUGIN_DIR"] = str(HOOKS_DIR.parent) |
| 79 | + env["CLAUDE_PLUGIN_ROOT"] = str(HOOKS_DIR.parent) |
| 80 | + # Isolate plugin data to temp dir |
| 81 | + env["CLAUDE_PLUGIN_DATA"] = os.path.join(self.home_dir, ".codingbuddy") |
| 82 | + # Prevent actual system language detection from interfering |
| 83 | + env["LANG"] = "en_US.UTF-8" |
| 84 | + env.update(self.env_vars) |
| 85 | + return env |
| 86 | + |
| 87 | + |
| 88 | +def run_hook( |
| 89 | + hook_script: str, |
| 90 | + input_data: Optional[Dict[str, Any]] = None, |
| 91 | + env: Optional[MockEnvironment] = None, |
| 92 | + timeout: int = 15, |
| 93 | +) -> HookResult: |
| 94 | + """Execute a hook script with simulated Claude Code protocol. |
| 95 | +
|
| 96 | + Args: |
| 97 | + hook_script: Filename of the hook in the hooks directory (e.g. "session-start.py"). |
| 98 | + input_data: JSON-serializable dict to feed via stdin. |
| 99 | + env: Mock environment for isolation. Uses real env if None. |
| 100 | + timeout: Max seconds before killing the process. |
| 101 | +
|
| 102 | + Returns: |
| 103 | + HookResult with captured output. |
| 104 | + """ |
| 105 | + script_path = HOOKS_DIR / hook_script |
| 106 | + if not script_path.exists(): |
| 107 | + raise FileNotFoundError(f"Hook script not found: {script_path}") |
| 108 | + |
| 109 | + stdin_bytes = json.dumps(input_data).encode() if input_data else b"" |
| 110 | + |
| 111 | + proc_env = env.build_env() if env else os.environ.copy() |
| 112 | + |
| 113 | + try: |
| 114 | + result = subprocess.run( |
| 115 | + [sys.executable, str(script_path)], |
| 116 | + input=stdin_bytes, |
| 117 | + capture_output=True, |
| 118 | + timeout=timeout, |
| 119 | + env=proc_env, |
| 120 | + cwd=env.project_dir if env else None, |
| 121 | + ) |
| 122 | + except subprocess.TimeoutExpired: |
| 123 | + return HookResult(exit_code=1, stdout="", stderr="TIMEOUT") |
| 124 | + |
| 125 | + stdout_text = result.stdout.decode("utf-8", errors="replace") |
| 126 | + stderr_text = result.stderr.decode("utf-8", errors="replace") |
| 127 | + |
| 128 | + # Try to parse stdout as JSON (hooks that use safe_main output JSON) |
| 129 | + json_output = None |
| 130 | + stdout_stripped = stdout_text.strip() |
| 131 | + if stdout_stripped: |
| 132 | + # Some hooks output plain text before JSON; try to find JSON at the end |
| 133 | + for candidate in [stdout_stripped, stdout_stripped.split("\n")[-1]]: |
| 134 | + try: |
| 135 | + json_output = json.loads(candidate) |
| 136 | + break |
| 137 | + except (json.JSONDecodeError, IndexError): |
| 138 | + continue |
| 139 | + |
| 140 | + return HookResult( |
| 141 | + exit_code=result.returncode, |
| 142 | + stdout=stdout_text, |
| 143 | + stderr=stderr_text, |
| 144 | + json_output=json_output, |
| 145 | + ) |
| 146 | + |
| 147 | + |
| 148 | +@dataclass |
| 149 | +class LifecycleRunner: |
| 150 | + """Simulates a full Claude Code session hook lifecycle. |
| 151 | +
|
| 152 | + Runs hooks in order: SessionStart → PreToolUse → PostToolUse → Stop |
| 153 | + """ |
| 154 | + |
| 155 | + env: MockEnvironment |
| 156 | + results: List[HookResult] = field(default_factory=list) |
| 157 | + |
| 158 | + def session_start(self) -> HookResult: |
| 159 | + """Execute SessionStart hook.""" |
| 160 | + result = run_hook("session-start.py", env=self.env) |
| 161 | + self.results.append(result) |
| 162 | + return result |
| 163 | + |
| 164 | + def user_prompt_submit(self, prompt: str) -> HookResult: |
| 165 | + """Execute UserPromptSubmit hook with a user prompt.""" |
| 166 | + result = run_hook( |
| 167 | + "user-prompt-submit.py", |
| 168 | + input_data={"prompt": prompt}, |
| 169 | + env=self.env, |
| 170 | + ) |
| 171 | + self.results.append(result) |
| 172 | + return result |
| 173 | + |
| 174 | + def pre_tool_use( |
| 175 | + self, |
| 176 | + tool_name: str, |
| 177 | + tool_input: Optional[Dict[str, Any]] = None, |
| 178 | + ) -> HookResult: |
| 179 | + """Execute PreToolUse hook.""" |
| 180 | + result = run_hook( |
| 181 | + "pre-tool-use.py", |
| 182 | + input_data={ |
| 183 | + "tool_name": tool_name, |
| 184 | + "tool_input": tool_input or {}, |
| 185 | + }, |
| 186 | + env=self.env, |
| 187 | + ) |
| 188 | + self.results.append(result) |
| 189 | + return result |
| 190 | + |
| 191 | + def post_tool_use( |
| 192 | + self, |
| 193 | + tool_name: str, |
| 194 | + tool_input: Optional[Dict[str, Any]] = None, |
| 195 | + tool_output: str = "", |
| 196 | + ) -> HookResult: |
| 197 | + """Execute PostToolUse hook.""" |
| 198 | + result = run_hook( |
| 199 | + "post-tool-use.py", |
| 200 | + input_data={ |
| 201 | + "tool_name": tool_name, |
| 202 | + "tool_input": tool_input or {}, |
| 203 | + "tool_output": tool_output, |
| 204 | + }, |
| 205 | + env=self.env, |
| 206 | + ) |
| 207 | + self.results.append(result) |
| 208 | + return result |
| 209 | + |
| 210 | + def stop(self) -> HookResult: |
| 211 | + """Execute Stop hook.""" |
| 212 | + result = run_hook("stop.py", input_data={}, env=self.env) |
| 213 | + self.results.append(result) |
| 214 | + return result |
| 215 | + |
| 216 | + @property |
| 217 | + def all_succeeded(self) -> bool: |
| 218 | + """Check all hooks exited with code 0.""" |
| 219 | + return all(r.succeeded for r in self.results) |
0 commit comments