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

feat: shared hook sound module — mute all 14 handlers via drone @hooks hooksound on/off

Browse Source
This commit is contained in:
AIOSAI
2026-05-22 19:52:42 -07:00
parent 2215fcb260
commit 7d02c3d753
39 changed files with 537 additions and 1089 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.aipass/hooks.json
-4
View File
@@ -10,7 +10,6 @@
},
"email_notification": {
"enabled": true,
"audio": true,
"handler": "aipass.hooks.apps.handlers.notification.email.handle",
"matcher": ""
},
@@ -29,7 +28,6 @@
"PreToolUse": {
"tool_use_sound": {
"enabled": true,
"audio": true,
"handler": "aipass.hooks.apps.handlers.notification.tool_sound.handle",
"matcher": "Bash|Edit|MultiEdit|Write|Read|Grep|Glob|WebSearch|WebFetch|Task"
},
@@ -76,7 +74,6 @@
"Stop": {
"stop_sound": {
"enabled": true,
"audio": true,
"handler": "aipass.hooks.apps.handlers.notification.stop_sound.handle",
"matcher": ""
}
@@ -85,7 +82,6 @@
"Notification": {
"notification_sound": {
"enabled": true,
"audio": true,
"handler": "aipass.hooks.apps.handlers.notification.announce.handle",
"matcher": ""
}
CHANGELOG.md
+6 -6
View File
@@ -27,12 +27,12 @@ versioned release with notes.
can enable, disable, or customize individual hooks without touching
provider-level settings. Previously hooks fired globally with no
per-project control.
- **Audio feedback on hook events** using Piper TTS (muted by default).
Tool usage, response completion, notifications, email alerts, and
sub-agent activity each produce distinct spoken audio cues so operators
can monitor sessions without watching the terminal. Toggle with
`drone hook-sounds on|off`. Audio hooks are tagged `"audio": true` in
`.aipass/hooks.json` and the engine skips them when muted.
- **Audio feedback on hook events** using Piper TTS. All 14 handlers
produce distinct spoken audio cues so operators can monitor sessions
without watching the terminal. A shared sound module
(`hooks/apps/sound.py`) provides `speak()` and `play()` with built-in
mute support. Toggle with `drone @hooks hooksound on|off` — muting
silences all 14 handlers without skipping their functional logic.
- **Hooks agent** — the 13th citizen in the AIPass registry, owning all
hook infrastructure: the engine, bridge, handlers, and configuration
schema.
src/aipass/drone/README.md
+6 -7
View File
@@ -72,7 +72,6 @@ drone list # List registered custom command shortcuts
drone remove <name> # Remove a custom command shortcut
# Utilities
drone hook-sounds on|off # Toggle hook notification sounds
drone --version # Show version (v1.1.0)
drone --help # Show usage information
```
@@ -184,8 +183,8 @@ drone/
│ │ ├── merge_plugin.py # PR merge (--merge) + local sync
│ │ ├── sync_plugin.py # Smart sync (fetch, divergence detect, rebase)
│ │ └── fix_plugin.py # Auto-fix stuck rebase / detached HEAD
│ └── hook_sounds/
│ └── hook_sounds_plugin.py # Toggle notification sounds on/off
│ └── hook_sounds/ # DISABLED — moved to hooks branch (drone @hooks hooksound on/off)
│ └── hook_sounds_plugin.py.disabled
├── docs/ # Public documentation
├── docs.local/ # Investigation reports and policies
└── tests/ # 704 tests across 21 test files
@@ -194,7 +193,7 @@ drone/
### Routing Flow
1. **CLI input** → `drone.py:main()`
2. **Built-in commands** checked first: `systems`, `scan`, `activate`, `list`, `remove`, `hook-sounds`
2. **Built-in commands** checked first: `systems`, `scan`, `activate`, `list`, `remove`
3. **`@target` routing** → branch resolution via `AIPASS_REGISTRY.json` → subprocess dispatch
4. **Module fallback** → if branch not found but is a registered module, routes internally
5. **Bare module names** → auto-discovered from `apps/modules/*.py`, routed via `importlib`
@@ -277,9 +276,9 @@ Auth-gated operations for system administration. `auth.py` walks CWD for `.trini
| `sync_plugin` | `smart-sync` | Fetch + detect divergence + rebase |
| `fix_plugin` | `fix` | Auto-fix stuck rebase / detached HEAD |
### hook_sounds
### hook_sounds (DISABLED)
Simple toggle for hook notification sounds. Creates/removes `/tmp/aipass-hooks-muted` flag file.
Moved to hooks branch as `drone @hooks hooksound on/off`. Plugin file renamed to `.disabled`.
---
@@ -325,7 +324,7 @@ Tip: set AIPASS_HOME=/path/to/AIPass to access all branches
| Git operations | `test_git_module.py`, `test_system_pr.py`, `test_devpulse_plugins.py`, `test_git_access.py` | ~150 |
| Handlers | `test_executor.py`, `test_registry_handler.py`, `test_discovery.py` | ~99 |
| Infrastructure | `test_generic_adapter.py`, `test_module_registry.py`, `test_config.py` | ~66 |
| Features | `test_commands.py`, `test_scan.py`, `test_hook_sounds.py`, `test_json_handler.py` | ~125 |
| Features | `test_commands.py`, `test_scan.py`, `test_json_handler.py` | ~125 |
| Standards | `test_cli_routing.py`, `test_contracts.py`, `test_error_resilience.py`, `test_init_provisioning.py` | ~21 |
Run tests: `cd src/aipass/drone && python -m pytest tests/ -q`
src/aipass/drone/apps/drone.py
-9
View File
@@ -85,7 +85,6 @@ def show_help() -> None:
table.add_row("activate @target", "Register all commands from a branch")
table.add_row("list", "List registered custom commands")
table.add_row("remove <name>", "Remove a custom command")
table.add_row("hook-sounds on|off", "Toggle hook notification sounds")
table.add_row("--help", "Show this help")
table.add_row("--version", "Show version")
@@ -551,14 +550,6 @@ def main() -> int:
if command == "list":
return _handle_list()
# hook-sounds — toggle hook notification sounds
if command == "hook-sounds":
from aipass.drone.apps.plugins.hook_sounds.hook_sounds_plugin import handle_command as hs_handle
cmd = args[1] if len(args) > 1 else None
hs_handle(cmd)
return 0
# remove — remove a custom command by name
if command == "remove":
if len(args) < 2:
src/aipass/drone/apps/plugins/hook_sounds/__init__.py → src/aipass/drone/apps/plugins/hook_sounds/__init__.py.disabled
View File
src/aipass/drone/apps/plugins/hook_sounds/hook_sounds_plugin.py → src/aipass/drone/apps/plugins/hook_sounds/hook_sounds_plugin.py.disabled
View File
src/aipass/drone/tests/test_cli_routing.py
-30
View File
@@ -265,36 +265,6 @@ class TestMainList:
mock_list.assert_called_once()
class TestMainHookSounds:
"""drone hook-sounds command."""
_HS = "aipass.drone.apps.plugins.hook_sounds.hook_sounds_plugin.handle_command"
def test_hook_sounds_on(self) -> None:
"""hook-sounds on delegates to plugin."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "hook-sounds", "on"]),
patch(self._HS) as mock_hs,
):
result = main()
assert result == 0
mock_hs.assert_called_once_with("on")
def test_hook_sounds_no_arg(self) -> None:
"""hook-sounds with no arg passes None."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "hook-sounds"]),
patch(self._HS) as mock_hs,
):
result = main()
assert result == 0
mock_hs.assert_called_once_with(None)
class TestMainRemove:
"""drone remove command."""
src/aipass/drone/tests/test_hook_sounds.py → src/aipass/drone/tests/test_hook_sounds.py.disabled
View File
src/aipass/hooks/README.md
+8 -4
View File
@@ -21,7 +21,9 @@ Every hook event flows through one engine. Platform bridges normalize the event
|---|---|
| `drone @hooks status` | Show hook config for current project |
| `drone @hooks log` | Tail recent hook activity (last 20 JSONL entries) |
| `drone @hooks test` | Run hook test suite (planned) |
| `drone @hooks hooksound` | Show current sound mute status |
| `drone @hooks hooksound off` | Mute all hook sounds |
| `drone @hooks hooksound on` | Unmute all hook sounds |
| `drone @hooks --help` | Full help reference |
| `drone @hooks --version` | Version info |
@@ -32,8 +34,10 @@ src/aipass/hooks/
├── .trinity/ # Identity & memory
├── apps/
│ ├── hooks.py # Entry point (drone @hooks)
│ ├── sound.py # Shared sound utilities (speak, play, mute)
│ ├── modules/
│ │ └── engine.py # Core dispatch — routes events to handlers
│ │ ├── engine.py # Core dispatch — routes events to handlers
│ │ └── hooksound.py # Sound control (drone @hooks hooksound on/off)
│ ├── handlers/
│ │ ├── bridges/ # One per provider (thin normalization)
│ │ │ └── claude.py # Claude Code bridge
@@ -44,7 +48,7 @@ src/aipass/hooks/
│ └── config/ # hooks.json validation
├── logs/
│ └── engine.jsonl # JSONL diagnostics (every hook execution)
├── tests/
├── tests/ # 236 tests
└── STATUS.local.md
```
@@ -70,7 +74,7 @@ src/aipass/hooks/
All branches via hook dispatch. Every Claude Code session routes through the engine.
*Last Updated: 2026-05-18*
*Last Updated: 2026-05-22*
---
src/aipass/hooks/apps/handlers/lifecycle/auto_fix.py
+2 -26
View File
@@ -14,35 +14,11 @@ import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] auto_fix: speak error: %s", exc)
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
STATE_FILE = Path(__file__).parent.parent.parent.parent.parent / ".diagnostics_state.json"
SKIP_EXTENSIONS = {".md", ".txt", ".log", ".csv", ".html"}
@@ -345,7 +321,7 @@ def handle(hook_data: dict) -> dict:
if ext in SKIP_EXTENSIONS:
return {"stdout": "", "exit_code": 0}
_speak("auto fix diagnostics")
speak("auto fix diagnostics")
errors: list[str] = []
src/aipass/hooks/apps/handlers/lifecycle/auto_watchdog.py
+2 -28
View File
@@ -11,34 +11,8 @@
"""Checks for dispatch commands and reminds the agent to arm the watchdog."""
import json
import subprocess
import tempfile
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] auto_watchdog: speak error: %s", exc)
from aipass.hooks.apps.sound import speak
def handle(hook_data: dict) -> dict:
@@ -65,7 +39,7 @@ def handle(hook_data: dict) -> dict:
if "dispatch wake" in command and "dispatch @" not in command:
return {"stdout": "", "exit_code": 0}
_speak("auto watchdog")
speak("auto watchdog")
result = {
"additionalContext": (
src/aipass/hooks/apps/handlers/lifecycle/compact.py
+2 -25
View File
@@ -13,34 +13,11 @@
import json
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] compact: speak error: %s", exc)
def _find_branch_dir(cwd: str) -> Path | None:
parts = Path(cwd).parts
@@ -116,7 +93,7 @@ def _get_git_info() -> str | None:
def handle(hook_data: dict) -> dict:
"""Inject live branch state for post-compact recovery."""
_speak("pre compact")
speak("pre compact")
try:
cwd = hook_data.get("cwd", "") or str(Path.cwd())
src/aipass/hooks/apps/handlers/lifecycle/rollover.py
+2 -25
View File
@@ -13,34 +13,11 @@
import json
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] rollover: speak error: %s", exc)
def _find_repo_root() -> Path | None:
aipass_home = os.environ.get("AIPASS_HOME", "")
@@ -154,7 +131,7 @@ def _run_rollover(repo_root: Path) -> tuple[bool, str]:
def handle(hook_data: dict) -> dict:
"""Check memory files for overflow and trigger rollover if needed."""
_speak("pre compact rollover")
speak("pre compact rollover")
try:
repo_root = _find_repo_root()
src/aipass/hooks/apps/handlers/notification/announce.py
+2 -53
View File
@@ -11,65 +11,14 @@
"""Plays announcement tone + Piper voice ID on Notification events."""
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.hooks.apps.sound import speak
AIPASS_HOME = Path(os.environ.get("AIPASS_HOME", ""))
SOUNDS_DIR = AIPASS_HOME / ".claude" / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-clear-announce-tones-2861.wav"
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _play(sound_path: Path) -> None:
"""Play a WAV file via aplay (fire-and-forget)."""
if not sound_path.exists():
logger.info("[HOOKS] announce: file not found: %s", sound_path)
return
try:
subprocess.Popen(
["aplay", "-q", str(sound_path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except OSError as exc:
logger.info("[HOOKS] announce: playback error: %s", exc)
def _speak(text: str) -> None:
"""Generate speech via Piper TTS and play it (fire-and-forget)."""
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
logger.info("[HOOKS] announce: piper not available")
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
piper_result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if piper_result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(
["aplay", "-q", wav_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
logger.info("[HOOKS] announce: piper timed out")
except OSError as exc:
logger.info("[HOOKS] announce: speak error: %s", exc)
def handle(hook_data: dict) -> dict:
"""Play notification tone and speak hook name for identification.
@@ -80,5 +29,5 @@ def handle(hook_data: dict) -> dict:
Returns:
Result dict with stdout (empty) and exit_code.
"""
_speak("notification sound")
speak("notification sound")
return {"stdout": "", "exit_code": 0}
src/aipass/hooks/apps/handlers/notification/email.py
+2 -37
View File
@@ -11,46 +11,11 @@
"""Checks branch inbox for unread emails and returns notification text."""
import json
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
"""Generate speech via Piper TTS and play it (fire-and-forget)."""
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
logger.info("[HOOKS] email: piper not available")
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
piper_result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if piper_result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(
["aplay", "-q", wav_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
logger.info("[HOOKS] email: piper timed out")
except OSError as exc:
logger.info("[HOOKS] email: speak error: %s", exc)
def _find_branch_root() -> Path | None:
"""Find the branch root by walking up from CWD looking for branch markers."""
@@ -132,7 +97,7 @@ def handle(hook_data: dict) -> dict:
return {"stdout": "", "exit_code": 0}
plural = "s" if new_count != 1 else ""
_speak(f"email notification: {new_count} new email{plural}")
speak(f"email notification: {new_count} new email{plural}")
msg = f"You have {new_count} new email{plural} - check with: drone @ai_mail inbox | then: drone @ai_mail view <id> | close with: drone @ai_mail close <id>"
logger.info("[HOOKS] email: %d new email%s", new_count, plural)
return {"stdout": msg, "exit_code": 0}
src/aipass/hooks/apps/handlers/notification/stop_sound.py
+2 -53
View File
@@ -11,65 +11,14 @@
"""Plays achievement bell when the AI finishes responding (Stop event)."""
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.hooks.apps.sound import speak
AIPASS_HOME = Path(os.environ.get("AIPASS_HOME", ""))
SOUNDS_DIR = AIPASS_HOME / ".claude" / "sounds"
SOUND_FILE = SOUNDS_DIR / "mixkit-achievement-bell-600.wav"
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _play(sound_path: Path) -> None:
"""Play a WAV file via aplay (fire-and-forget)."""
if not sound_path.exists():
logger.info("[HOOKS] stop_sound: file not found: %s", sound_path)
return
try:
subprocess.Popen(
["aplay", "-q", str(sound_path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except OSError as exc:
logger.info("[HOOKS] stop_sound: playback error: %s", exc)
def _speak(text: str) -> None:
"""Generate speech via Piper TTS and play it (fire-and-forget)."""
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
logger.info("[HOOKS] stop_sound: piper not available")
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
piper_result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if piper_result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(
["aplay", "-q", wav_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
logger.info("[HOOKS] stop_sound: piper timed out")
except OSError as exc:
logger.info("[HOOKS] stop_sound: speak error: %s", exc)
def handle(hook_data: dict) -> dict:
"""Play achievement bell and speak hook name on Stop event.
@@ -83,5 +32,5 @@ def handle(hook_data: dict) -> dict:
if hook_data.get("stop_hook_active", False):
return {"stdout": "", "exit_code": 0}
_speak("stop sound")
speak("stop sound")
return {"stdout": "", "exit_code": 0}
src/aipass/hooks/apps/handlers/notification/tool_sound.py
+2 -40
View File
@@ -10,45 +10,7 @@
"""Announces hook name via Piper TTS when the AI uses tools (PreToolUse event)."""
import subprocess
import tempfile
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
"""Generate speech via Piper TTS and play it (fire-and-forget)."""
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
logger.info("[HOOKS] tool_sound: piper not available")
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
piper_result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if piper_result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(
["aplay", "-q", wav_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
logger.info("[HOOKS] tool_sound: piper timed out")
except OSError as exc:
logger.info("[HOOKS] tool_sound: playback error: %s", exc)
from aipass.hooks.apps.sound import speak
def handle(hook_data: dict) -> dict:
@@ -64,5 +26,5 @@ def handle(hook_data: dict) -> dict:
if not tool_name:
return {"stdout": "", "exit_code": 0}
_speak(f"tool sound: {tool_name}")
speak(f"tool sound: {tool_name}")
return {"stdout": "", "exit_code": 0}
src/aipass/hooks/apps/handlers/prompt/branch_loader.py
+2 -26
View File
@@ -10,35 +10,11 @@
"""Loads .aipass/aipass_local_prompt.md and private integration prompts for injection."""
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] branch_loader: speak error: %s", exc)
def _find_branch_root(cwd: str) -> Path | None:
"""Walk up from CWD looking for .trinity/ or apps/ — stop at repo root."""
@@ -54,7 +30,7 @@ def _find_branch_root(cwd: str) -> Path | None:
def handle(hook_data: dict) -> dict:
"""Load branch prompt and private integration prompts."""
_speak("branch prompt")
speak("branch prompt")
try:
cwd = hook_data.get("cwd", "") or str(Path.cwd())
src/aipass/hooks/apps/handlers/prompt/global_loader.py
+2 -26
View File
@@ -11,39 +11,15 @@
"""Loads .aipass/aipass_global_prompt.md from AIPASS_HOME for prompt injection."""
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] global_loader: speak error: %s", exc)
def handle(hook_data: dict) -> dict:
"""Load AIPass global prompt from AIPASS_HOME."""
_speak("global prompt")
speak("global prompt")
try:
aipass_home = os.environ.get("AIPASS_HOME", "")
src/aipass/hooks/apps/handlers/prompt/identity.py
+2 -26
View File
@@ -11,35 +11,11 @@
"""Reads .trinity/passport.json and outputs formatted identity for prompt injection."""
import json
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] identity: speak error: %s", exc)
def _find_passport(cwd: str) -> Path | None:
"""Walk up from CWD looking for .trinity/passport.json."""
@@ -93,7 +69,7 @@ def _format_identity(data: dict) -> str:
def handle(hook_data: dict) -> dict:
"""Inject branch identity from passport.json into prompt context."""
_speak("identity")
speak("identity")
try:
cwd = hook_data.get("cwd", "") or str(Path.cwd())
src/aipass/hooks/apps/handlers/security/edit_gate.py
+2 -26
View File
@@ -12,35 +12,11 @@
import json
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] edit_gate: speak error: %s", exc)
STATE_FILE = Path(__file__).parent.parent.parent.parent.parent / ".diagnostics_state.json"
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
@@ -64,7 +40,7 @@ def handle(hook_data: dict) -> dict:
Returns:
Result dict with stdout (block JSON or empty) and exit_code.
"""
_speak("edit gate")
speak("edit gate")
try:
tool_name = hook_data.get("tool_name", "")
src/aipass/hooks/apps/handlers/security/git_gate.py
+2 -26
View File
@@ -13,35 +13,11 @@
import json
import os
import re
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] git_gate: speak error: %s", exc)
RAW_GIT_RE = re.compile(r"(?<![@\w/.])git\s")
RAW_GH_RE = re.compile(r"(?<![@\w/.])gh\s")
@@ -132,7 +108,7 @@ def handle(hook_data: dict) -> dict:
Returns:
Result dict with stdout (block JSON or empty) and exit_code.
"""
_speak("git gate")
speak("git gate")
try:
tool_name = hook_data.get("tool_name", "")
src/aipass/hooks/apps/handlers/security/subagent_gate.py
+2 -25
View File
@@ -13,37 +13,14 @@
import json
import os
import subprocess
import tempfile
from pathlib import Path
from aipass.hooks.apps.sound import speak
from aipass.prax.apps.modules.logger import system_logger as logger
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
_ALLOW = {"stdout": "", "exit_code": 0}
def _speak(text: str) -> None:
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False, mode="wb")
wav_path = wav_file.name
wav_file.close()
result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(["aplay", "-q", wav_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
except (subprocess.TimeoutExpired, OSError) as exc:
logger.info("[HOOKS] subagent_gate: speak error: %s", exc)
def _block(reason: str) -> dict:
return {"stdout": json.dumps({"decision": "block", "reason": reason}), "exit_code": 2}
@@ -160,7 +137,7 @@ def _check_hook_readme_accountability(cwd: str, repo_root: Path) -> str | None:
def handle(hook_data: dict) -> dict:
"""Check modified files against seedgo standards on subagent stop."""
_speak("subagent stop gate")
speak("subagent stop gate")
try:
cwd = hook_data.get("cwd", "") or os.getcwd()
src/aipass/hooks/apps/modules/engine.py
-6
View File
@@ -108,18 +108,12 @@ def dispatch(event_type: str, stdin_data: str, config: dict) -> str:
outputs = []
total_start = time.monotonic()
muted = Path("/tmp/aipass-hooks-muted").exists()
for hook_name, hook_def in event_hooks.items():
if not hook_def.get("enabled", True):
logger.info("[HOOKS] %s.%s skipped (disabled)", event_type, hook_name)
_log({"ts": time.time(), "event": event_type, "hook": hook_name, "action": "skipped_disabled"})
continue
if muted and hook_def.get("audio"):
_log({"ts": time.time(), "event": event_type, "hook": hook_name, "action": "skipped_muted"})
continue
handler = hook_def.get("handler", "")
command = hook_def.get("command", "")
matcher = hook_def.get("matcher", "")
src/aipass/hooks/apps/modules/hooksound.py
+59
View File
@@ -0,0 +1,59 @@
# =================== AIPass ====================
# Name: hooksound.py
# Version: 1.0.0
# Description: Hook sound control — mute/unmute all hook audio
# Branch: hooks
# Layer: apps/modules
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Hook sound control — mute and unmute all hook audio via drone @hooks hooksound."""
from aipass.cli.apps.modules import err_console
from aipass.hooks.apps.sound import MUTE_FLAG, is_muted
CONSOLE = err_console
def print_introspection():
"""Print module structure for drone routing."""
status = "MUTED" if is_muted() else "ACTIVE"
CONSOLE.print(f"[bold cyan]hooksound[/bold cyan] — Hook sound control ({status})")
def handle_command(command: str, args: list) -> bool:
"""Route hooksound commands from drone @hooks."""
if command == "hooksound":
sub = args[0] if args else None
if sub in ("--help", "-h", "help"):
CONSOLE.print("[bold cyan]hooksound[/bold cyan] — Mute/unmute all hook audio")
CONSOLE.print()
CONSOLE.print(" drone @hooks hooksound Show current status")
CONSOLE.print(" drone @hooks hooksound on Unmute all hook sounds")
CONSOLE.print(" drone @hooks hooksound off Mute all hook sounds")
return True
if sub == "off":
MUTE_FLAG.touch()
CONSOLE.print("[yellow]Hook sounds MUTED[/yellow]")
return True
if sub == "on":
if MUTE_FLAG.exists():
MUTE_FLAG.unlink()
CONSOLE.print("[green]Hook sounds ACTIVE[/green]")
return True
if sub is None:
if is_muted():
CONSOLE.print("[yellow]Hook sounds: MUTED[/yellow]")
CONSOLE.print(f" Flag: {MUTE_FLAG}")
CONSOLE.print(" Run: drone @hooks hooksound on")
else:
CONSOLE.print("[green]Hook sounds: ACTIVE[/green]")
CONSOLE.print(" Run: drone @hooks hooksound off")
return True
return False
src/aipass/hooks/apps/sound.py
+85
View File
@@ -0,0 +1,85 @@
# =================== AIPass ====================
# Name: sound.py
# Version: 1.0.0
# Description: Shared sound utilities — Piper TTS and WAV playback with mute support
# Branch: hooks
# Layer: apps
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Shared sound functions for hook handlers. Checks mute flag before playing."""
import subprocess
import tempfile
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import err_console
CONSOLE = err_console
MUTE_FLAG = Path("/tmp/aipass-hooks-muted")
PIPER_BIN = Path.home() / ".local" / "share" / "piper" / "piper"
PIPER_VOICE = Path.home() / ".local" / "share" / "piper-voices" / "en_US-amy-medium.onnx"
def print_introspection():
"""Print module structure for drone routing."""
CONSOLE.print("[bold cyan]sound[/bold cyan] — Shared sound utilities (speak, play, mute)")
def is_muted() -> bool:
"""Check whether hook sounds are currently muted."""
return MUTE_FLAG.exists()
def speak(text: str) -> None:
"""Generate speech via Piper TTS and play it. Skips if muted."""
if is_muted():
return
if not PIPER_BIN.exists() or not PIPER_VOICE.exists():
return
try:
wav_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
wav_path = wav_file.name
wav_file.close()
piper_result = subprocess.run(
[str(PIPER_BIN), "-m", str(PIPER_VOICE), "-f", wav_path],
input=text,
capture_output=True,
text=True,
timeout=5,
)
if piper_result.returncode == 0 and Path(wav_path).exists():
subprocess.Popen(
["aplay", "-q", wav_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except subprocess.TimeoutExpired:
logger.info("[HOOKS] speak: piper timed out")
except OSError as exc:
logger.info("[HOOKS] speak: playback error: %s", exc)
def play(sound_path: Path) -> None:
"""Play a WAV file via aplay. Skips if muted."""
if is_muted():
return
if not sound_path.exists():
logger.info("[HOOKS] play: file not found: %s", sound_path)
return
try:
subprocess.Popen(
["aplay", "-q", str(sound_path)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
except OSError as exc:
logger.info("[HOOKS] play: playback error: %s", exc)
src/aipass/hooks/tests/test_announce.py
+5 -139
View File
@@ -1,15 +1,15 @@
# =================== AIPass ====================
# Name: test_announce.py
# Version: 1.1.0
# Version: 1.2.0
# Description: Tests for announce notification handler
# Branch: hooks
# Created: 2026-05-20
# Modified: 2026-05-20
# Modified: 2026-05-22
# =============================================
"""Tests for handlers/notification/announce.py."""
from unittest.mock import patch, MagicMock
from unittest.mock import patch
class TestAnnounceHandler:
@@ -18,10 +18,7 @@ class TestAnnounceHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.announce import handle
with (
patch("aipass.hooks.apps.handlers.notification.announce._play"),
patch("aipass.hooks.apps.handlers.notification.announce._speak"),
):
with patch("aipass.hooks.apps.handlers.notification.announce.speak"):
result = handle({})
assert isinstance(result, dict)
@@ -31,138 +28,7 @@ class TestAnnounceHandler:
def test_handle_speaks_notification_sound(self):
from aipass.hooks.apps.handlers.notification.announce import handle
with (
patch("aipass.hooks.apps.handlers.notification.announce._play"),
patch("aipass.hooks.apps.handlers.notification.announce._speak") as mock_speak,
):
with patch("aipass.hooks.apps.handlers.notification.announce.speak") as mock_speak:
handle({})
mock_speak.assert_called_once_with("notification sound")
def test_handle_does_not_play_wav(self):
from aipass.hooks.apps.handlers.notification.announce import handle
with (
patch("aipass.hooks.apps.handlers.notification.announce._play") as mock_play,
patch("aipass.hooks.apps.handlers.notification.announce._speak"),
):
handle({})
mock_play.assert_not_called()
class TestPlayFunction:
"""WAV playback tests."""
def test_play_calls_aplay(self):
from aipass.hooks.apps.handlers.notification.announce import _play
mock_path = MagicMock()
mock_path.exists.return_value = True
with patch("aipass.hooks.apps.handlers.notification.announce.subprocess.Popen") as mock_popen:
_play(mock_path)
mock_popen.assert_called_once()
args = mock_popen.call_args[0][0]
assert args[0] == "aplay"
assert args[1] == "-q"
def test_play_skips_when_file_missing(self):
from aipass.hooks.apps.handlers.notification.announce import _play
mock_path = MagicMock()
mock_path.exists.return_value = False
with patch("aipass.hooks.apps.handlers.notification.announce.subprocess.Popen") as mock_popen:
_play(mock_path)
mock_popen.assert_not_called()
def test_play_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.announce import _play
mock_path = MagicMock()
mock_path.exists.return_value = True
with patch(
"aipass.hooks.apps.handlers.notification.announce.subprocess.Popen",
side_effect=OSError("broken"),
):
_play(mock_path)
class TestSpeakFunction:
"""Piper TTS tests."""
def test_speak_calls_piper_then_aplay(self):
from aipass.hooks.apps.handlers.notification.announce import _speak
with (
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.announce.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.announce.tempfile") as mock_tmp,
patch("aipass.hooks.apps.handlers.notification.announce.Path") as mock_path,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.return_value = MagicMock(returncode=0)
mock_path.return_value.exists.return_value = True
_speak("test text")
mock_sub.run.assert_called_once()
mock_sub.Popen.assert_called_once()
def test_speak_skips_when_piper_missing(self):
from aipass.hooks.apps.handlers.notification.announce import _speak
with (
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.announce.subprocess") as mock_sub,
):
mock_piper_bin.exists.return_value = False
_speak("test")
mock_sub.run.assert_not_called()
def test_speak_graceful_on_timeout(self):
import subprocess as real_sub
from aipass.hooks.apps.handlers.notification.announce import _speak
with (
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.announce.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.announce.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.side_effect = real_sub.TimeoutExpired("piper", 5)
mock_sub.TimeoutExpired = real_sub.TimeoutExpired
_speak("test")
def test_speak_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.announce import _speak
with (
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.announce.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.announce.subprocess.run", side_effect=OSError("broken")),
patch("aipass.hooks.apps.handlers.notification.announce.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
_speak("test")
src/aipass/hooks/tests/test_auto_fix.py
+11 -56
View File
@@ -61,14 +61,14 @@ class TestAutoFixSkips:
def test_skip_unknown_extension(self):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak"):
result = handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/file.xyz"}})
assert result["stdout"] == ""
assert result["exit_code"] == 0
class TestAutofixPython:
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@@ -81,7 +81,7 @@ class TestAutofixPython:
parsed = json.loads(result["stdout"])
assert parsed["systemMessage"] == "[diagnostics] ok"
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@@ -97,7 +97,7 @@ class TestAutofixPython:
assert "SYNTAX" in parsed["hookSpecificOutput"]["additionalContext"]
assert "1 error(s)" in parsed["systemMessage"]
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@@ -110,7 +110,7 @@ class TestAutofixPython:
parsed = json.loads(result["stdout"])
assert "LINT" in parsed["hookSpecificOutput"]["additionalContext"]
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
@@ -126,7 +126,7 @@ class TestAutofixPython:
parsed = json.loads(result["stdout"])
assert "TYPE: L42" in parsed["hookSpecificOutput"]["additionalContext"]
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
@@ -144,7 +144,7 @@ class TestAutofixPython:
class TestAutoFixStateFile:
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured")
@@ -171,7 +171,7 @@ class TestAutoFixStateFile:
if state_path.exists():
state_path.unlink()
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@@ -194,7 +194,7 @@ class TestAutoFixStateFile:
class TestAutoFixJson:
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
def test_json_valid(self, mock_speak, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
@@ -205,7 +205,7 @@ class TestAutoFixJson:
parsed = json.loads(result["stdout"])
assert parsed["systemMessage"] == "[diagnostics] ok"
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
def test_json_invalid_syntax(self, mock_speak, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
@@ -216,7 +216,7 @@ class TestAutoFixJson:
parsed = json.loads(result["stdout"])
assert "JSON SYNTAX" in parsed["hookSpecificOutput"]["additionalContext"]
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._speak")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.speak")
def test_json_corruption_detected(self, mock_speak, tmp_path):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
@@ -228,51 +228,6 @@ class TestAutoFixJson:
assert "EMOJI CORRUPTION" in parsed["hookSpecificOutput"]["additionalContext"]
class TestAutoFixPiper:
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.PIPER_VOICE")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.PIPER_BIN")
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_pyright_check", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_ruff_lint_structured", return_value=[])
@patch("aipass.hooks.apps.handlers.lifecycle.auto_fix._run_python_checks", return_value=[])
@patch("subprocess.run")
@patch("subprocess.Popen")
def test_piper_fires_on_edit(
self,
mock_popen,
mock_run,
mock_py,
mock_ruff_s,
mock_pyright,
mock_seedgo,
mock_piper_bin,
mock_piper_voice,
):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import handle
mock_piper_bin.exists.return_value = True
mock_piper_voice.exists.return_value = True
mock_run_result = MagicMock()
mock_run_result.returncode = 0
mock_run.return_value = mock_run_result
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.Path") as mock_path_cls:
mock_path_cls.return_value.suffix.lower.return_value = ".py"
mock_path_cls.return_value.name = "test.py"
mock_path_cls.return_value.exists.return_value = True
handle({"tool_name": "Edit", "tool_input": {"file_path": "/tmp/test.py"}})
assert mock_run.called
def test_piper_skips_when_unavailable(self):
from aipass.hooks.apps.handlers.lifecycle.auto_fix import _speak
with patch("aipass.hooks.apps.handlers.lifecycle.auto_fix.PIPER_BIN") as mock_bin:
mock_bin.exists.return_value = False
_speak("test")
class TestAutoFixSubprocessChecks:
@patch("subprocess.run")
def test_check_syntax_error(self, mock_run):
src/aipass/hooks/tests/test_branch_loader.py
+9 -9
View File
@@ -24,7 +24,7 @@ class TestBranchLoaderHandler:
prompt = aipass_dir / "aipass_local_prompt.md"
prompt.write_text("# Test Branch\nSome instructions", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
@@ -41,7 +41,7 @@ class TestBranchLoaderHandler:
private = integration / "private_prompt.md"
private.write_text("# Private Integration\nSecret stuff", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert "Private Integration" in result["stdout"]
@@ -58,7 +58,7 @@ class TestBranchLoaderHandler:
integration.mkdir(parents=True)
(integration / "private_prompt.md").write_text("Compass prompt", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert "Branch prompt" in result["stdout"]
@@ -67,7 +67,7 @@ class TestBranchLoaderHandler:
def test_returns_empty_when_no_branch_root(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["stdout"] == ""
@@ -79,7 +79,7 @@ class TestBranchLoaderHandler:
nested = tmp_path / "some" / "deep" / "path"
nested.mkdir(parents=True)
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(nested)})
assert result["stdout"] == ""
@@ -95,7 +95,7 @@ class TestBranchLoaderHandler:
nested = tmp_path / "apps" / "handlers" / "security"
nested.mkdir(parents=True)
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(nested)})
assert "Found it" in result["stdout"]
@@ -103,7 +103,7 @@ class TestBranchLoaderHandler:
def test_empty_hook_data(self):
from aipass.hooks.apps.handlers.prompt.branch_loader import handle
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
result = handle({})
@@ -116,7 +116,7 @@ class TestBranchLoaderHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["stdout"] == ""
@@ -130,7 +130,7 @@ class TestBranchLoaderHandler:
aipass_dir.mkdir()
(aipass_dir / "aipass_local_prompt.md").write_text("content", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.branch_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.branch_loader.speak"):
result = handle({"cwd": str(tmp_path)})
assert "Source:" in result["stdout"]
src/aipass/hooks/tests/test_compact.py
+5 -5
View File
@@ -32,7 +32,7 @@ class TestCompactHandler:
status = tmp_path / "STATUS.local.md"
status.write_text("# Status\nCurrent work here", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.lifecycle.compact._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value="Git branch: dev"):
result = handle({"cwd": str(tmp_path)})
@@ -45,7 +45,7 @@ class TestCompactHandler:
def test_returns_recovery_when_no_branch_dir(self):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
with patch("aipass.hooks.apps.handlers.lifecycle.compact._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
result = handle({"cwd": "/tmp/nonexistent"})
@@ -58,7 +58,7 @@ class TestCompactHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
with patch("aipass.hooks.apps.handlers.lifecycle.compact._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
with patch.dict("os.environ", {"AIPASS_SESSION_TYPE": "dispatched"}):
result = handle({"cwd": str(tmp_path)})
@@ -71,7 +71,7 @@ class TestCompactHandler:
trinity = tmp_path / ".trinity"
trinity.mkdir()
with patch("aipass.hooks.apps.handlers.lifecycle.compact._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
result = handle({"cwd": str(tmp_path)})
@@ -80,7 +80,7 @@ class TestCompactHandler:
def test_empty_hook_data(self):
from aipass.hooks.apps.handlers.lifecycle.compact import handle
with patch("aipass.hooks.apps.handlers.lifecycle.compact._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.compact._get_git_info", return_value=None):
with patch("pathlib.Path.cwd", return_value=MagicMock(parts=("/", "tmp"))):
result = handle({})
src/aipass/hooks/tests/test_email.py
+8 -84
View File
@@ -1,17 +1,17 @@
# =================== AIPass ====================
# Name: test_email.py
# Version: 1.1.0
# Version: 1.2.0
# Description: Tests for email notification handler
# Branch: hooks
# Created: 2026-05-21
# Modified: 2026-05-21
# Modified: 2026-05-22
# =============================================
"""Tests for handlers/notification/email.py."""
import json
from pathlib import Path
from unittest.mock import patch, MagicMock
from unittest.mock import patch
class TestEmailHandler:
@@ -46,7 +46,7 @@ class TestEmailHandler:
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email._speak"),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
):
result = handle({})
@@ -70,7 +70,7 @@ class TestEmailHandler:
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email._speak") as mock_speak,
patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
):
handle({})
@@ -92,7 +92,7 @@ class TestEmailHandler:
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email._speak") as mock_speak,
patch("aipass.hooks.apps.handlers.notification.email.speak") as mock_speak,
):
handle({})
@@ -114,7 +114,7 @@ class TestEmailHandler:
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email._speak"),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
):
result = handle({})
@@ -145,7 +145,7 @@ class TestEmailHandler:
"aipass.hooks.apps.handlers.notification.email._find_branch_root",
return_value=tmp_path,
),
patch("aipass.hooks.apps.handlers.notification.email._speak"),
patch("aipass.hooks.apps.handlers.notification.email.speak"),
):
result = handle({})
@@ -302,79 +302,3 @@ class TestFindBranchRoot:
result = _find_branch_root()
assert result is None
class TestSpeakFunction:
"""Piper TTS tests."""
def test_speak_calls_piper_then_aplay(self):
from aipass.hooks.apps.handlers.notification.email import _speak
with (
patch("aipass.hooks.apps.handlers.notification.email.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.email.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.email.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.email.tempfile") as mock_tmp,
patch("aipass.hooks.apps.handlers.notification.email.Path") as mock_path,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.return_value = MagicMock(returncode=0)
mock_path.return_value.exists.return_value = True
_speak("test text")
mock_sub.run.assert_called_once()
mock_sub.Popen.assert_called_once()
def test_speak_skips_when_piper_missing(self):
from aipass.hooks.apps.handlers.notification.email import _speak
with (
patch("aipass.hooks.apps.handlers.notification.email.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.email.subprocess") as mock_sub,
):
mock_piper_bin.exists.return_value = False
_speak("test")
mock_sub.run.assert_not_called()
def test_speak_graceful_on_timeout(self):
import subprocess as real_sub
from aipass.hooks.apps.handlers.notification.email import _speak
with (
patch("aipass.hooks.apps.handlers.notification.email.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.email.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.email.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.email.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.side_effect = real_sub.TimeoutExpired("piper", 5)
mock_sub.TimeoutExpired = real_sub.TimeoutExpired
_speak("test")
def test_speak_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.email import _speak
with (
patch("aipass.hooks.apps.handlers.notification.email.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.email.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.email.subprocess.run", side_effect=OSError("broken")),
patch("aipass.hooks.apps.handlers.notification.email.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
_speak("test")
src/aipass/hooks/tests/test_global_loader.py
+4 -4
View File
@@ -21,7 +21,7 @@ class TestGlobalLoaderHandler:
prompt = aipass_dir / "aipass_global_prompt.md"
prompt.write_text("# AIPass Global\nContext here", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.global_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
result = handle({})
@@ -32,7 +32,7 @@ class TestGlobalLoaderHandler:
def test_returns_empty_when_no_aipass_home(self):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
with patch("aipass.hooks.apps.handlers.prompt.global_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
with patch.dict("os.environ", {}, clear=True):
result = handle({})
@@ -42,7 +42,7 @@ class TestGlobalLoaderHandler:
def test_returns_empty_when_file_missing(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.global_loader import handle
with patch("aipass.hooks.apps.handlers.prompt.global_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
result = handle({})
@@ -56,7 +56,7 @@ class TestGlobalLoaderHandler:
aipass_dir.mkdir()
(aipass_dir / "aipass_global_prompt.md").write_text("content", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.global_loader._speak"):
with patch("aipass.hooks.apps.handlers.prompt.global_loader.speak"):
with patch.dict("os.environ", {"AIPASS_HOME": str(tmp_path)}):
result = handle({})
src/aipass/hooks/tests/test_hooksound.py
+94
View File
@@ -0,0 +1,94 @@
# =================== AIPass ====================
# Name: test_hooksound.py
# Version: 1.0.0
# Description: Tests for hooksound module (drone @hooks hooksound)
# Branch: hooks
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Tests for modules/hooksound.py — mute/unmute hook audio."""
from unittest.mock import patch
class TestHandleCommand:
"""Command routing tests."""
def test_returns_false_for_unknown_command(self):
from aipass.hooks.apps.modules.hooksound import handle_command
assert handle_command("unknown", []) is False
def test_routes_hooksound_command(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.is_muted", return_value=False):
assert handle_command("hooksound", []) is True
def test_off_creates_mute_flag(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.MUTE_FLAG") as mock_flag:
result = handle_command("hooksound", ["off"])
assert result is True
mock_flag.touch.assert_called_once()
def test_on_removes_mute_flag(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.MUTE_FLAG") as mock_flag:
mock_flag.exists.return_value = True
result = handle_command("hooksound", ["on"])
assert result is True
mock_flag.unlink.assert_called_once()
def test_on_noop_when_not_muted(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.MUTE_FLAG") as mock_flag:
mock_flag.exists.return_value = False
result = handle_command("hooksound", ["on"])
assert result is True
mock_flag.unlink.assert_not_called()
def test_status_shows_muted(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.is_muted", return_value=True):
assert handle_command("hooksound", []) is True
def test_status_shows_active(self):
from aipass.hooks.apps.modules.hooksound import handle_command
with patch("aipass.hooks.apps.modules.hooksound.is_muted", return_value=False):
assert handle_command("hooksound", []) is True
def test_help_flag(self):
from aipass.hooks.apps.modules.hooksound import handle_command
assert handle_command("hooksound", ["--help"]) is True
def test_help_word(self):
from aipass.hooks.apps.modules.hooksound import handle_command
assert handle_command("hooksound", ["help"]) is True
class TestPrintIntrospection:
"""Module introspection tests."""
def test_prints_without_error(self):
from aipass.hooks.apps.modules.hooksound import print_introspection
with patch("aipass.hooks.apps.modules.hooksound.is_muted", return_value=False):
print_introspection()
def test_shows_muted_status(self):
from aipass.hooks.apps.modules.hooksound import print_introspection
with patch("aipass.hooks.apps.modules.hooksound.is_muted", return_value=True):
print_introspection()
src/aipass/hooks/tests/test_identity.py
+8 -28
View File
@@ -11,7 +11,7 @@
import json
from pathlib import Path
from unittest.mock import patch, MagicMock
from unittest.mock import patch
SAMPLE_PASSPORT = {
@@ -40,7 +40,7 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps(SAMPLE_PASSPORT), encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
@@ -51,7 +51,7 @@ class TestIdentityHandler:
def test_returns_empty_when_no_passport(self, tmp_path):
from aipass.hooks.apps.handlers.prompt.identity import handle
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
@@ -67,7 +67,7 @@ class TestIdentityHandler:
nested = tmp_path / "apps" / "handlers"
nested.mkdir(parents=True)
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(nested)})
assert "devpulse Identity" in result["stdout"]
@@ -80,7 +80,7 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps(SAMPLE_PASSPORT), encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(tmp_path)})
out = result["stdout"]
@@ -100,7 +100,7 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text(json.dumps({"branch_info": {"branch_name": "test"}, "identity": {}}), encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
@@ -109,7 +109,7 @@ class TestIdentityHandler:
def test_empty_hook_data(self):
from aipass.hooks.apps.handlers.prompt.identity import handle
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
with patch("pathlib.Path.cwd", return_value=Path("/tmp/nonexistent")):
result = handle({})
@@ -124,28 +124,8 @@ class TestIdentityHandler:
passport = trinity / "passport.json"
passport.write_text("{broken json", encoding="utf-8")
with patch("aipass.hooks.apps.handlers.prompt.identity._speak"):
with patch("aipass.hooks.apps.handlers.prompt.identity.speak"):
result = handle({"cwd": str(tmp_path)})
assert result["exit_code"] == 0
assert result["stdout"] == ""
@patch("subprocess.Popen")
@patch("subprocess.run")
def test_piper_fires(self, mock_run, mock_popen):
from aipass.hooks.apps.handlers.prompt.identity import handle
mock_run.return_value = MagicMock(returncode=0)
with patch.object(Path, "exists", return_value=True):
handle({"cwd": "/tmp/nonexistent"})
assert mock_run.called or mock_popen.called
def test_piper_skips_when_not_available(self):
from aipass.hooks.apps.handlers.prompt.identity import handle
with patch("aipass.hooks.apps.handlers.prompt.identity.PIPER_BIN", Path("/nonexistent/piper")):
result = handle({"cwd": "/tmp/nonexistent"})
assert result["exit_code"] == 0
src/aipass/hooks/tests/test_rollover.py
+3 -3
View File
@@ -17,7 +17,7 @@ class TestRolloverHandler:
def test_no_repo_root_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=None):
result = handle({})
@@ -27,7 +27,7 @@ class TestRolloverHandler:
def test_no_overdue_returns_empty(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue", return_value=[]):
result = handle({})
@@ -38,7 +38,7 @@ class TestRolloverHandler:
def test_overdue_triggers_rollover(self):
from aipass.hooks.apps.handlers.lifecycle.rollover import handle
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover.speak"):
with patch("aipass.hooks.apps.handlers.lifecycle.rollover._find_repo_root", return_value=MagicMock()):
with patch(
"aipass.hooks.apps.handlers.lifecycle.rollover._find_overdue",
src/aipass/hooks/tests/test_sound.py
+177
View File
@@ -0,0 +1,177 @@
# =================== AIPass ====================
# Name: test_sound.py
# Version: 1.0.0
# Description: Tests for shared sound module
# Branch: hooks
# Created: 2026-05-22
# Modified: 2026-05-22
# =============================================
"""Tests for apps/sound.py — shared speak/play with mute support."""
from pathlib import Path
from unittest.mock import patch, MagicMock
class TestIsMuted:
"""Mute flag detection."""
def test_not_muted_when_flag_missing(self):
from aipass.hooks.apps.sound import is_muted
with patch("aipass.hooks.apps.sound.MUTE_FLAG") as mock_flag:
mock_flag.exists.return_value = False
assert is_muted() is False
def test_muted_when_flag_exists(self):
from aipass.hooks.apps.sound import is_muted
with patch("aipass.hooks.apps.sound.MUTE_FLAG") as mock_flag:
mock_flag.exists.return_value = True
assert is_muted() is True
class TestSpeak:
"""Piper TTS with mute support."""
def test_speak_calls_piper_when_not_muted(self):
from aipass.hooks.apps.sound import speak
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.PIPER_BIN") as mock_bin,
patch("aipass.hooks.apps.sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.sound.tempfile") as mock_tmp,
patch("aipass.hooks.apps.sound.Path") as mock_path,
):
mock_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.return_value = MagicMock(returncode=0)
mock_path.return_value.exists.return_value = True
speak("test text")
mock_sub.run.assert_called_once()
mock_sub.Popen.assert_called_once()
def test_speak_skips_when_muted(self):
from aipass.hooks.apps.sound import speak
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=True),
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
):
speak("test")
mock_sub.run.assert_not_called()
def test_speak_skips_when_piper_missing(self):
from aipass.hooks.apps.sound import speak
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.PIPER_BIN") as mock_bin,
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
):
mock_bin.exists.return_value = False
speak("test")
mock_sub.run.assert_not_called()
def test_speak_graceful_on_timeout(self):
import subprocess as real_sub
from aipass.hooks.apps.sound import speak
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.PIPER_BIN") as mock_bin,
patch("aipass.hooks.apps.sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.sound.tempfile") as mock_tmp,
):
mock_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.side_effect = real_sub.TimeoutExpired("piper", 5)
mock_sub.TimeoutExpired = real_sub.TimeoutExpired
speak("test")
def test_speak_graceful_on_os_error(self):
from aipass.hooks.apps.sound import speak
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.PIPER_BIN") as mock_bin,
patch("aipass.hooks.apps.sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.sound.subprocess.run", side_effect=OSError("broken")),
patch("aipass.hooks.apps.sound.tempfile") as mock_tmp,
):
mock_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
speak("test")
class TestPlay:
"""WAV playback with mute support."""
def test_play_calls_aplay_when_not_muted(self):
from aipass.hooks.apps.sound import play
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
):
mock_path = MagicMock()
mock_path.exists.return_value = True
play(mock_path)
mock_sub.Popen.assert_called_once()
def test_play_skips_when_muted(self):
from aipass.hooks.apps.sound import play
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=True),
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
):
play(Path("/tmp/sound.wav"))
mock_sub.Popen.assert_not_called()
def test_play_skips_when_file_missing(self):
from aipass.hooks.apps.sound import play
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.subprocess") as mock_sub,
):
mock_path = MagicMock()
mock_path.exists.return_value = False
play(mock_path)
mock_sub.Popen.assert_not_called()
def test_play_graceful_on_os_error(self):
from aipass.hooks.apps.sound import play
with (
patch("aipass.hooks.apps.sound.is_muted", return_value=False),
patch("aipass.hooks.apps.sound.subprocess.Popen", side_effect=OSError("no aplay")),
):
mock_path = MagicMock()
mock_path.exists.return_value = True
play(mock_path)
src/aipass/hooks/tests/test_stop_sound.py
+6 -140
View File
@@ -1,15 +1,15 @@
# =================== AIPass ====================
# Name: test_stop_sound.py
# Version: 1.1.0
# Version: 1.2.0
# Description: Tests for stop_sound notification handler
# Branch: hooks
# Created: 2026-05-20
# Modified: 2026-05-20
# Modified: 2026-05-22
# =============================================
"""Tests for handlers/notification/stop_sound.py."""
from unittest.mock import patch, MagicMock
from unittest.mock import patch
class TestStopSoundHandler:
@@ -18,10 +18,7 @@ class TestStopSoundHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound._play"),
patch("aipass.hooks.apps.handlers.notification.stop_sound._speak"),
):
with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak"):
result = handle({})
assert isinstance(result, dict)
@@ -31,147 +28,16 @@ class TestStopSoundHandler:
def test_handle_speaks_stop_sound(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound._play"),
patch("aipass.hooks.apps.handlers.notification.stop_sound._speak") as mock_speak,
):
with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak") as mock_speak:
handle({})
mock_speak.assert_called_once_with("stop sound")
def test_handle_does_not_play_wav(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound._play") as mock_play,
patch("aipass.hooks.apps.handlers.notification.stop_sound._speak"),
):
handle({})
mock_play.assert_not_called()
def test_handle_skips_when_stop_hook_active(self):
from aipass.hooks.apps.handlers.notification.stop_sound import handle
with patch("aipass.hooks.apps.handlers.notification.stop_sound._speak") as mock_speak:
with patch("aipass.hooks.apps.handlers.notification.stop_sound.speak") as mock_speak:
result = handle({"stop_hook_active": True})
mock_speak.assert_not_called()
assert result["exit_code"] == 0
class TestPlayFunction:
"""WAV playback tests."""
def test_play_calls_aplay(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _play
mock_path = MagicMock()
mock_path.exists.return_value = True
with patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess.Popen") as mock_popen:
_play(mock_path)
mock_popen.assert_called_once()
args = mock_popen.call_args[0][0]
assert args[0] == "aplay"
assert args[1] == "-q"
def test_play_skips_when_file_missing(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _play
mock_path = MagicMock()
mock_path.exists.return_value = False
with patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess.Popen") as mock_popen:
_play(mock_path)
mock_popen.assert_not_called()
def test_play_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _play
mock_path = MagicMock()
mock_path.exists.return_value = True
with patch(
"aipass.hooks.apps.handlers.notification.stop_sound.subprocess.Popen",
side_effect=OSError("broken"),
):
_play(mock_path)
class TestSpeakFunction:
"""Piper TTS tests."""
def test_speak_calls_piper_then_aplay(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.stop_sound.tempfile") as mock_tmp,
patch("aipass.hooks.apps.handlers.notification.stop_sound.Path") as mock_path,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.return_value = MagicMock(returncode=0)
mock_path.return_value.exists.return_value = True
_speak("test text")
mock_sub.run.assert_called_once()
mock_sub.Popen.assert_called_once()
def test_speak_skips_when_piper_missing(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess") as mock_sub,
):
mock_piper_bin.exists.return_value = False
_speak("test")
mock_sub.run.assert_not_called()
def test_speak_graceful_on_timeout(self):
import subprocess as real_sub
from aipass.hooks.apps.handlers.notification.stop_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.stop_sound.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.side_effect = real_sub.TimeoutExpired("piper", 5)
mock_sub.TimeoutExpired = real_sub.TimeoutExpired
_speak("test")
def test_speak_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.stop_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.stop_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.stop_sound.subprocess.run", side_effect=OSError("broken")),
patch("aipass.hooks.apps.handlers.notification.stop_sound.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
_speak("test")
src/aipass/hooks/tests/test_subagent_gate.py
+8 -30
View File
@@ -18,14 +18,14 @@ from aipass.hooks.apps.handlers.security.subagent_gate import handle
class TestSubagentGateHandler:
def test_no_repo_root_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._speak"):
with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
result = handle({"cwd": "/tmp/nowhere"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
def test_no_modified_files_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._speak"):
with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
result = handle({"cwd": "/tmp/somewhere"})
assert result["exit_code"] == 0
assert result["stdout"] == ""
@@ -34,7 +34,7 @@ class TestSubagentGateHandler:
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._speak")
@patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
def test_modified_files_no_violations_allows(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
@@ -48,7 +48,7 @@ class TestSubagentGateHandler:
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._speak")
@patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
def test_violations_blocks(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
@@ -64,7 +64,7 @@ class TestSubagentGateHandler:
assert "bad.py" in parsed["reason"]
@patch("subprocess.run")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._speak")
@patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
def test_skip_claude_hooks_from_modified_files(self, mock_speak, mock_run, tmp_path):
src = tmp_path / "src" / "aipass" / "hooks"
@@ -91,7 +91,7 @@ class TestSubagentGateHandler:
@patch("aipass.hooks.apps.handlers.security.subagent_gate._run_seedgo_checklist", return_value=[])
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._speak")
@patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
def test_readme_accountability_advisory(self, mock_speak, mock_root, mock_modified, mock_seedgo, mock_readme):
from pathlib import Path
@@ -107,38 +107,16 @@ class TestSubagentGateHandler:
assert parsed["decision"] == "allow"
assert "README" in parsed["reason"]
@patch("subprocess.Popen")
@patch("subprocess.run")
def test_piper_fires_when_available(self, mock_run, mock_popen):
from pathlib import Path
from aipass.hooks.apps.handlers.security.subagent_gate import _speak
mock_run.return_value = MagicMock(returncode=0)
with patch("aipass.hooks.apps.handlers.security.subagent_gate.PIPER_BIN", Path("/fake/piper")):
with patch("aipass.hooks.apps.handlers.security.subagent_gate.PIPER_VOICE", Path("/fake/voice.onnx")):
with patch("pathlib.Path.exists", return_value=True):
_speak("test")
mock_popen.assert_called_once()
@patch("subprocess.Popen")
def test_piper_skips_when_not_available(self, mock_popen):
from pathlib import Path
from aipass.hooks.apps.handlers.security.subagent_gate import _speak
with patch("aipass.hooks.apps.handlers.security.subagent_gate.PIPER_BIN", Path("/nonexistent/piper")):
_speak("test")
mock_popen.assert_not_called()
def test_empty_hook_data_allows(self):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root", return_value=None):
with patch("aipass.hooks.apps.handlers.security.subagent_gate._speak"):
with patch("aipass.hooks.apps.handlers.security.subagent_gate.speak"):
result = handle({})
assert result["exit_code"] == 0
assert result["stdout"] == ""
@patch("aipass.hooks.apps.handlers.security.subagent_gate._get_modified_py_files")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._find_repo_root")
@patch("aipass.hooks.apps.handlers.security.subagent_gate._speak")
@patch("aipass.hooks.apps.handlers.security.subagent_gate.speak")
def test_exception_in_get_modified_allows(self, mock_speak, mock_root, mock_modified):
from pathlib import Path
src/aipass/hooks/tests/test_tool_sound.py
+7 -83
View File
@@ -1,15 +1,15 @@
# =================== AIPass ====================
# Name: test_tool_sound.py
# Version: 1.1.0
# Version: 1.2.0
# Description: Tests for tool_sound notification handler
# Branch: hooks
# Created: 2026-05-19
# Modified: 2026-05-19
# Modified: 2026-05-22
# =============================================
"""Tests for handlers/notification/tool_sound.py."""
from unittest.mock import patch, MagicMock
from unittest.mock import patch
class TestToolSoundHandler:
@@ -18,7 +18,7 @@ class TestToolSoundHandler:
def test_handle_returns_result_dict(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
with patch("aipass.hooks.apps.handlers.notification.tool_sound._speak"):
with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak"):
result = handle({"tool_name": "Bash"})
assert isinstance(result, dict)
@@ -30,7 +30,7 @@ class TestToolSoundHandler:
def test_speaks_tool_name(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
with patch("aipass.hooks.apps.handlers.notification.tool_sound._speak") as mock_speak:
with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
handle({"tool_name": "Edit"})
mock_speak.assert_called_once_with("tool sound: Edit")
@@ -38,7 +38,7 @@ class TestToolSoundHandler:
def test_no_speak_when_no_tool_name(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
with patch("aipass.hooks.apps.handlers.notification.tool_sound._speak") as mock_speak:
with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
handle({})
mock_speak.assert_not_called()
@@ -46,83 +46,7 @@ class TestToolSoundHandler:
def test_no_speak_when_empty_tool_name(self):
from aipass.hooks.apps.handlers.notification.tool_sound import handle
with patch("aipass.hooks.apps.handlers.notification.tool_sound._speak") as mock_speak:
with patch("aipass.hooks.apps.handlers.notification.tool_sound.speak") as mock_speak:
handle({"tool_name": ""})
mock_speak.assert_not_called()
class TestSpeakFunction:
"""Piper TTS integration tests."""
def test_speak_calls_piper_then_aplay(self):
from aipass.hooks.apps.handlers.notification.tool_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.tool_sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.tool_sound.tempfile") as mock_tmp,
patch("aipass.hooks.apps.handlers.notification.tool_sound.Path") as mock_path,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.return_value = MagicMock(returncode=0)
mock_path.return_value.exists.return_value = True
_speak("test text")
mock_sub.run.assert_called_once()
mock_sub.Popen.assert_called_once()
def test_speak_skips_when_piper_missing(self):
from aipass.hooks.apps.handlers.notification.tool_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.tool_sound.subprocess") as mock_sub,
):
mock_piper_bin.exists.return_value = False
_speak("test")
mock_sub.run.assert_not_called()
def test_speak_graceful_on_timeout(self):
import subprocess as real_sub
from aipass.hooks.apps.handlers.notification.tool_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.tool_sound.subprocess") as mock_sub,
patch("aipass.hooks.apps.handlers.notification.tool_sound.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
mock_sub.run.side_effect = real_sub.TimeoutExpired("piper", 5)
mock_sub.TimeoutExpired = real_sub.TimeoutExpired
_speak("test")
def test_speak_graceful_on_os_error(self):
from aipass.hooks.apps.handlers.notification.tool_sound import _speak
with (
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_BIN") as mock_piper_bin,
patch("aipass.hooks.apps.handlers.notification.tool_sound.PIPER_VOICE") as mock_voice,
patch("aipass.hooks.apps.handlers.notification.tool_sound.subprocess.run", side_effect=OSError("broken")),
patch("aipass.hooks.apps.handlers.notification.tool_sound.tempfile") as mock_tmp,
):
mock_piper_bin.exists.return_value = True
mock_voice.exists.return_value = True
mock_file = MagicMock()
mock_file.name = "/tmp/test.wav"
mock_tmp.NamedTemporaryFile.return_value = mock_file
_speak("test")