slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(seedgo): archive pre-DPLAN-0184 hook cruft (FPLAN-0241)

Browse Source 
Archive orphaned hook bridge/probe/manifest modules + their tests to
.archive/ — their only callers were the .claude/hooks scripts disabled
by DPLAN-0184. Seedgo audits hooks via standards; the hooks branch owns
the engine/bridge/handlers. README + bypass.json + prompts updated to
match. 1045 tests green, pyright clean.
This commit is contained in:
AIOSAI
2026-06-05 19:44:35 -07:00
parent 8a843429ca
commit cc8f809a50
16 changed files with 8 additions and 3249 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/seedgo/.aipass/aipass_local_prompt.md
-2
View File
@@ -24,8 +24,6 @@ The **hooks branch** (`src/aipass/hooks/`) owns all hook infrastructure — engi
Provider settings route all events through the bridge: `src/aipass/hooks/apps/handlers/bridges/claude.py <Event>:<handler>`. The bridge dispatches to native Python handlers in `hooks/apps/handlers/` (prompt, security, lifecycle, notification categories).
Seedgo's bridge installer (`drone @seedgo bridge install`) manages hook installation to `~/.claude/settings.json`.
## Apps Layout (extra layer vs standard branch)
```
src/aipass/seedgo/.seedgo/bypass.json
-25
View File
@@ -10,11 +10,6 @@
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_hooks_probe.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_checkers_batch5.py",
"standard": "encapsulation",
@@ -245,26 +240,6 @@
"standard": "unused_function",
"reason": "is_trusted_caller() and identify_caller() are public API consumed by pre_edit_gate.py (hook layer) and drone auth.py. Checker cannot trace cross-file dynamic dispatch to the hook scripts."
},
{
"file": "apps/handlers/hooks/bridge_handler.py",
"standard": "help_text",
"reason": "python3 references appear in AIPASS_HOOK_MANIFEST command strings (hook definitions), not user-facing help text. Data constants, not documentation."
},
{
"file": "tests/test_hook_bridge.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_hook_bridge.py",
"standard": "encapsulation",
"reason": "Unit tests must import handlers directly to test them in isolation. Same pattern as test_checkers_batch5.py."
},
{
"file": "tests/test_hook_bridge.py",
"standard": "help_text",
"reason": "python3 references appear in test data strings (hook command fixtures), not user-facing help text."
},
{
"file": "tests/test_coverage_audit.py",
"standard": "architecture",
src/aipass/seedgo/README.md
+6 -29
View File
@@ -16,12 +16,10 @@
- Score files 0-100 per standard and report violations with actionable details
- Manage bypass rules (`.seedgo/bypass.json`) for deliberate exceptions
- Run pyright diagnostics across branches for type error detection
- Hook bridge installer (`drone @seedgo bridge install`) for `~/.claude/settings.json` management
- Single-file checklist validation against all standards (consumed by PostToolUse auto-fix hook)
- Proof certification via proof/proof_query (triplet, plugin integrity, README currency)
- Custom function test coverage mapping via test_map
- README auto-generation and freshness checking
- Hook testing, listing, and probe infrastructure
### What I Don't Do
- Runtime monitoring (that's prax)
@@ -65,17 +63,6 @@ drone @seedgo test_map @seedgo # Function-level test cov
# README
drone @seedgo readme update @flow # README auto-generation for a branch
# Hooks
drone @seedgo hooks test # Run hook test suite (pytest per file)
drone @seedgo hooks list # Show all wired hooks from settings
drone @seedgo hooks probe # Hook event probe infrastructure
# Bridge
drone @seedgo bridge install # Install AIPass hooks to ~/.claude/settings.json
drone @seedgo bridge uninstall # Remove AIPass hooks
drone @seedgo bridge reinstall # Clean reinstall
drone @seedgo bridge status # Show hook installation status
```
### Via Python Module
@@ -97,21 +84,18 @@ seedgo/
│ ├── seedgo.py # Entry point — thin router (~290 lines)
│ │ # discover_modules() loads apps/modules/*.py
│ │ # route_command() dispatches to first handler returning True
│ ├── modules/ # 13 business logic modules
│ ├── modules/ # 9 business logic modules
│ │ ├── standards_audit.py # Pack-aware compliance audit orchestrator
│ │ ├── standards_query.py # Pack-aware content query
│ │ ├── diagnostics_audit.py # Pyright diagnostics via audit pipeline
│ │ ├── checklist.py # Per-file/dir standards check (hook consumer)
│ │ ├── seedgo_proof.py # Proof certification orchestrator
│ │ ├── proof_query.py # Proof content query
│ │ ├── hook_bridge.py # Hook bridge installer (install/uninstall/reinstall/status)
│ │ ├── hooks.py # Hook probe display and testing
│ │ ├── hooks_ext.py # Hook test + list subcommands (split from hooks.py)
│ │ ├── inbox_audit.py # Inbox message-ID validation
│ │ ├── permissions.py # TRUSTED_CROSS_WRITERS list for hook + drone auth
│ │ ├── readme_update.py # README generation module
│ │ └── test_map.py # Custom function test coverage mapping
│ └── handlers/ # 11 handler directories
│ └── handlers/ # 10 handler directories
│ ├── aipass_standards/ # 34 checker standards (67 files)
│ │ ├── *_check.py # Checker implementations (score 0-100)
│ │ ├── *_content.py # Queryable standard content
@@ -132,11 +116,10 @@ seedgo/
│ ├── config/ # Configuration handlers
│ ├── diagnostics/ # Pyright integration + branch discovery
│ ├── file/ # File operations
│ ├── hooks/ # Hook test runner
│ ├── json/ # JSON tracking (json_handler)
│ ├── readme/ # README generator + branch resolution
│ └── test_map/ # Function test coverage scanner
├── tests/ # 39 test files, 1192 tests
├── tests/ # 34 test files, 1045 tests
├── drone_adapter.py # Drone routing bridge
├── .trinity/ # Identity + memory
├── .seedgo/ # Self-bypass rules
@@ -221,16 +204,13 @@ Provider settings route all events through the bridge (`claude.py`), which dispa
| `lifecycle.compact` | PreCompact | lifecycle |
| `lifecycle.rollover` | PreCompact | lifecycle |
The `bridge` module (`drone @seedgo bridge install`) manages hook installation to `~/.claude/settings.json`.
---
## Tests
- **39 test files**, **1192 tests**, all passing
- **200 public functions**, 200 tested (100% coverage)
- **34 test files**, all passing
- **0 type errors** (pyright)
- Key test areas: standards audit, checklist, bypass, JSON handler, hooks (probe, track A/B/E, utility, bridge), proof, README, diagnostics, line coverage (plugin integrity, diagnostics, audit display, branch audit, architecture, checklist)
- Key test areas: standards audit, checklist, bypass, JSON handler, hooks snapshot, permissions, proof, README, diagnostics, line coverage (plugin integrity, diagnostics, audit display, branch audit, architecture, checklist)
---
@@ -253,13 +233,10 @@ The `bridge` module (`drone @seedgo bridge install`) manages hook installation t
## Known Issues / Tech Debt
- `audit_display.py`: 16 hardcoded display blocks for specific standards (DPLAN-0047 tracks dynamic refactor)
- `bridge_handler.py` AIPASS_HOOK_MANIFEST references old script paths — needs update to bridge architecture
- `test_hooks_track_a/b/e.py` import old hook scripts from `.claude/hooks/` — should migrate to test new native handlers
- `proof`, `proof_query`, `test_map` not listed in `--help` output
- `documentation_check.py` 5-line lookahead limitation for multi-line function signatures
- `dead_code_check.py` doesn't recognize `iterdir()` as valid discovery pattern
- Cross-branch file write detection recommended but not yet in standards (S73 finding)
- Test coverage: 200/200 public functions (100%) but line coverage still has gaps in 6 handler files
---
@@ -272,7 +249,7 @@ The `bridge` module (`drone @seedgo bridge install`) manages hook installation t
---
**Last Updated:** 2026-05-16
**Last Updated:** 2026-06-05
---
[<- Back to AIPass](../../../README.md)
src/aipass/seedgo/apps/handlers/hooks/__init__.py
-47
View File
@@ -1,47 +0,0 @@
# =================== AIPass ====================
# Name: hooks/__init__.py
# Description: Hook test runner — subprocess execution for hooks_ext module
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""Hook test runner handler.
Encapsulates subprocess execution so hooks_ext module stays
at the display/coordination layer.
"""
import re
import subprocess
import sys
import time
from pathlib import Path
def run_pytest_file(test_file: Path, repo_root: Path, timeout: int = 60) -> tuple[int, int, float]:
"""Run pytest on a single test file. Returns (passed, failed, duration_seconds)."""
t0 = time.monotonic()
proc = subprocess.run(
[sys.executable, "-m", "pytest", str(test_file), "--tb=no", "-q", "--no-header"],
capture_output=True,
text=True,
timeout=timeout,
cwd=str(repo_root),
)
duration = time.monotonic() - t0
passed = 0
failed = 0
for line in proc.stdout.splitlines():
line = line.strip()
if "passed" in line or "failed" in line or "error" in line.lower():
m_passed = re.search(r"(\d+) passed", line)
m_failed = re.search(r"(\d+) failed", line)
m_error = re.search(r"(\d+) error", line)
if m_passed:
passed = int(m_passed.group(1))
if m_failed:
failed = int(m_failed.group(1))
if m_error:
failed += int(m_error.group(1))
return passed, failed, duration
src/aipass/seedgo/apps/handlers/hooks/bridge_handler.py
-248
View File
@@ -1,248 +0,0 @@
# =================== AIPass ====================
# Name: bridge_handler.py
# Description: Hook bridge — settings.json I/O for hook installation
# Version: 1.0.0
# Created: 2026-04-22
# Modified: 2026-04-22
# =============================================
"""Hook bridge handler — reads/writes Claude Code settings.json for hook management.
Provides AIPASS_HOME resolution, hook manifest, install/uninstall logic.
The module layer (hook_bridge.py) handles command routing and display.
"""
from __future__ import annotations
import json
import os
import shutil
from pathlib import Path
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
# ---------------------------------------------------------------------------
# AIPASS_HOME resolution
# ---------------------------------------------------------------------------
def resolve_aipass_home() -> Path | None:
"""Resolve AIPASS_HOME from env var or importlib fallback."""
env_home = os.environ.get("AIPASS_HOME")
if env_home:
p = Path(env_home)
if p.is_dir():
return p
try:
import importlib.util
spec = importlib.util.find_spec("aipass")
if spec and spec.submodule_search_locations:
pkg_dir = Path(list(spec.submodule_search_locations)[0])
project_root = pkg_dir.parents[1]
if (project_root / ".aipass").exists():
return project_root
except Exception as exc:
logger.info("bridge_handler: importlib fallback failed: %s", exc)
return None
# ---------------------------------------------------------------------------
# Settings I/O
# ---------------------------------------------------------------------------
def read_settings(path: Path) -> dict:
"""Read and parse settings.json. Returns {} on failure."""
if not path.exists():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception as exc:
logger.warning("bridge_handler: could not read %s: %s", path, exc)
return {}
def write_settings(path: Path, data: dict) -> bool:
"""Write settings.json with backup and pretty formatting."""
try:
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
shutil.copy2(path, path.with_suffix(".json.bak"))
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
return True
except Exception as exc:
logger.warning("bridge_handler: could not write %s: %s", path, exc)
return False
# ---------------------------------------------------------------------------
# Hook detection
# ---------------------------------------------------------------------------
_AIPASS_COMMAND_MARKERS = (
"AIPass/.claude/hooks/",
"$AIPASS_HOME",
"aipass_global_prompt",
"aipass/hooks/apps/handlers/bridges/claude.py",
)
def is_aipass_hook_entry(entry: dict) -> bool:
"""Check if a hook entry is an AIPass hook (marked or pattern-matched)."""
if entry.get("_aipass"):
return True
for hook in entry.get("hooks", []):
cmd = hook.get("command", "")
if any(marker in cmd for marker in _AIPASS_COMMAND_MARKERS):
return True
return False
def count_aipass_hooks(settings: dict) -> int:
"""Count AIPass hook entries across all events."""
count = 0
for entries in settings.get("hooks", {}).values():
if isinstance(entries, list):
for entry in entries:
if isinstance(entry, dict) and is_aipass_hook_entry(entry):
count += 1
return count
# ---------------------------------------------------------------------------
# Hook manifest — the canonical set of AIPass hooks
# ---------------------------------------------------------------------------
_BRIDGE = "$AIPASS_HOME/.venv/bin/python3 $AIPASS_HOME/src/aipass/hooks/apps/handlers/bridges/claude.py"
AIPASS_HOOK_MANIFEST: dict[str, list[dict]] = {
"UserPromptSubmit": [
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} UserPromptSubmit:global_prompt"}],
},
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} UserPromptSubmit:branch_prompt"}],
},
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} UserPromptSubmit:identity_injector"}],
},
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} UserPromptSubmit:email_notification"}],
},
],
"PreToolUse": [
{
"_aipass": True,
"matcher": "Bash|Edit|MultiEdit|Write|Read|Grep|Glob|WebSearch|WebFetch|Task",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PreToolUse"}],
},
],
"PostToolUse": [
{
"_aipass": True,
"matcher": "Bash|Edit|MultiEdit|Write|NotebookEdit",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PostToolUse"}],
},
],
"SubagentStop": [
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} SubagentStop"}],
},
],
"Stop": [
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} Stop"}],
},
],
"Notification": [
{
"_aipass": True,
"hooks": [{"type": "command", "command": f"{_BRIDGE} Notification"}],
},
],
"PreCompact": [
{
"_aipass": True,
"matcher": "manual",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PreCompact:pre_compact", "timeout": 60}],
},
{
"_aipass": True,
"matcher": "auto",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PreCompact:pre_compact", "timeout": 60}],
},
{
"_aipass": True,
"matcher": "manual",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PreCompact:pre_compact_rollover", "timeout": 120}],
},
{
"_aipass": True,
"matcher": "auto",
"hooks": [{"type": "command", "command": f"{_BRIDGE} PreCompact:pre_compact_rollover", "timeout": 120}],
},
],
}
# ---------------------------------------------------------------------------
# Install / uninstall
# ---------------------------------------------------------------------------
def ensure_aipass_env(settings: dict, aipass_home: str) -> bool:
"""Ensure AIPASS_HOME is in the env section. Returns True if added."""
env = settings.setdefault("env", {})
if "AIPASS_HOME" in env:
return False
env["AIPASS_HOME"] = aipass_home
return True
def install_hooks(settings: dict) -> tuple[dict, int]:
"""Add AIPass hooks to settings. Skips events that already have AIPass entries."""
hooks = settings.setdefault("hooks", {})
added = 0
for event_name, manifest_entries in AIPASS_HOOK_MANIFEST.items():
existing = hooks.get(event_name, [])
if not isinstance(existing, list):
existing = []
has_aipass = any(is_aipass_hook_entry(e) for e in existing if isinstance(e, dict))
if has_aipass:
continue
existing.extend(manifest_entries)
hooks[event_name] = existing
added += len(manifest_entries)
json_handler.log_operation("bridge_install_hooks", {"added": added})
return settings, added
def uninstall_hooks(settings: dict) -> tuple[dict, int]:
"""Remove AIPass hook entries only. Preserves all non-AIPass entries."""
hooks = settings.get("hooks", {})
removed = 0
for event_name in list(hooks.keys()):
entries = hooks[event_name]
if not isinstance(entries, list):
continue
filtered = []
for entry in entries:
if isinstance(entry, dict) and is_aipass_hook_entry(entry):
removed += 1
else:
filtered.append(entry)
if filtered:
hooks[event_name] = filtered
else:
del hooks[event_name]
json_handler.log_operation("bridge_uninstall_hooks", {"removed": removed})
return settings, removed
src/aipass/seedgo/apps/modules/hook_bridge.py
-182
View File
@@ -1,182 +0,0 @@
# =================== AIPass ====================
# Name: hook_bridge.py
# Description: Hook bridge installer — adds/removes AIPass hooks in settings
# Version: 1.0.0
# Created: 2026-04-22
# Modified: 2026-04-22
# =============================================
"""Hook bridge installer — adds/removes AIPass hooks in Claude Code settings.
bridge install Add AIPass hooks to ~/.claude/settings.json (idempotent)
bridge uninstall Remove only AIPass-added hook entries
bridge reinstall Remove + re-add (upgrade path)
bridge status Show current installation state
"""
from pathlib import Path
from aipass.prax import logger
from aipass.cli import console
from aipass.cli.apps.modules import error, warning
from aipass.seedgo.apps.handlers.hooks.bridge_handler import (
count_aipass_hooks,
ensure_aipass_env,
install_hooks,
read_settings,
resolve_aipass_home,
uninstall_hooks,
write_settings,
)
from aipass.seedgo.apps.handlers.json import json_handler
SETTINGS_PATH = Path.home() / ".claude" / "settings.json"
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def _run_install() -> None:
"""Install AIPass hooks into global settings."""
aipass_home = resolve_aipass_home()
if not aipass_home:
error("Could not resolve AIPASS_HOME — set env var or install aipass package")
return
settings = read_settings(SETTINGS_PATH)
existing = count_aipass_hooks(settings)
if existing > 0:
warning(f"AIPass hooks already installed ({existing} entries). Use 'bridge reinstall' to replace.")
return
ensure_aipass_env(settings, str(aipass_home))
settings, added = install_hooks(settings)
if added == 0:
warning("No hooks to add (all events already populated).")
return
if not write_settings(SETTINGS_PATH, settings):
error("Failed to write settings.json")
return
console.print(f"[green]Installed {added} AIPass hook entries into {SETTINGS_PATH}[/green]")
console.print(f"[dim]AIPASS_HOME={aipass_home}[/dim]")
json_handler.log_operation("bridge_install", {"added": added, "aipass_home": str(aipass_home)})
def _run_uninstall() -> None:
"""Remove AIPass hooks from global settings."""
settings = read_settings(SETTINGS_PATH)
existing = count_aipass_hooks(settings)
if existing == 0:
warning("No AIPass hooks found in settings.")
return
settings, removed = uninstall_hooks(settings)
if not write_settings(SETTINGS_PATH, settings):
error("Failed to write settings.json")
return
console.print(f"[green]Removed {removed} AIPass hook entries from {SETTINGS_PATH}[/green]")
json_handler.log_operation("bridge_uninstall", {"removed": removed})
def _run_reinstall() -> None:
"""Uninstall then install AIPass hooks."""
aipass_home = resolve_aipass_home()
if not aipass_home:
error("Could not resolve AIPASS_HOME")
return
settings = read_settings(SETTINGS_PATH)
settings, removed = uninstall_hooks(settings)
ensure_aipass_env(settings, str(aipass_home))
settings, added = install_hooks(settings)
if not write_settings(SETTINGS_PATH, settings):
error("Failed to write settings.json")
return
console.print(f"[green]Reinstalled: removed {removed}, added {added} hook entries[/green]")
json_handler.log_operation("bridge_reinstall", {"removed": removed, "added": added})
def _run_status() -> None:
"""Show current bridge installation status."""
settings = read_settings(SETTINGS_PATH)
count = count_aipass_hooks(settings)
aipass_home = resolve_aipass_home()
console.print()
console.print("[bold cyan]Hook Bridge Status[/bold cyan]")
console.print(f" Settings: {SETTINGS_PATH}")
console.print(f" AIPASS_HOME: {aipass_home or '[red]not resolved[/red]'}")
status = "[green](installed)[/green]" if count > 0 else "[dim](not installed)[/dim]"
console.print(f" AIPass hooks: {count} entries {status}")
console.print()
# ---------------------------------------------------------------------------
# Introspection + routing
# ---------------------------------------------------------------------------
def print_help() -> None:
"""CLI --help entry point."""
print_introspection()
def print_introspection() -> None:
"""Display module info and usage."""
console.print()
console.print("[bold cyan]hook_bridge Module[/bold cyan]")
console.print("Installs/removes AIPass hooks in Claude Code settings.json")
console.print()
console.print("[yellow]Subcommands:[/yellow]")
console.print(" [green]drone @seedgo bridge install[/green] [dim]# Add AIPass hooks (idempotent)[/dim]")
console.print(" [green]drone @seedgo bridge uninstall[/green] [dim]# Remove AIPass hooks only[/dim]")
console.print(" [green]drone @seedgo bridge reinstall[/green] [dim]# Remove + re-add (upgrade)[/dim]")
console.print(" [green]drone @seedgo bridge status[/green] [dim]# Show installation status[/dim]")
console.print()
console.print("[yellow]Connected Handlers:[/yellow]")
console.print(" [cyan]handlers/hooks/[/cyan]")
console.print(" [dim]- bridge_handler.py (settings I/O, hook manifest)[/dim]")
console.print()
def handle_command(command: str, args: list) -> bool:
"""Handle 'bridge' command — hook installation management."""
if command != "bridge":
return False
if not args:
print_introspection()
return True
if args[0] in ("--help", "-h", "help"):
print_help()
return True
subcommand = args[0]
if subcommand == "install":
_run_install()
return True
if subcommand == "uninstall":
_run_uninstall()
return True
if subcommand == "reinstall":
_run_reinstall()
return True
if subcommand == "status":
_run_status()
return True
logger.info("hook_bridge: unknown subcommand %r", subcommand)
console.print(f"[dim]Unknown subcommand: {subcommand!r}[/dim]")
print_introspection()
return True
src/aipass/seedgo/apps/modules/hooks.py
-489
View File
@@ -1,489 +0,0 @@
# =================== AIPass ====================
# Name: hooks.py
# Description: Hook Probe, Test, and List Module
# Version: 1.1.0
# Created: 2026-04-20
# Modified: 2026-04-21
# =============================================
"""
Hook Probe, Test, and List Module
Reads last_ping.jsonl produced by .claude/hooks/probes/ scripts and
surfaces probe data via Rich tables and reports.
Run: drone @seedgo hooks probe [--subagent|--matrix]
Subcommands:
hooks probe Show recent probe entries as a Rich table
hooks probe --subagent Spawn headless Claude, check if PostToolUse/SubagentStop fired
hooks probe --matrix Full event matrix + markdown report
hooks test Run hook test suite, display per-file pass/fail table
hooks list Show every wired hook from project and global settings
"""
import json
import subprocess
import sys
import time
from datetime import datetime
from pathlib import Path
from typing import List
# =============================================================================
# INFRASTRUCTURE SETUP
# =============================================================================
# IMPORTS
# =============================================================================
# Prax logger (system-wide, always first)
from aipass.prax import logger
# CLI services (display/output formatting)
from aipass.cli import console
from aipass.cli.apps.modules import warning
# JSON handler for tracking
from aipass.seedgo.apps.handlers.json import json_handler
# File handler — modules must not call open()/write_text() directly
from aipass.seedgo.apps.handlers.file import write_text_safe
# Extended subcommands (test + list)
from aipass.seedgo.apps.modules.hooks_ext import cmd_hooks_list, run_hooks_report, run_hooks_test
# Probe matrix (split to keep this file under 600 lines)
from aipass.seedgo.apps.modules.hooks_probe import run_probe_matrix
# Rich output
from rich.panel import Panel
from rich.table import Table
# =============================================================================
# PATHS
# =============================================================================
_REPO_ROOT: Path | None = None
# Enable snippet shown when no probe data exists — uses a placeholder path
# so the help_text checker does not flag a literal drone command path.
_PROBE_ENABLE_NOTE = (
"Add probe hook commands to .claude/settings.json (inside the hooks object).\n"
"See .claude/hooks/probes/README.md for per-event snippets."
)
def _get_repo_root() -> Path:
"""Return the AIPass repo root derived from this file's location."""
global _REPO_ROOT
if _REPO_ROOT is not None:
return _REPO_ROOT
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
_REPO_ROOT = parent
return parent
# Fallback: assume 5 levels up from this file
_REPO_ROOT = Path(__file__).resolve().parents[5]
return _REPO_ROOT
def _log_file() -> Path:
"""Return path to last_ping.jsonl."""
return _get_repo_root() / ".claude" / "hooks" / "probes" / "last_ping.jsonl"
def _probes_dir() -> Path:
"""Return path to probes directory."""
return _get_repo_root() / ".claude" / "hooks" / "probes"
# =============================================================================
# JSONL READING
# =============================================================================
def _parse_jsonl_line(line: str) -> dict | None:
"""Parse a single JSONL line. Returns None on parse failure."""
stripped = line.strip()
if not stripped:
return None
try:
return json.loads(stripped)
except json.JSONDecodeError as exc:
logger.info("hooks.py: skipping malformed JSONL line: %s", exc)
return None
def _read_entries(log_path: Path | None = None) -> List[dict]:
"""Read all entries from last_ping.jsonl. Returns empty list if missing or unreadable."""
path = log_path or _log_file()
if not path.exists():
return []
try:
with open(path, "r", encoding="utf-8") as fh:
raw_lines = fh.readlines()
except OSError as exc:
logger.info("hooks.py: could not read %s: %s", path, exc)
return []
entries = []
for line in raw_lines:
entry = _parse_jsonl_line(line)
if entry is not None:
entries.append(entry)
return entries
def _truncate(s: str, n: int) -> str:
"""Truncate string to n chars with ellipsis."""
if len(s) <= n:
return s
return s[: n - 3] + "..."
# =============================================================================
# SUBCOMMAND: hooks probe (no flags)
# =============================================================================
def _cmd_probe_display(log_path: Path | None = None) -> None:
"""Display recent probe entries as a Rich table."""
entries = _read_entries(log_path)
if not entries:
console.print(
Panel(
"[yellow]No probe data yet.[/yellow]\n\n"
"Enable probes by adding hook commands to [cyan].claude/settings.json[/cyan].\n"
f"{_PROBE_ENABLE_NOTE}",
title="[bold cyan][PROBE][/bold cyan]",
border_style="cyan",
)
)
return
# Show last 50 entries
recent = entries[-50:]
table = Table(
title="[bold cyan][PROBE][/bold cyan] Recent Hook Pings",
show_header=True,
header_style="bold cyan",
border_style="dim",
expand=False,
)
table.add_column("Event", style="cyan", no_wrap=True)
table.add_column("Tool", style="green")
table.add_column("Timestamp", style="dim")
table.add_column("CWD", style="dim")
table.add_column("Agent ID", style="dim")
table.add_column("Elapsed ms", justify="right", style="yellow")
table.add_column("CLI Ver", style="dim")
table.add_column("ProjectDir", justify="center")
table.add_column("AIPassHome", justify="center")
for e in recent:
table.add_row(
e.get("event", "?"),
_truncate(e.get("tool", ""), 20),
_truncate(e.get("timestamp", ""), 24),
_truncate(e.get("cwd", ""), 30),
_truncate(e.get("agent_id", ""), 14),
str(round(e.get("script_elapsed_ms", 0), 1)),
_truncate(e.get("cli_version", "?"), 10),
"[green]Y[/green]" if e.get("env_has_claude_project_dir") else "[red]N[/red]",
"[green]Y[/green]" if e.get("env_has_aipass_home") else "[red]N[/red]",
)
console.print()
console.print(table)
console.print(f"\n[dim]Showing {len(recent)} of {len(entries)} total entries from {_log_file()}[/dim]\n")
json_handler.log_operation("hooks_probe_display", {"entries_shown": len(recent), "total": len(entries)})
# =============================================================================
# SUBCOMMAND: hooks probe --subagent
# =============================================================================
def _run_headless_claude() -> int:
"""Spawn headless claude with a Read tool call. Returns exit code."""
canary = Path("/tmp/probe_canary.txt")
if not write_text_safe(canary, "probe canary 2026-04-20\n"):
logger.info("hooks.py: could not write canary")
console.print("[dim]Spawning headless claude...[/dim]")
# --permission-mode bypassPermissions is the AIPass-approved bypass flag
try:
result = subprocess.run(
[
"claude",
"-p",
"--permission-mode",
"bypassPermissions",
"read the file /tmp/probe_canary.txt",
"--allowedTools",
"Read",
],
capture_output=True,
text=True,
timeout=30,
)
console.print(f"[dim]claude exited: {result.returncode}[/dim]")
return result.returncode
except FileNotFoundError as exc:
logger.info("hooks.py: claude binary not found: %s", exc)
warning("claude binary not found — skipping spawn test")
return -1
except subprocess.TimeoutExpired as exc:
logger.info("hooks.py: claude timed out: %s", exc)
warning("claude timed out after 30s")
return -2
except OSError as exc:
logger.info("hooks.py: spawn error: %s", exc)
warning(f"spawn error: {exc}")
return -3
def _cmd_probe_subagent() -> None:
"""Spawn a headless Claude Code process and check if PostToolUse/SubagentStop fired."""
console.print()
console.print(
Panel(
"[bold cyan]Headless Probe — PostToolUse + SubagentStop[/bold cyan]\n\n"
"This test spawns a headless claude process with a Read tool call, "
"then checks last_ping.jsonl for "
"[yellow]PostToolUse[/yellow] and [yellow]SubagentStop[/yellow] entries "
"created within the last 10 seconds.\n\n"
"[dim]Requires PostToolUse and SubagentStop probes to be enabled in settings.json.[/dim]",
title="[bold cyan][PROBE][/bold cyan]",
border_style="cyan",
)
)
start_ts = time.time()
_run_headless_claude()
# Wait a beat for hooks to flush
time.sleep(0.5)
# Check last_ping.jsonl for recent entries
cutoff = start_ts - 1.0 # 1s before spawn
entries = _read_entries()
recent_events = {e.get("event") for e in entries if _entry_ts(e) >= cutoff}
post_tool_use_fired = "PostToolUse" in recent_events
subagent_stop_fired = "SubagentStop" in recent_events
result_table = Table(show_header=True, header_style="bold cyan", border_style="dim")
result_table.add_column("Check", style="cyan")
result_table.add_column("Result", justify="center")
result_table.add_row(
"PostToolUse fired in headless",
"[green]YES[/green]" if post_tool_use_fired else "[red]NO[/red]",
)
result_table.add_row(
"SubagentStop fired in headless",
"[green]YES[/green]" if subagent_stop_fired else "[red]NO[/red]",
)
console.print(result_table)
console.print()
# Manual test section
console.print(
Panel(
"[bold yellow]Manual Interactive Agent-Tool Test[/bold yellow]\n\n"
"To test hooks fired by the interactive Agent tool:\n\n"
"1. Enable PostToolUse and SubagentStop probes in .claude/settings.json\n"
"2. In an interactive Claude Code session, invoke an Agent tool call\n"
"3. After the agent completes, run:\n"
" [cyan]drone @seedgo hooks probe[/cyan]\n"
"4. Check for SubagentStop and PostToolUse entries near the agent's timestamp\n\n"
"[dim]Headless (-p) and interactive Agent tool have different session contexts.[/dim]\n"
"[dim]SubagentStop fires when an agent tool invocation completes.[/dim]",
title="[bold cyan][PROBE][/bold cyan] Manual Test",
border_style="dim",
)
)
json_handler.log_operation(
"hooks_probe_subagent",
{"post_tool_use_fired": post_tool_use_fired, "subagent_stop_fired": subagent_stop_fired},
)
def _entry_ts(entry: dict) -> float:
"""Parse entry timestamp to unix float. Returns 0 on failure."""
ts_str = entry.get("timestamp", "")
if not ts_str:
return 0.0
try:
normalized = ts_str[:-1] + "+00:00" if ts_str.endswith("Z") else ts_str
dt = datetime.fromisoformat(normalized)
return dt.timestamp()
except ValueError as exc:
logger.info("hooks.py: could not parse timestamp %r: %s", ts_str, exc)
return 0.0
# =============================================================================
# SUBCOMMAND: hooks probe --matrix (delegated to hooks_probe)
# =============================================================================
def _cmd_probe_matrix() -> None:
"""Group entries by event, show matrix, write markdown report — delegates to hooks_probe."""
run_probe_matrix(_read_entries, _probes_dir)
# =============================================================================
# SUBCOMMAND: hooks test + hooks list (delegated to hooks_ext)
# =============================================================================
def _cmd_hooks_test() -> None:
"""Run hook test suite — delegates to hooks_ext."""
run_hooks_test(_get_repo_root())
def _cmd_hooks_list() -> None:
"""Show wired hooks — delegates to hooks_ext."""
cmd_hooks_list(_get_repo_root())
def _cmd_hooks_report(args: list) -> None:
"""Run hook execution report — delegates to hooks_ext."""
run_hooks_report(_get_repo_root(), args)
# =============================================================================
# INTROSPECTION
# =============================================================================
def print_help() -> None:
"""CLI --help entry point — delegates to print_introspection."""
print_introspection()
def print_introspection() -> None:
"""Display module info and usage."""
console.print()
console.print("[bold cyan]hooks Module[/bold cyan]")
console.print("Hook probe display — reads last_ping.jsonl from .claude/hooks/probes/")
console.print()
log = _log_file()
if log.exists():
entries = _read_entries()
console.print(f"[yellow]Probe log:[/yellow] {log} ([green]{len(entries)} entries[/green])")
else:
console.print(f"[yellow]Probe log:[/yellow] {log} [dim](not yet created — enable probes first)[/dim]")
console.print()
console.print("[yellow]Subcommands:[/yellow]")
console.print(" [green]drone @seedgo hooks probe[/green] [dim]# Display recent entries table[/dim]")
console.print(
" [green]drone @seedgo hooks probe --subagent[/green] [dim]# Test headless PostToolUse/SubagentStop[/dim]"
)
console.print(
" [green]drone @seedgo hooks probe --matrix[/green] [dim]# Full event matrix + markdown report[/dim]"
)
console.print(" [green]drone @seedgo hooks test[/green] [dim]# Run hook test suite[/dim]")
console.print(" [green]drone @seedgo hooks report[/green] [dim]# Hook execution log report[/dim]")
console.print(" [green]drone @seedgo hooks list[/green] [dim]# List all wired hooks[/dim]")
console.print()
console.print("[yellow]Probe scripts:[/yellow]")
probes_dir = _probes_dir()
for script in sorted(probes_dir.glob("probe_*.py")):
console.print(f" [dim]{script.name}[/dim]")
console.print()
console.print("[yellow]Connected Handlers:[/yellow]")
console.print(" [cyan]handlers/json/[/cyan]")
console.print(" [dim]- json_handler.py (log_operation — operation tracking)[/dim]")
console.print()
console.print("[yellow]External Dependencies:[/yellow]")
console.print(" [dim]- aipass.prax (logger)[/dim]")
console.print(" [dim]- aipass.cli (console)[/dim]")
console.print()
# =============================================================================
# COMMAND HANDLER
# =============================================================================
def handle_command(command: str, args: List[str]) -> bool:
"""
Handle 'hooks' command — hook probe display and testing.
Args:
command: Command name
args: Additional arguments
[] -> print_introspection()
["probe"] -> display last_ping.jsonl table
["probe", "--subagent"] -> headless probe test
["probe", "--matrix"] -> full event matrix + report
["--help"] | ["help"] -> print_introspection()
Returns:
True if handled, False if not this module's command
"""
if command != "hooks":
return False
if not args:
print_introspection()
return True
if args[0] in ("--help", "-h", "help"):
print_help()
return True
subcommand = args[0]
if subcommand == "probe":
sub_args = args[1:]
if "--subagent" in sub_args:
_cmd_probe_subagent()
elif "--matrix" in sub_args:
_cmd_probe_matrix()
else:
_cmd_probe_display()
return True
if subcommand == "test":
_cmd_hooks_test()
return True
if subcommand == "report":
_cmd_hooks_report(args[1:])
return True
if subcommand == "list":
_cmd_hooks_list()
return True
# Unknown subcommand — show introspection
console.print(f"[dim]Unknown subcommand: {subcommand!r} — showing help[/dim]")
print_introspection()
return True
# =============================================================================
# STANDALONE EXECUTION
# =============================================================================
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] in ("--help", "-h", "help"):
print_introspection()
sys.exit(0)
logger.info("Prax logger connected to hooks")
handle_command("hooks", sys.argv[1:])
src/aipass/seedgo/apps/modules/hooks_ext.py
-277
View File
@@ -1,277 +0,0 @@
# =================== AIPass ====================
# Name: hooks_ext.py
# Description: Hook Test and List Subcommands (split from hooks.py)
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""
Hook Test and List Subcommands
Extended subcommands for hooks.py — split out to keep hooks.py under 700 lines.
hooks test Run hook test suite, display per-file pass/fail table
hooks list Show every wired hook from project and global settings
"""
import glob as _glob
import json
from pathlib import Path
from aipass.prax import logger
from aipass.cli import console
from aipass.cli.apps.modules import warning
from aipass.seedgo.apps.handlers.file import read_lines_safe, read_text_safe
from aipass.seedgo.apps.handlers.hooks import run_pytest_file
from aipass.seedgo.apps.handlers.json import json_handler
from rich.table import Table
# =============================================================================
# SUBCOMMAND: hooks test
# =============================================================================
def run_hooks_test(repo_root: Path) -> None:
"""Run hook test suite, display per-file pass/fail table."""
pattern = str(repo_root / "src" / "aipass" / "seedgo" / "tests" / "test_hooks*.py")
test_files = sorted(_glob.glob(pattern))
if not test_files:
warning("No test_hooks*.py files found")
return
table = Table(
title="[bold cyan][HOOKS TEST][/bold cyan] Hook Test Suite",
show_header=True,
header_style="bold cyan",
border_style="dim",
)
table.add_column("Test File", style="cyan", no_wrap=True)
table.add_column("Tests", justify="right", style="yellow")
table.add_column("Passed", justify="right", style="green")
table.add_column("Failed", justify="right", style="red")
table.add_column("Duration", justify="right", style="dim")
table.add_column("Status", justify="center")
total_passed = 0
total_failed = 0
for tf in test_files:
stem = Path(tf).stem
passed, failed, duration = run_pytest_file(Path(tf), repo_root)
total_passed += passed
total_failed += failed
status = (
"[green]PASS[/green]"
if failed == 0 and passed > 0
else ("[red]FAIL[/red]" if failed > 0 else "[yellow]NONE[/yellow]")
)
table.add_row(
stem,
str(passed + failed),
str(passed),
str(failed),
f"{duration:.1f}s",
status,
)
console.print()
console.print(table)
console.print(f"\n[bold]Total:[/bold] [green]{total_passed} passed[/green] [red]{total_failed} failed[/red]\n")
json_handler.log_operation(
"hooks_test_run",
{"files": len(test_files), "total_passed": total_passed, "total_failed": total_failed},
)
# =============================================================================
# SUBCOMMAND: hooks report
# =============================================================================
def run_hooks_report(repo_root: Path, args: list) -> None:
"""Run hook execution report — delegates to hook_report.py."""
import subprocess
script = repo_root / ".claude" / "hooks" / "hook_report.py"
if not script.exists():
warning(f"hook_report.py not found at {script}")
return
cmd = ["python3", str(script)] + args
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
if result.stdout:
console.print(result.stdout.rstrip())
if result.stderr:
console.print(f"[red]{result.stderr.rstrip()}[/red]")
except subprocess.TimeoutExpired:
logger.info("hooks_ext.py: hook_report.py timed out after 30s")
warning("hook_report.py timed out after 30s")
json_handler.log_operation("hooks_report", {"args": args})
# =============================================================================
# SUBCOMMAND: hooks list — helpers
# =============================================================================
def read_settings_file(path: Path) -> dict:
"""Read and parse a settings.json file. Returns {} on failure."""
if not path.exists():
return {}
text = read_text_safe(path)
if text is None:
logger.info("hooks_ext.py: could not read %s", path)
return {}
try:
return json.loads(text)
except Exception as exc:
logger.info("hooks_ext.py: could not parse %s: %s", path, exc)
return {}
def extract_hook_script(command: str) -> str:
"""Extract the Python script filename from a hook command string."""
parts = command.split()
for part in reversed(parts):
if part.endswith(".py"):
return Path(part).name
return command[:40] if command else "(unknown)"
def _parse_version_from_lines(lines: list) -> str | None:
"""Scan lines for a Version: comment. Returns version string or None."""
for line in lines:
stripped = line.strip()
if stripped.startswith("# Version:") or stripped.startswith("Version:"):
return stripped.split(":", 1)[1].strip()
return None
def read_hook_version(command: str) -> str:
"""Extract version string from a hook file's first 20 lines."""
parts = command.split()
for part in reversed(parts):
if part.endswith(".py"):
raw_lines = read_lines_safe(Path(part), n=20)
version = _parse_version_from_lines(raw_lines)
if version is not None:
return version
return "?"
def _add_hook_defs_to_table(
table: Table,
hook_defs: list,
event_name: str,
matcher: str,
location: str,
seen: set,
) -> int:
"""Add hook definitions to table, skipping duplicates. Returns count added."""
count = 0
for hook_def in hook_defs:
if not isinstance(hook_def, dict):
continue
command = hook_def.get("command", "")
if not command:
continue
script_name = extract_hook_script(command)
dedup_key = (script_name, event_name, location)
if dedup_key in seen:
continue
seen.add(dedup_key)
version = read_hook_version(command)
script_path_parts = [p for p in command.split() if p.endswith(".py")]
exists = Path(script_path_parts[0]).exists() if script_path_parts else False
table.add_row(
script_name,
event_name,
matcher[:35] if len(matcher) > 35 else matcher,
version,
location,
"[green]✓[/green]" if exists else "[red]✗[/red]",
)
count += 1
return count
def _populate_hooks_table(table: Table, sources: list, seen: set) -> int:
"""Populate hooks list table from settings sources. Returns row count."""
total = 0
for _label, path, location in sources:
settings = read_settings_file(path)
for event_name, entries in settings.get("hooks", {}).items():
hook_entries = entries if isinstance(entries, list) else []
for entry in hook_entries:
matcher = entry.get("matcher", "*") if isinstance(entry, dict) else "*"
hook_defs = entry.get("hooks", []) if isinstance(entry, dict) else []
total += _add_hook_defs_to_table(table, hook_defs, event_name, matcher, location, seen)
return total
# =============================================================================
# SUBCOMMAND: hooks list
# =============================================================================
def cmd_hooks_list(repo_root: Path) -> None:
"""Show every wired hook from both project and user-global settings."""
global_settings_path = Path.home() / ".claude" / "settings.json"
project_settings_path = repo_root / ".claude" / "settings.json"
sources = [
("~/.claude/settings.json", global_settings_path, "global"),
(".claude/settings.json", project_settings_path, "project"),
]
table = Table(
title="[bold cyan][HOOKS LIST][/bold cyan] Wired Hooks",
show_header=True,
header_style="bold cyan",
border_style="dim",
expand=True,
)
table.add_column("Hook Script", style="cyan", no_wrap=True)
table.add_column("Event", style="yellow")
table.add_column("Matcher", style="dim")
table.add_column("Version", justify="center", style="green")
table.add_column("Source", style="dim")
table.add_column("Exists", justify="center")
seen: set = set()
total = _populate_hooks_table(table, sources, seen)
console.print()
console.print(table)
console.print(f"\n[dim]{total} hook entries across both settings files[/dim]\n")
json_handler.log_operation("hooks_list", {"total": total})
# =============================================================================
# DRONE ROUTING (helper module — delegates to hooks.py)
# =============================================================================
def print_introspection() -> None:
"""Print subcommand info for this helper module."""
console.print("[bold cyan]hooks_ext[/bold cyan] — hook test and list helper")
console.print(" [dim]Use via hooks.py: drone @seedgo hooks test | hooks list[/dim]")
def handle_command(command: str, args: list) -> bool:
"""Not a primary drone module — delegates to hooks.py for routing."""
if command != "hooks_ext":
return False
if not args:
print_introspection()
return True
if args[0] in ("--help", "-h", "help"):
print_introspection()
return True
return False
src/aipass/seedgo/apps/modules/hooks_probe.py
-169
View File
@@ -1,169 +0,0 @@
# =================== AIPass ====================
# Name: hooks_probe.py
# Description: Hook Probe Matrix Subcommand (split from hooks.py)
# Version: 1.0.0
# Created: 2026-05-16
# Modified: 2026-05-16
# =============================================
"""
Hook Probe Matrix — split from hooks.py to keep modules under 600 lines.
hooks probe --matrix Full event matrix + markdown report
"""
from collections import Counter, defaultdict
from pathlib import Path
from typing import Callable, List
from aipass.prax import logger
from aipass.cli import console
from aipass.seedgo.apps.handlers.file import write_text_safe
from aipass.seedgo.apps.handlers.json import json_handler
from rich.panel import Panel
from rich.table import Table
_PROBE_ENABLE_NOTE = (
"Add probe hook commands to .claude/settings.json (inside the hooks object).\n"
"See .claude/hooks/probes/README.md for per-event snippets."
)
def print_introspection() -> None:
"""Display module info."""
console.print("[bold cyan]hooks_probe[/bold cyan] — matrix subcommand helper for hooks.py")
console.print(" Not directly routable. Called via [green]drone @seedgo hooks probe --matrix[/green]")
def handle_command(command: str, args: list) -> bool:
"""Not a routed module — always returns False."""
if not args:
print_introspection()
return False
if args[0] in ("--help", "-h", "help"):
print_introspection()
return False
logger.info("hooks_probe: not a routed module, command=%s args=%s", command, args)
return False
def _build_matrix_rows(entries: list) -> tuple[list, dict]:
"""Group entries by event and build matrix row dicts. Returns (rows, groups)."""
groups: dict = defaultdict(list)
for e in entries:
groups[e.get("event", "unknown")].append(e)
matrix_rows = []
for event_name in sorted(groups.keys()):
evs = groups[event_name]
count = len(evs)
pd_true = sum(1 for e in evs if e.get("env_has_claude_project_dir"))
pd_false = count - pd_true
ah_true = sum(1 for e in evs if e.get("env_has_aipass_home"))
ah_false = count - ah_true
agents = len({e.get("agent_id", "unknown") for e in evs})
matrix_rows.append(
{
"event": event_name,
"count": count,
"project_dir_true": pd_true,
"project_dir_false": pd_false,
"aipass_home_true": ah_true,
"aipass_home_false": ah_false,
"unique_agents": agents,
}
)
return matrix_rows, groups
def _write_matrix_report(report_path: Path, matrix_rows: list, entries: list) -> None:
"""Write markdown matrix report to disk."""
lines = [
"# Hook Probe Matrix — Q12 Findings 2026-04-20",
"",
"Generated by `drone @seedgo hooks probe --matrix`.",
"",
f"Total entries: {len(entries)}",
"",
"## Event Matrix",
"",
"| Event | Count | ProjectDir T/F | AIPassHome T/F | Unique Agents |",
"|-------|-------|----------------|----------------|---------------|",
]
for row in matrix_rows:
lines.append(
f"| {row['event']} | {row['count']} "
f"| {row['project_dir_true']}/{row['project_dir_false']} "
f"| {row['aipass_home_true']}/{row['aipass_home_false']} "
f"| {row['unique_agents']} |"
)
lines += [
"",
"## Notes",
"",
"- `ProjectDir T/F`: entries where `CLAUDE_PROJECT_DIR` env var was set (T) vs unset (F)",
"- `AIPassHome T/F`: entries where `AIPASS_HOME` env var was set (T) vs unset (F)",
"- Unique Agents: distinct `CLAUDE_CODE_SESSION_ID` values seen for this event",
"",
"## Raw entry count by event",
"",
]
event_counts = Counter(e.get("event", "unknown") for e in entries)
for ev, cnt in sorted(event_counts.items()):
lines.append(f"- {ev}: {cnt}")
lines.append("")
write_text_safe(report_path, "\n".join(lines))
def run_probe_matrix(
read_entries: Callable[[], List[dict]],
probes_dir: Callable[[], Path],
) -> None:
"""Group entries by event, show matrix, write markdown report."""
entries = read_entries()
if not entries:
console.print(
Panel(
f"[yellow]No probe data yet — enable probes first.[/yellow]\n\n{_PROBE_ENABLE_NOTE}",
title="[bold cyan][PROBE] Matrix[/bold cyan]",
border_style="cyan",
)
)
return
matrix_rows, groups = _build_matrix_rows(entries)
table = Table(
title="[bold cyan][PROBE][/bold cyan] Event Matrix",
show_header=True,
header_style="bold cyan",
border_style="dim",
)
table.add_column("Event", style="cyan", no_wrap=True)
table.add_column("Count", justify="right", style="yellow")
table.add_column("ProjectDir T/F", justify="center")
table.add_column("AIPassHome T/F", justify="center")
table.add_column("Unique Agents", justify="right")
for row in matrix_rows:
table.add_row(
row["event"],
str(row["count"]),
f"[green]{row['project_dir_true']}[/green]/[red]{row['project_dir_false']}[/red]",
f"[green]{row['aipass_home_true']}[/green]/[red]{row['aipass_home_false']}[/red]",
str(row["unique_agents"]),
)
console.print()
console.print(table)
console.print()
report_path = probes_dir() / "Q12_findings_2026-04-20.md"
_write_matrix_report(report_path, matrix_rows, entries)
console.print(f"[green]Report written:[/green] {report_path}\n")
json_handler.log_operation("hooks_probe_matrix", {"events": len(groups), "total_entries": len(entries)})
src/aipass/seedgo/tests/test_handler_functions.py
-37
View File
@@ -93,7 +93,6 @@ def _mock_infrastructure(monkeypatch):
"aipass.seedgo.apps.handlers.json.json_handler",
"aipass.seedgo.apps.handlers.readme.readme_ops",
"aipass.seedgo.apps.handlers.readme.readme_generator",
"aipass.seedgo.apps.modules.hooks_ext",
]:
monkeypatch.delitem(sys.modules, mod_name, raising=False)
@@ -569,39 +568,3 @@ def test_update_readme_auto_sections_missing_markers(tmp_path):
result = update_readme_auto_sections(str(branch_dir))
# Some sections should report missing markers
assert len(result["missing_markers"]) > 0 or len(result["updated"]) == 0
# ===========================================================================
# 10. hooks_ext -- run_hooks_test
# ===========================================================================
def test_run_hooks_test_no_files(tmp_path):
"""run_hooks_test with no test files shows warning."""
from aipass.seedgo.apps.modules.hooks_ext import run_hooks_test
# repo_root with no test_hooks*.py files
run_hooks_test(tmp_path)
# Should call warning() -- verify through mock
import sys
cli_modules = sys.modules["aipass.cli.apps.modules"]
cli_modules.warning.assert_called()
def test_run_hooks_test_with_files(tmp_path):
"""run_hooks_test with test files builds and prints table."""
import sys
# Create the expected directory structure
test_dir = tmp_path / "src" / "aipass" / "seedgo" / "tests"
test_dir.mkdir(parents=True)
(test_dir / "test_hooks_basic.py").write_text("pass", encoding="utf-8")
from aipass.seedgo.apps.modules.hooks_ext import run_hooks_test
run_hooks_test(tmp_path)
# Verify console.print was called (table output)
cli_mod = sys.modules["aipass.cli"]
assert cli_mod.console.print.called
src/aipass/seedgo/tests/test_hook_bridge.py
-311
View File
@@ -1,311 +0,0 @@
# =================== AIPass ====================
# Name: test_hook_bridge.py
# Description: Tests for hook bridge installer (DPLAN-0141 Phase 2)
# Version: 1.0.0
# Created: 2026-04-22
# Modified: 2026-04-22
# =============================================
"""Tests for hook bridge installer — bridge_handler.py + hook_bridge.py."""
import json
from unittest.mock import patch
# ---------------------------------------------------------------------------
# bridge_handler tests
# ---------------------------------------------------------------------------
def test_resolve_aipass_home_from_env(tmp_path):
"""resolve_aipass_home returns path from AIPASS_HOME env var."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import resolve_aipass_home
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
result = resolve_aipass_home()
assert result == tmp_path
def test_resolve_aipass_home_returns_none_when_all_fail():
"""resolve_aipass_home returns None when env var is bad and importlib fallback disabled."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import resolve_aipass_home
with (
patch.dict("os.environ", {"AIPASS_HOME": "/nonexistent/path/xyz"}),
patch("importlib.util.find_spec", return_value=None),
):
result = resolve_aipass_home()
assert result is None
def test_is_aipass_hook_entry_detects_marker():
"""is_aipass_hook_entry returns True for entries with _aipass marker."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import is_aipass_hook_entry
entry = {"_aipass": True, "hooks": [{"type": "command", "command": "echo test"}]}
assert is_aipass_hook_entry(entry) is True
def test_is_aipass_hook_entry_detects_pattern():
"""is_aipass_hook_entry returns True for entries with AIPass path patterns."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import is_aipass_hook_entry
entry = {"hooks": [{"type": "command", "command": "python3 /home/user/AIPass/.claude/hooks/foo.py"}]}
assert is_aipass_hook_entry(entry) is True
def test_is_aipass_hook_entry_detects_env_var():
"""is_aipass_hook_entry returns True for entries using $AIPASS_HOME."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import is_aipass_hook_entry
entry = {"hooks": [{"type": "command", "command": "python3 $AIPASS_HOME/.claude/hooks/bar.py"}]}
assert is_aipass_hook_entry(entry) is True
def test_is_aipass_hook_entry_ignores_non_aipass():
"""is_aipass_hook_entry returns False for unrelated entries."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import is_aipass_hook_entry
entry = {"hooks": [{"type": "command", "command": "echo hello"}]}
assert is_aipass_hook_entry(entry) is False
def test_count_aipass_hooks_empty():
"""count_aipass_hooks returns 0 for empty settings."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import count_aipass_hooks
assert count_aipass_hooks({}) == 0
assert count_aipass_hooks({"hooks": {}}) == 0
def test_count_aipass_hooks_counts_marked():
"""count_aipass_hooks counts entries with _aipass marker."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import count_aipass_hooks
settings = {
"hooks": {
"Stop": [
{"_aipass": True, "hooks": [{"type": "command", "command": "echo"}]},
{"hooks": [{"type": "command", "command": "other"}]},
]
}
}
assert count_aipass_hooks(settings) == 1
def test_install_hooks_adds_entries():
"""install_hooks adds manifest entries to empty settings."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import (
AIPASS_HOOK_MANIFEST,
install_hooks,
)
settings = {"hooks": {}}
settings, added = install_hooks(settings)
assert added > 0
total_manifest = sum(len(v) for v in AIPASS_HOOK_MANIFEST.values())
assert added == total_manifest
def test_install_hooks_idempotent():
"""install_hooks does not duplicate entries on second call."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import install_hooks
settings = {"hooks": {}}
settings, first_added = install_hooks(settings)
settings, second_added = install_hooks(settings)
assert first_added > 0
assert second_added == 0
def test_install_hooks_preserves_existing():
"""install_hooks preserves non-AIPass hooks in same event."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import install_hooks
user_hook = {"hooks": [{"type": "command", "command": "echo user-hook"}]}
settings = {"hooks": {"Stop": [user_hook]}}
settings, added = install_hooks(settings)
stop_entries = settings["hooks"]["Stop"]
assert user_hook in stop_entries
assert added > 0
def test_install_hooks_skips_event_with_existing_aipass():
"""install_hooks skips events that already contain AIPass entries."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import install_hooks
existing = {
"_aipass": True,
"hooks": [{"type": "command", "command": "python3 $AIPASS_HOME/.claude/hooks/stop_sound.py"}],
}
settings = {"hooks": {"Stop": [existing]}}
settings, added = install_hooks(settings)
assert len(settings["hooks"]["Stop"]) == 1
def test_uninstall_hooks_removes_marked():
"""uninstall_hooks removes entries with _aipass marker."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import uninstall_hooks
settings = {
"hooks": {
"Stop": [
{"_aipass": True, "hooks": [{"type": "command", "command": "echo aipass"}]},
{"hooks": [{"type": "command", "command": "echo user"}]},
]
}
}
settings, removed = uninstall_hooks(settings)
assert removed == 1
assert len(settings["hooks"]["Stop"]) == 1
assert settings["hooks"]["Stop"][0]["hooks"][0]["command"] == "echo user"
def test_uninstall_hooks_removes_empty_events():
"""uninstall_hooks deletes event keys that become empty."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import uninstall_hooks
settings = {
"hooks": {
"Stop": [
{"_aipass": True, "hooks": [{"type": "command", "command": "echo"}]},
]
}
}
settings, removed = uninstall_hooks(settings)
assert removed == 1
assert "Stop" not in settings["hooks"]
def test_uninstall_hooks_preserves_non_aipass():
"""uninstall_hooks never touches non-AIPass entries."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import uninstall_hooks
user_hook = {"hooks": [{"type": "command", "command": "echo user"}]}
settings = {"hooks": {"Stop": [user_hook]}}
settings, removed = uninstall_hooks(settings)
assert removed == 0
assert settings["hooks"]["Stop"] == [user_hook]
def test_ensure_aipass_env_adds_when_missing():
"""ensure_aipass_env adds AIPASS_HOME to env section."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import ensure_aipass_env
settings = {}
added = ensure_aipass_env(settings, "/home/user/AIPass")
assert added is True
assert settings["env"]["AIPASS_HOME"] == "/home/user/AIPass"
def test_ensure_aipass_env_skips_when_present():
"""ensure_aipass_env does not overwrite existing AIPASS_HOME."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import ensure_aipass_env
settings = {"env": {"AIPASS_HOME": "/existing/path"}}
added = ensure_aipass_env(settings, "/new/path")
assert added is False
assert settings["env"]["AIPASS_HOME"] == "/existing/path"
def test_read_settings_returns_empty_for_missing(tmp_path):
"""read_settings returns {} when file does not exist."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import read_settings
result = read_settings(tmp_path / "missing.json")
assert result == {}
def test_read_settings_parses_json(tmp_path):
"""read_settings parses valid JSON."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import read_settings
f = tmp_path / "settings.json"
f.write_text(json.dumps({"hooks": {}}), encoding="utf-8")
result = read_settings(f)
assert result == {"hooks": {}}
def test_write_settings_creates_backup(tmp_path):
"""write_settings creates .json.bak before overwriting."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import write_settings
f = tmp_path / "settings.json"
f.write_text('{"old": true}', encoding="utf-8")
write_settings(f, {"new": True})
bak = tmp_path / "settings.json.bak"
assert bak.exists()
assert json.loads(bak.read_text()) == {"old": True}
assert json.loads(f.read_text()) == {"new": True}
def test_write_settings_creates_parent_dirs(tmp_path):
"""write_settings creates parent directories if needed."""
from aipass.seedgo.apps.handlers.hooks.bridge_handler import write_settings
f = tmp_path / "sub" / "dir" / "settings.json"
result = write_settings(f, {"test": True})
assert result is True
assert f.exists()
# ---------------------------------------------------------------------------
# hook_bridge module tests
# ---------------------------------------------------------------------------
def test_handle_command_rejects_non_bridge():
"""handle_command returns False for non-bridge commands."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
assert handle_command("audit", ["install"]) is False
assert handle_command("hooks", ["install"]) is False
def test_handle_command_accepts_bridge():
"""handle_command returns True for bridge command."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
assert handle_command("bridge", []) is True
def test_handle_command_accepts_help():
"""handle_command returns True for bridge --help."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
assert handle_command("bridge", ["--help"]) is True
assert handle_command("bridge", ["-h"]) is True
assert handle_command("bridge", ["help"]) is True
def test_handle_command_routes_install():
"""handle_command routes 'install' subcommand."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
with patch("aipass.seedgo.apps.modules.hook_bridge._run_install"):
result = handle_command("bridge", ["install"])
assert result is True
def test_handle_command_routes_uninstall():
"""handle_command routes 'uninstall' subcommand."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
with patch("aipass.seedgo.apps.modules.hook_bridge._run_uninstall"):
result = handle_command("bridge", ["uninstall"])
assert result is True
def test_handle_command_routes_status():
"""handle_command routes 'status' subcommand."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
with patch("aipass.seedgo.apps.modules.hook_bridge._run_status"):
result = handle_command("bridge", ["status"])
assert result is True
def test_handle_command_unknown_subcommand():
"""handle_command returns True for unknown subcommand (shows help)."""
from aipass.seedgo.apps.modules.hook_bridge import handle_command
assert handle_command("bridge", ["bogus"]) is True
src/aipass/seedgo/tests/test_hooks_probe.py
-515
View File
@@ -1,515 +0,0 @@
"""Tests for the hook probe scripts and hooks module.
# =================== META ====================
# Name: test_hooks_probe.py
# Description: Tests for hook probe scripts and hooks seedgo module
# Version: 1.0.0
# Created: 2026-04-20
# Modified: 2026-04-20
# =============================================
"""
import importlib.util
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# Helpers — probe script loader
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
"""Walk up from this file to find the git repo root."""
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4] # fallback
_PROBES_DIR = _find_repo_root() / ".claude" / "hooks" / "probes"
_PROBE_SCRIPTS = [
"probe_pre_tool_use.py",
"probe_post_tool_use.py",
"probe_user_prompt_submit.py",
"probe_subagent_stop.py",
"probe_pre_compact.py",
"probe_stop.py",
"probe_notification.py",
]
_EVENT_NAMES = [
"PreToolUse",
"PostToolUse",
"UserPromptSubmit",
"SubagentStop",
"PreCompact",
"Stop",
"Notification",
]
def _load_probe(script_name: str):
"""Import a probe script by filename via importlib (outside package)."""
path = _PROBES_DIR / script_name
if not path.exists():
pytest.skip(f"Probe script not found: {path}")
spec = importlib.util.spec_from_file_location(path.stem, path)
assert spec is not None and spec.loader is not None
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore[union-attr]
return module
# ---------------------------------------------------------------------------
# Fixtures — infrastructure mocks for hooks module
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_hooks_infrastructure(monkeypatch):
"""Mock aipass infrastructure imports for hooks module tests."""
mock_logger = MagicMock()
mock_console = MagicMock()
mock_warning = MagicMock()
mock_json_handler = MagicMock()
# -- prax ---------------------------------------------------------------
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
# -- cli ----------------------------------------------------------------
cli_mod = MagicMock()
cli_mod.console = mock_console
monkeypatch.setitem(sys.modules, "aipass.cli", cli_mod)
cli_apps = MagicMock()
monkeypatch.setitem(sys.modules, "aipass.cli.apps", cli_apps)
cli_modules = MagicMock()
cli_modules.warning = mock_warning
monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules)
# -- seedgo json handler ------------------------------------------------
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
# Force re-import of hooks module
monkeypatch.delitem(sys.modules, "aipass.seedgo.apps.modules.hooks", raising=False)
# ---------------------------------------------------------------------------
# Tests — probe scripts: happy-path stdin
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_happy_path(script_name, event_name, tmp_path, monkeypatch):
"""Each probe writes a correctly-shaped entry to last_ping.jsonl."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
payload = json.dumps(
{
"tool_name": "Read",
"cwd": "/tmp/test_cwd",
"session_id": "test-session-123",
}
)
monkeypatch.setattr("sys.stdin", io.StringIO(payload))
probe.main()
assert log_file.exists(), f"{script_name} did not create log file"
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
assert len(lines) == 1, f"Expected 1 entry, got {len(lines)}"
entry = json.loads(lines[0])
assert entry["event"] == event_name
assert entry["tool"] == "Read"
assert entry["cwd"] == "/tmp/test_cwd"
assert "timestamp" in entry
assert "script_elapsed_ms" in entry
assert isinstance(entry["script_elapsed_ms"], float)
assert "agent_id" in entry
assert "cli_version" in entry
assert "env_has_claude_project_dir" in entry
assert "env_has_aipass_home" in entry
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_appends_not_overwrites(script_name, event_name, tmp_path, monkeypatch):
"""Each probe appends to an existing log file instead of overwriting."""
import io
log_file = tmp_path / "last_ping.jsonl"
existing = json.dumps(
{
"event": "existing",
"tool": "",
"cwd": "/",
"agent_id": "x",
"timestamp": "2026-01-01T00:00:00Z",
"script_elapsed_ms": 0.1,
"cli_version": "0",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
}
)
log_file.write_text(existing + "\n", encoding="utf-8")
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
assert len(lines) == 2, "Probe should append, not overwrite"
assert json.loads(lines[0])["event"] == "existing"
assert json.loads(lines[1])["event"] == event_name
# ---------------------------------------------------------------------------
# Tests — probe scripts: malformed stdin
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_malformed_stdin_exits_cleanly(script_name, event_name, tmp_path, monkeypatch):
"""Each probe handles malformed stdin without raising an exception."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO("not json at all !!!"))
# main() returns normally on parse failure — no exception should propagate.
# sys.exit(0) is only called from the __main__ block, not from main() itself.
probe.main() # must not raise
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_empty_stdin_exits_cleanly(script_name, event_name, tmp_path, monkeypatch):
"""Each probe handles empty stdin without raising an exception."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setattr("sys.stdin", io.StringIO(""))
probe.main() # must not raise
# ---------------------------------------------------------------------------
# Tests — probe env var extraction
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_reads_session_id_from_env(script_name, event_name, tmp_path, monkeypatch):
"""Probe picks up CLAUDE_CODE_SESSION_ID from environment."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setenv("CLAUDE_CODE_SESSION_ID", "sess-env-test-999")
monkeypatch.delenv("CLAUDE_SESSION_ID", raising=False)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "sess-env-test-999"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_falls_back_to_claude_session_id(script_name, event_name, tmp_path, monkeypatch):
"""Probe falls back to CLAUDE_SESSION_ID when CLAUDE_CODE_SESSION_ID is absent."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
monkeypatch.setenv("CLAUDE_SESSION_ID", "fallback-session-42")
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "fallback-session-42"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_unknown_agent_id_when_no_env(script_name, event_name, tmp_path, monkeypatch):
"""Probe uses 'unknown' when no session env vars are set."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.delenv("CLAUDE_CODE_SESSION_ID", raising=False)
monkeypatch.delenv("CLAUDE_SESSION_ID", raising=False)
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["agent_id"] == "unknown"
@pytest.mark.parametrize("script_name,event_name", list(zip(_PROBE_SCRIPTS, _EVENT_NAMES)))
def test_probe_env_has_aipass_home(script_name, event_name, tmp_path, monkeypatch):
"""Probe records env_has_aipass_home correctly."""
import io
log_file = tmp_path / "last_ping.jsonl"
probe = _load_probe(script_name)
monkeypatch.setattr(probe, "_LOG_FILE", log_file)
monkeypatch.setenv("AIPASS_HOME", "/home/user/Projects/AIPass")
monkeypatch.setattr("sys.stdin", io.StringIO("{}"))
probe.main()
lines = [ln for ln in log_file.read_text(encoding="utf-8").splitlines() if ln.strip()]
entry = json.loads(lines[0])
assert entry["env_has_aipass_home"] is True
# ---------------------------------------------------------------------------
# Tests — hooks module: handle_command routing
# ---------------------------------------------------------------------------
def test_handle_command_wrong_command_returns_false():
"""handle_command returns False for unrecognised commands."""
from aipass.seedgo.apps.modules.hooks import handle_command
assert handle_command("wrong_command", []) is False
def test_handle_command_no_args_calls_introspection():
"""No args triggers introspection (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", [])
assert result is True
def test_handle_command_help_flag():
"""--help flag is handled without error."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["--help"])
assert result is True
def test_handle_command_probe_no_flags():
"""hooks probe with no extra flags calls display (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["probe"])
assert result is True
def test_handle_command_unknown_subcommand():
"""Unknown subcommand falls back to introspection (returns True)."""
from aipass.seedgo.apps.modules.hooks import handle_command
result = handle_command("hooks", ["nonexistent_subcommand"])
assert result is True
# ---------------------------------------------------------------------------
# Tests — hooks module: probe display with mock data
# ---------------------------------------------------------------------------
def test_probe_display_renders_table(tmp_path):
"""_cmd_probe_display renders a Rich table without error given sample data."""
entries = [
{
"event": "PreToolUse",
"tool": "Bash",
"cwd": "/tmp/test",
"agent_id": "sess-abc",
"timestamp": "2026-04-20T12:00:00.000Z",
"script_elapsed_ms": 2.1,
"cli_version": "1.0.0",
"env_has_claude_project_dir": True,
"env_has_aipass_home": False,
},
{
"event": "PostToolUse",
"tool": "Read",
"cwd": "/tmp/test",
"agent_id": "sess-abc",
"timestamp": "2026-04-20T12:00:01.000Z",
"script_elapsed_ms": 1.5,
"cli_version": "1.0.0",
"env_has_claude_project_dir": True,
"env_has_aipass_home": True,
},
]
log_file = tmp_path / "last_ping.jsonl"
with open(log_file, "w", encoding="utf-8") as fh:
for e in entries:
fh.write(json.dumps(e) + "\n")
from aipass.seedgo.apps.modules.hooks import _cmd_probe_display
_cmd_probe_display(log_path=log_file)
def test_probe_display_empty_log(tmp_path):
"""_cmd_probe_display handles missing log file gracefully."""
from aipass.seedgo.apps.modules.hooks import _cmd_probe_display
missing = tmp_path / "no_such_file.jsonl"
_cmd_probe_display(log_path=missing)
def test_read_entries_skips_bad_lines(tmp_path):
"""_read_entries skips malformed lines and returns valid ones."""
log_file = tmp_path / "last_ping.jsonl"
good = json.dumps(
{
"event": "Stop",
"tool": "",
"cwd": "/",
"agent_id": "x",
"timestamp": "2026-04-20T00:00:00Z",
"script_elapsed_ms": 1.0,
"cli_version": "1",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
}
)
log_file.write_text(f"not json\n{good}\nalso not json\n", encoding="utf-8")
from aipass.seedgo.apps.modules.hooks import _read_entries
entries = _read_entries(log_path=log_file)
assert len(entries) == 1
assert entries[0]["event"] == "Stop"
def test_truncate_helper():
"""_truncate shortens long strings correctly."""
from aipass.seedgo.apps.modules.hooks import _truncate
assert _truncate("short", 20) == "short"
assert _truncate("a" * 30, 10) == "a" * 7 + "..."
assert len(_truncate("x" * 50, 15)) == 15
def test_entry_ts_parses_z_suffix():
"""_entry_ts handles ISO 8601 Z-suffix timestamps."""
from aipass.seedgo.apps.modules.hooks import _entry_ts
ts = _entry_ts({"timestamp": "2026-04-20T12:00:00.000000Z"})
assert ts > 0
def test_entry_ts_returns_zero_on_bad_input():
"""_entry_ts returns 0.0 for unparseable timestamps."""
from aipass.seedgo.apps.modules.hooks import _entry_ts
assert _entry_ts({"timestamp": "not-a-date"}) == 0.0
assert _entry_ts({}) == 0.0
# ---------------------------------------------------------------------------
# Tests — matrix builder
# ---------------------------------------------------------------------------
def test_build_matrix_rows_groups_correctly():
"""_build_matrix_rows groups entries by event and counts correctly."""
entries = [
{
"event": "Stop",
"env_has_claude_project_dir": True,
"env_has_aipass_home": False,
"agent_id": "a",
},
{
"event": "Stop",
"env_has_claude_project_dir": False,
"env_has_aipass_home": False,
"agent_id": "b",
},
{
"event": "PreToolUse",
"env_has_claude_project_dir": True,
"env_has_aipass_home": True,
"agent_id": "a",
},
]
from aipass.seedgo.apps.modules.hooks_probe import _build_matrix_rows
rows, groups = _build_matrix_rows(entries)
assert len(rows) == 2
stop_row = next(r for r in rows if r["event"] == "Stop")
assert stop_row["count"] == 2
assert stop_row["project_dir_true"] == 1
assert stop_row["project_dir_false"] == 1
assert stop_row["unique_agents"] == 2
def test_probe_matrix_writes_report(tmp_path):
"""_cmd_probe_matrix creates the markdown report given sample data."""
from aipass.seedgo.apps.modules.hooks_probe import _write_matrix_report
entries = [
{
"event": "Stop",
"tool": "",
"cwd": "/tmp",
"agent_id": "sess-1",
"timestamp": "2026-04-20T10:00:00Z",
"script_elapsed_ms": 1.2,
"cli_version": "1.0",
"env_has_claude_project_dir": False,
"env_has_aipass_home": True,
}
]
matrix_rows = [
{
"event": "Stop",
"count": 1,
"project_dir_true": 0,
"project_dir_false": 1,
"aipass_home_true": 1,
"aipass_home_false": 0,
"unique_agents": 1,
}
]
report_path = tmp_path / "Q12_findings_2026-04-20.md"
_write_matrix_report(report_path, matrix_rows, entries)
assert report_path.exists()
content = report_path.read_text(encoding="utf-8")
assert "Stop" in content
assert "Q12 Findings" in content
src/aipass/seedgo/tests/test_hooks_track_a.py
-382
View File
@@ -1,382 +0,0 @@
"""Tests for Track A hook scripts — auto_fix_diagnostics, pre_edit_gate, subagent_stop_gate.
# =================== META ====================
# Name: test_hooks_track_a.py
# Description: Track A hook tests — DPLAN-0139 coverage for auto_fix, pre_edit_gate, subagent_stop
# Version: 1.0.0
# Created: 2026-04-20
# Modified: 2026-04-20
# =============================================
"""
import importlib.util
import io
import json
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# Helpers — hook loader
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
"""Walk up from this file to find the git repo root."""
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4] # fallback
HOOKS_DIR = _find_repo_root() / ".claude" / "hooks"
def _load_hook(name: str):
"""Import a hook script by filename via importlib (outside package)."""
path = HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
# ---------------------------------------------------------------------------
# auto_fix_diagnostics.py tests
# ---------------------------------------------------------------------------
def test_auto_fix_skips_non_edit_tool(capsys):
"""stdin with tool_name=Read → no output (not an edit tool)."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": "/tmp/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_auto_fix_skips_non_py_file(capsys):
"""stdin with tool_name=Edit, file_path ending .md → no output (skipped extension)."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/foo.md"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_auto_fix_does_not_crash_empty_stdin():
"""Empty/malformed JSON on stdin → no exception raised."""
mod = _load_hook("auto_fix_diagnostics.py")
with patch("sys.stdin", io.StringIO("")):
mod.main() # must not raise
def test_auto_fix_does_not_crash_missing_fields():
"""Valid JSON but no tool_name → no exception raised."""
mod = _load_hook("auto_fix_diagnostics.py")
payload = json.dumps({"some_other_key": "value"})
with patch("sys.stdin", io.StringIO(payload)):
mod.main() # must not raise
def test_auto_fix_label_is_auto_fix(tmp_path, capsys):
"""When run_python_checks returns an error, output contains [AUTO-FIX] label."""
mod = _load_hook("auto_fix_diagnostics.py")
fake_py = tmp_path / "fake.py"
fake_py.write_text("x = 1\n", encoding="utf-8")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(fake_py)}})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "run_python_checks", return_value=["LINT: E501 line too long"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(mod, "run_pyright_check", return_value=[]),
patch.object(mod, "run_ruff_lint_structured", return_value=[]),
patch.object(mod, "save_diagnostics_state"),
):
mod.main()
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
context = output["hookSpecificOutput"]["additionalContext"]
assert "[AUTO-FIX]" in context
assert "[AUTO-FIX]" in output["systemMessage"]
def test_run_pyright_check_returns_empty_on_file_not_found():
"""run_pyright_check returns [] when pyright binary is missing."""
mod = _load_hook("auto_fix_diagnostics.py")
with patch("subprocess.run", side_effect=FileNotFoundError("pyright not found")):
result = mod.run_pyright_check("/tmp/some_file.py")
assert result == []
def test_run_pyright_check_returns_empty_on_timeout():
"""run_pyright_check returns [] on timeout."""
import subprocess
mod = _load_hook("auto_fix_diagnostics.py")
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("pyright", 15)):
result = mod.run_pyright_check("/tmp/some_file.py")
assert result == []
def test_run_pyright_check_skips_hook_files():
"""run_pyright_check returns [] for hook files without calling subprocess."""
mod = _load_hook("auto_fix_diagnostics.py")
mock_run = MagicMock()
with patch("subprocess.run", mock_run):
result = mod.run_pyright_check("/home/user/.claude/hooks/some_hook.py")
mock_run.assert_not_called()
assert result == []
# ---------------------------------------------------------------------------
# pre_edit_gate.py tests
# (pre_edit_gate lives in ~/.claude/hooks — load from global location)
# ---------------------------------------------------------------------------
GLOBAL_HOOKS_DIR = Path.home() / ".claude" / "hooks"
def _load_global_hook(name: str):
"""Import a hook script from the global ~/.claude/hooks/ directory."""
path = GLOBAL_HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Global hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
def test_gate_allows_no_state_file(tmp_path, capsys, monkeypatch):
"""No state file present → no output (allow)."""
mod = _load_global_hook("pre_edit_gate.py")
monkeypatch.setattr(mod, "STATE_FILE", tmp_path / "no_such_state.json")
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/seedgo/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_same_file(tmp_path, capsys, monkeypatch):
"""State has error for file A, editing file A → allow (no block output)."""
mod = _load_global_hook("pre_edit_gate.py")
errored_file = tmp_path / "foo.py"
errored_file.write_text("x = 1\n", encoding="utf-8")
state = {"file": str(errored_file), "errors": [{"line": 1, "message": "some error"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": str(errored_file)}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_blocks_different_file_same_branch(tmp_path, capsys, monkeypatch):
"""State has error for seedgo/foo.py, editing seedgo/bar.py → block + sys.exit(2)."""
mod = _load_global_hook("pre_edit_gate.py")
# Use real AIPass src paths so _get_branch works
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "seedgo" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 5, "message": "type error here"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
assert output["decision"] == "block"
def test_gate_allows_different_branch(tmp_path, capsys, monkeypatch):
"""State has error for flow/foo.py, editing seedgo/bar.py → allow (different branches)."""
mod = _load_global_hook("pre_edit_gate.py")
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "flow" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 1, "message": "error in flow"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Edit", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_does_not_crash_malformed_stdin(monkeypatch, tmp_path):
"""Bad JSON on stdin → no exception raised."""
mod = _load_global_hook("pre_edit_gate.py")
monkeypatch.setattr(mod, "STATE_FILE", tmp_path / "no_state.json")
with patch("sys.stdin", io.StringIO("not json!!!!")):
mod.main() # must not raise
def test_gate_allows_non_edit_tool(tmp_path, capsys, monkeypatch):
"""Non-edit tool_name → no output regardless of state."""
mod = _load_global_hook("pre_edit_gate.py")
aipass_src = Path("/home/patrick/Projects/AIPass/src/aipass")
errored_file = str(aipass_src / "seedgo" / "foo.py")
edit_target = str(aipass_src / "seedgo" / "bar.py")
state = {"file": errored_file, "errors": [{"line": 1, "message": "error"}]}
state_file = tmp_path / "state.json"
state_file.write_text(json.dumps(state), encoding="utf-8")
monkeypatch.setattr(mod, "STATE_FILE", state_file)
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": edit_target}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
# ---------------------------------------------------------------------------
# subagent_stop_gate.py tests
# ---------------------------------------------------------------------------
def test_subagent_gate_no_crash_empty_stdin():
"""Bad JSON on stdin → no exception raised."""
mod = _load_hook("subagent_stop_gate.py")
with patch("sys.stdin", io.StringIO("bad json")):
mod.main() # must not raise
def test_subagent_gate_no_crash_no_files(capsys):
"""Valid stdin but no modified files → no exception, no block output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=[]),
):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_subagent_gate_no_block_when_no_violations(capsys):
"""Modified files present but no seedgo violations → no block output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(mod, "check_hook_readme_accountability", return_value=None),
):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_subagent_gate_blocks_on_violations(capsys):
"""Modified files with seedgo violations → block decision in output."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=["open() without encoding='utf-8'"]),
patch.object(mod, "check_hook_readme_accountability", return_value=None),
):
mod.main()
captured = capsys.readouterr()
assert captured.out.strip() != ""
output = json.loads(captured.out)
assert output["decision"] == "block"
def test_subagent_gate_readme_reminder_soft(capsys):
"""Hook files changed without README update → allow with reminder."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=[]),
patch.object(
mod,
"check_hook_readme_accountability",
return_value="Hook files were modified but .claude/hooks/README.md was not updated.",
),
):
mod.main()
captured = capsys.readouterr()
output = json.loads(captured.out)
assert output["decision"] == "allow"
assert "README" in output["reason"]
def test_subagent_gate_readme_reminder_appended_to_block(capsys):
"""Hook files changed + violations → block includes README reminder."""
mod = _load_hook("subagent_stop_gate.py")
payload = json.dumps({"stop_hook_active": True})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "get_modified_py_files", return_value=["/tmp/foo.py"]),
patch.object(mod, "run_seedgo_checklist", return_value=["missing docstring"]),
patch.object(
mod,
"check_hook_readme_accountability",
return_value="Hook files were modified but .claude/hooks/README.md was not updated.",
),
):
mod.main()
captured = capsys.readouterr()
output = json.loads(captured.out)
assert output["decision"] == "block"
assert "README" in output["reason"]
assert "missing docstring" in output["reason"]
src/aipass/seedgo/tests/test_hooks_track_b.py
-171
View File
@@ -1,171 +0,0 @@
"""Tests for Track B hooks subcommands — hooks test and hooks list.
# =================== META ====================
# Name: test_hooks_track_b.py
# Description: Tests for new hooks test and hooks list subcommands (DPLAN-0139 Track B)
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# Infrastructure mocks (same pattern as test_hooks_probe.py)
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_hooks_infrastructure(monkeypatch):
"""Mock aipass infrastructure imports for hooks module tests."""
mock_logger = MagicMock()
mock_console = MagicMock()
mock_warning = MagicMock()
mock_json_handler = MagicMock()
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
cli_mod = MagicMock()
cli_mod.console = mock_console
monkeypatch.setitem(sys.modules, "aipass.cli", cli_mod)
cli_apps = MagicMock()
monkeypatch.setitem(sys.modules, "aipass.cli.apps", cli_apps)
cli_modules = MagicMock()
cli_modules.warning = mock_warning
monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", cli_modules)
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
file_pkg = MagicMock()
file_pkg.read_lines_safe = MagicMock(return_value=[])
file_pkg.read_text_safe = MagicMock(return_value=None)
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.file", file_pkg)
# Force re-import of both modules
monkeypatch.delitem(sys.modules, "aipass.seedgo.apps.modules.hooks_ext", raising=False)
monkeypatch.delitem(sys.modules, "aipass.seedgo.apps.modules.hooks", raising=False)
# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _find_repo_root() -> Path:
current = Path(__file__).resolve().parent
for parent in (current, *current.parents):
if (parent / ".git").exists():
return parent
return Path(__file__).resolve().parents[4]
_AUTO_FIX_PATH = _find_repo_root() / ".claude" / "hooks" / "auto_fix_diagnostics.py"
# ---------------------------------------------------------------------------
# hooks test subcommand
# ---------------------------------------------------------------------------
def test_cmd_hooks_test_runs_without_crash():
"""_cmd_hooks_test() runs without crash; subprocess.run is mocked to avoid real pytest spawn."""
from unittest.mock import patch as _patch
from aipass.seedgo.apps.modules import hooks as hooks_mod
fake_result = MagicMock()
fake_result.stdout = "1 passed in 0.1s\n"
fake_result.returncode = 0
with _patch("subprocess.run", return_value=fake_result):
result = hooks_mod.handle_command("hooks", ["test"])
assert result is True
def test_cmd_hooks_test_handles_no_test_files(monkeypatch):
"""When glob returns no files, hooks test warns and returns without crash."""
import glob as _glob_mod
from aipass.seedgo.apps.modules import hooks as hooks_mod
monkeypatch.setattr(_glob_mod, "glob", lambda p: [])
result = hooks_mod.handle_command("hooks", ["test"])
assert result is True
# ---------------------------------------------------------------------------
# hooks list subcommand
# ---------------------------------------------------------------------------
def test_cmd_hooks_list_runs_without_crash():
"""_cmd_hooks_list() via handle_command should not raise."""
from aipass.seedgo.apps.modules import hooks as hooks_mod
result = hooks_mod.handle_command("hooks", ["list"])
assert result is True
def test_read_settings_file_missing():
"""_read_settings_file returns {} for a nonexistent path."""
from aipass.seedgo.apps.modules.hooks_ext import read_settings_file
result = read_settings_file(Path("/tmp/nonexistent_track_b_test.json"))
assert result == {}
def test_read_settings_file_valid(tmp_path, monkeypatch):
"""_read_settings_file returns the parsed dict for a valid JSON file."""
from aipass.seedgo.apps.modules import hooks_ext as hooks_ext_mod
data = {"hooks": {"PreToolUse": []}, "version": 1}
settings_file = tmp_path / "settings.json"
settings_file.write_text(json.dumps(data), encoding="utf-8")
monkeypatch.setattr(hooks_ext_mod, "read_text_safe", lambda path: path.read_text(encoding="utf-8"))
result = hooks_ext_mod.read_settings_file(settings_file)
assert result == data
def test_extract_hook_script_name():
"""extract_hook_script extracts the .py filename from a command string."""
from aipass.seedgo.apps.modules.hooks_ext import extract_hook_script
# Build command string dynamically to avoid help_text checker pattern
interpreter = "python" + "3"
cmd = interpreter + " /some/path/auto_fix_diagnostics.py"
result = extract_hook_script(cmd)
assert result == "auto_fix_diagnostics.py"
def test_read_hook_version_returns_string(monkeypatch):
"""read_hook_version returns the version string from the hook file header."""
from aipass.seedgo.apps.modules import hooks_ext as hooks_ext_mod
if not _AUTO_FIX_PATH.exists():
pytest.skip(f"auto_fix_diagnostics.py not found at {_AUTO_FIX_PATH}")
# Provide real lines from the file via read_lines_safe mock
real_lines = _AUTO_FIX_PATH.read_text(encoding="utf-8").splitlines()[:20]
monkeypatch.setattr(hooks_ext_mod, "read_lines_safe", lambda path, n=20: real_lines)
# Build command string dynamically to avoid help_text checker pattern
interpreter = "python" + "3"
cmd = interpreter + " " + str(_AUTO_FIX_PATH)
result = hooks_ext_mod.read_hook_version(cmd)
assert result == "5.2.0"
src/aipass/seedgo/tests/test_hooks_track_e.py
+2 -128
View File
@@ -1,22 +1,20 @@
# =================== AIPass ====================
# Name: test_hooks_track_e.py
# Description: DPLAN-0139 Track E — single-path enforcement tests
# Version: 1.0.0
# Version: 1.1.0
# Created: 2026-04-21
# Modified: 2026-04-21
# Modified: 2026-06-05
# =============================================
"""Tests for DPLAN-0139 Track E — single-path enforcement.
Covers:
- permissions.py: TRUSTED_CROSS_WRITERS, is_trusted_caller(), identify_caller()
- pre_edit_gate.py v1.3.0: inbox lock rule + cross-branch write rule
- drone auth.py: ALLOWED_CALLERS derived from TRUSTED_CROSS_WRITERS
- inbox_audit.py: handle_command routing + _scan_inbox validation
- delivery.py: deliver_to_inbox_file single-path helper
"""
import importlib.util
import io
import json
from pathlib import Path
from unittest.mock import patch
@@ -39,19 +37,6 @@ def _find_repo_root() -> Path:
REPO_ROOT = _find_repo_root()
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
def _load_hook(name: str):
"""Import a hook script by filename via importlib (outside package)."""
path = HOOKS_DIR / name
if not path.exists():
pytest.skip(f"Hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
# ---------------------------------------------------------------------------
@@ -130,117 +115,6 @@ def test_identify_caller_falls_back_to_identity_name(tmp_path):
assert result == "fallback_branch"
# ---------------------------------------------------------------------------
# pre_edit_gate.py v1.3.0 — Track E rules
# ---------------------------------------------------------------------------
def test_gate_allows_non_edit_tool():
"""Non-edit tools (Read) must pass through without blocking."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": "/tmp/foo.py"}})
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
def test_gate_blocks_inbox_json_write(capsys):
"""Any write targeting .ai_mail.local/inbox.json must be blocked."""
mod = _load_hook("pre_edit_gate.py")
inbox_path = "/home/user/Projects/AIPass/src/aipass/flow/.ai_mail.local/inbox.json"
payload = json.dumps({"tool_name": "Write", "tool_input": {"file_path": inbox_path}})
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
result = json.loads(captured.out)
assert result["decision"] == "block"
assert "inbox.json" in result["reason"].lower() or "drone" in result["reason"].lower()
def test_gate_blocks_cross_branch_write_from_untrusted(capsys):
"""Untrusted branch writing to a different branch must be blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/memory",
}
)
with patch("sys.stdin", io.StringIO(payload)):
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 2
captured = capsys.readouterr()
result = json.loads(captured.out)
assert result["decision"] == "block"
def test_gate_allows_cross_branch_write_from_devpulse(capsys):
"""devpulse may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/devpulse",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_cross_branch_write_from_seedgo(capsys):
"""seedgo may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/seedgo",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_cross_branch_write_from_spawn(capsys):
"""spawn may write to any branch without being blocked."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Write",
"tool_input": {"file_path": "/repo/src/aipass/prax/apps/modules/bar.py"},
"cwd": "/repo/src/aipass/spawn",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
def test_gate_allows_same_branch_write(capsys):
"""Writes within the same branch must not be blocked by the cross-branch rule."""
mod = _load_hook("pre_edit_gate.py")
payload = json.dumps(
{
"tool_name": "Edit",
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
"cwd": "/repo/src/aipass/flow",
}
)
with patch("sys.stdin", io.StringIO(payload)):
mod.main()
captured = capsys.readouterr()
assert captured.out == ""
# ---------------------------------------------------------------------------
# drone auth.py — ALLOWED_CALLERS derived from permissions
# ---------------------------------------------------------------------------
src/aipass/seedgo/tests/test_hooks_utility.py
-237
View File
@@ -1,237 +0,0 @@
"""Smoke tests for utility hook scripts.
# =================== META ====================
# Name: test_hooks_utility.py
# Description: Smoke tests for branch_prompt_loader, email_notification,
# identity_injector, pre_compact, notification_sound, stop_sound, tool_use_sound
# Version: 1.0.0
# Created: 2026-04-21
# Modified: 2026-04-21
# =============================================
"""
import importlib.util
import io
import json
from pathlib import Path
from unittest.mock import patch
# ---------------------------------------------------------------------------
# Loader helper
# ---------------------------------------------------------------------------
HOOKS_DIR = Path(__file__).resolve().parents[4] / ".claude" / "hooks"
def _load_hook(name: str):
"""Import a hook script by filename via importlib (outside package)."""
path = HOOKS_DIR / name
if not path.exists():
import pytest
pytest.skip(f"Hook script not found: {path}")
spec = importlib.util.spec_from_file_location(name.replace(".py", "").replace("-", "_"), path)
assert spec is not None and spec.loader is not None
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod) # type: ignore[union-attr]
return mod
# ---------------------------------------------------------------------------
# 1. branch_prompt_loader.py
# ---------------------------------------------------------------------------
def test_branch_prompt_loader_no_crash_empty_stdin():
"""branch_prompt_loader.main() doesn't crash — it reads no stdin at all."""
mod = _load_hook("branch_prompt_loader.py")
# main() doesn't read stdin; just ensure it runs without raising
with patch.object(mod, "find_branch_root", return_value=None):
mod.main() # must not raise
def test_branch_prompt_loader_no_branch_structure(tmp_path):
"""main() returns without crashing when CWD has no branch structure."""
mod = _load_hook("branch_prompt_loader.py")
# Run with a tmp_path that has no .trinity or .aipass
with patch("os.getcwd", return_value=str(tmp_path)):
mod.main() # must not raise
# ---------------------------------------------------------------------------
# 2. email_notification.py
# ---------------------------------------------------------------------------
def test_email_notification_no_crash_empty_stdin():
"""email_notification.main() doesn't crash even with no branch root."""
mod = _load_hook("email_notification.py")
with patch.object(mod, "find_branch_root", return_value=None):
mod.main() # must not raise
def test_email_notification_output_is_valid_when_emails_present(capsys, tmp_path):
"""When new emails exist, output is plain text (not JSON) containing count."""
mod = _load_hook("email_notification.py")
# Create a fake inbox with one new message
mail_dir = tmp_path / ".ai_mail.local"
mail_dir.mkdir()
inbox = mail_dir / "inbox.json"
inbox.write_text(
json.dumps({"messages": [{"id": "1", "status": "new", "subject": "hi"}]}),
encoding="utf-8",
)
with patch.object(mod, "find_branch_root", return_value=tmp_path):
mod.main()
captured = capsys.readouterr()
assert "1 new email" in captured.out
# ---------------------------------------------------------------------------
# 3. identity_injector.py
# ---------------------------------------------------------------------------
def test_identity_injector_no_crash_empty_stdin():
"""identity_injector.main() returns without raising when no branch root found."""
mod = _load_hook("identity_injector.py")
with patch.object(mod, "find_branch_root", return_value=None):
mod.main() # must not raise
def test_identity_injector_finds_passport_in_seedgo(capsys):
"""identity_injector finds .trinity/passport.json when CWD is in seedgo dir."""
mod = _load_hook("identity_injector.py")
# Use the actual seedgo directory which has a passport.json
seedgo_dir = Path(__file__).resolve().parents[2] # src/aipass/seedgo
trinity = seedgo_dir / ".trinity"
if not (trinity / "passport.json").exists():
# No passport here — just verify no crash
with patch.object(mod, "find_branch_root", return_value=None):
mod.main()
return
branch_root = mod.find_id_file(seedgo_dir)
# Either finds passport or returns None — both are valid
assert branch_root is None or branch_root.exists()
# ---------------------------------------------------------------------------
# 4. pre_compact.py
# ---------------------------------------------------------------------------
def test_pre_compact_no_crash_empty_hook_event(capsys):
"""pre_compact.main() handles stdin with hook_event_name field cleanly."""
mod = _load_hook("pre_compact.py")
payload = json.dumps({"hook_event_name": "manual"})
with patch("sys.stdin", io.StringIO(payload)), patch("sys.exit"):
mod.main() # must not raise
def test_pre_compact_output_json_or_empty(capsys):
"""pre_compact.main() produces output or is empty — never raises."""
mod = _load_hook("pre_compact.py")
payload = json.dumps({"hook_event_name": "manual"})
with patch("sys.stdin", io.StringIO(payload)), patch("sys.exit"):
mod.main()
# Capture is fine either way — no assertion on content, just no exception
# ---------------------------------------------------------------------------
# 5. notification_sound.py
# ---------------------------------------------------------------------------
def test_notification_sound_no_crash_empty_stdin():
"""notification_sound.main() doesn't crash on empty stdin {}."""
mod = _load_hook("notification_sound.py")
with (
patch("sys.stdin", io.StringIO("{}")),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main()
# No Popen call expected — event name didn't match
mock_popen.assert_not_called()
def test_notification_sound_runs_without_exception_when_sound_missing():
"""main() runs without exception even when sound file is missing."""
mod = _load_hook("notification_sound.py")
payload = json.dumps({"hook_event_name": "Notification"})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "SOUND_FILE", Path("/tmp/no_such_sound_file.wav")),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main() # must not raise
# Sound file doesn't exist, so play_sound() returns early — Popen not called
mock_popen.assert_not_called()
# ---------------------------------------------------------------------------
# 6. stop_sound.py
# ---------------------------------------------------------------------------
def test_stop_sound_no_crash_empty_stdin():
"""stop_sound.main() doesn't crash on empty stdin {}."""
mod = _load_hook("stop_sound.py")
with (
patch("sys.stdin", io.StringIO("{}")),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main()
mock_popen.assert_not_called()
def test_stop_sound_runs_without_exception_when_sound_missing():
"""main() runs without exception even when sound file is missing."""
mod = _load_hook("stop_sound.py")
payload = json.dumps({"hook_event_name": "Stop"})
with (
patch("sys.stdin", io.StringIO(payload)),
patch.object(mod, "SOUND_FILE", Path("/tmp/no_such_sound_file.wav")),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main() # must not raise
mock_popen.assert_not_called()
# ---------------------------------------------------------------------------
# 7. tool_use_sound.py
# ---------------------------------------------------------------------------
def test_tool_use_sound_no_crash_empty_stdin():
"""tool_use_sound.main() doesn't crash on empty stdin {}."""
mod = _load_hook("tool_use_sound.py")
with (
patch("sys.stdin", io.StringIO("{}")),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main()
mock_popen.assert_not_called()
def test_tool_use_sound_does_not_fire_for_non_tool_use_event():
"""tool_use_sound.main() does not play sound for a non-PreToolUse event."""
mod = _load_hook("tool_use_sound.py")
payload = json.dumps({"hook_event_name": "Stop", "tool_name": "Bash"})
with (
patch("sys.stdin", io.StringIO(payload)),
patch("subprocess.Popen") as mock_popen,
patch("sys.exit"),
):
mod.main()
mock_popen.assert_not_called()