feat(system): feat(seedgo): audit cleanup — hooks.py + hooks_ext.py + new handlers/hooks/ (rescue from @seedgo silent-finish)
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
@@ -33,3 +33,12 @@ def read_lines_safe(path: Path, n: int = 0, encoding: str = "utf-8") -> list[str
|
||||
return fh.readlines()
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
|
||||
def write_text_safe(path: Path, text: str, encoding: str = "utf-8") -> bool:
|
||||
"""Write text to a file. Returns True on success, False on OSError."""
|
||||
try:
|
||||
path.write_text(text, encoding=encoding)
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
# =================== 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
|
||||
@@ -48,6 +48,9 @@ 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, cmd_hooks_test
|
||||
|
||||
@@ -206,10 +209,8 @@ def _cmd_probe_display(log_path: Path | None = None) -> None:
|
||||
def _run_headless_claude() -> int:
|
||||
"""Spawn headless claude with a Read tool call. Returns exit code."""
|
||||
canary = Path("/tmp/probe_canary.txt")
|
||||
try:
|
||||
canary.write_text("probe canary 2026-04-20\n", encoding="utf-8")
|
||||
except OSError as exc:
|
||||
logger.info("hooks.py: could not write canary: %s", exc)
|
||||
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
|
||||
@@ -545,8 +546,10 @@ def handle_command(command: str, args: List[str]) -> bool:
|
||||
if command != "hooks":
|
||||
return False
|
||||
|
||||
# No args or help -> introspection
|
||||
if not args or args[0] in ("--help", "-h", "help"):
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
if args[0] in ("--help", "-h", "help"):
|
||||
print_introspection()
|
||||
return True
|
||||
|
||||
|
||||
@@ -17,16 +17,13 @@ Extended subcommands for hooks.py — split out to keep hooks.py under 700 lines
|
||||
|
||||
import glob as _glob
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
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
|
||||
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
|
||||
|
||||
@@ -36,25 +33,6 @@ from rich.table import Table
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _parse_pytest_counts(stdout: str) -> tuple[int, int]:
|
||||
"""Parse passed/failed counts from pytest -q output. Returns (passed, failed)."""
|
||||
passed = 0
|
||||
failed = 0
|
||||
for line in 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
|
||||
|
||||
|
||||
def cmd_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")
|
||||
@@ -82,16 +60,7 @@ def cmd_hooks_test(repo_root: Path) -> None:
|
||||
|
||||
for tf in test_files:
|
||||
stem = Path(tf).stem
|
||||
t0 = time.monotonic()
|
||||
proc = subprocess.run(
|
||||
[sys.executable, "-m", "pytest", tf, "--tb=no", "-q", "--no-header"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
cwd=str(repo_root),
|
||||
)
|
||||
duration = time.monotonic() - t0
|
||||
passed, failed = _parse_pytest_counts(proc.stdout)
|
||||
passed, failed, duration = run_pytest_file(Path(tf), repo_root)
|
||||
total_passed += passed
|
||||
total_failed += failed
|
||||
status = (
|
||||
@@ -126,10 +95,14 @@ 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(path.read_text(encoding="utf-8"))
|
||||
return json.loads(text)
|
||||
except Exception as exc:
|
||||
logger.info("hooks_ext.py: could not read %s: %s", path, exc)
|
||||
logger.info("hooks_ext.py: could not parse %s: %s", path, exc)
|
||||
return {}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user