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

Merge pull request #469 from AIOSAI/system/devpulse-testdrone-improve-line-coverage-on-entry-point-reg

Browse Source 
feat(system): test(drone): improve line coverage on entry point + registry
This commit is contained in:
AIPass
2026-04-26 07:59:34 -07:00
committed by GitHub
parent 10547fd9ca 92e14308b2
commit bdc0c6c421
12 changed files with 5073 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/ai_mail/.seedgo/bypass.json
+10
View File
@@ -359,6 +359,16 @@
"file": "apps/modules/email_send.py",
"standard": "modules",
"reason": "Internal helper extracted from email.py for size compliance. Not a drone-routable command module — no handle_command() needed."
},
{
"file": "tests/test_daemon.py",
"standard": "trigger",
"reason": "Test cleanup in finally blocks — .unlink() removes temp files created in ~/.claude/projects during _set_session_name tests. Not production file deletion."
},
{
"file": "tests/test_email_module.py",
"standard": "architecture",
"reason": "Test file lives in tests/ directory — not subject to 3-layer app structure rule."
}
],
"notes": {
src/aipass/ai_mail/tests/test_daemon.py
+1140
View File
File diff suppressed because it is too large Load Diff
src/aipass/ai_mail/tests/test_dispatch_module.py
+952
View File
@@ -0,0 +1,952 @@
# =================== AIPass ====================
# Name: test_dispatch_module.py
# Description: Tests for dispatch.py orchestrator functions
# Version: 1.0.0
# Created: 2026-04-26
# Modified: 2026-04-26
# =============================================
"""Tests for dispatch.py orchestrator functions.
Covers: print_help, handle_command, _orchestrate_status,
_orchestrate_wake, _orchestrate_dispatch_send, _orchestrate_daemon,
print_introspection.
All handler dependencies are mocked -- these tests verify orchestration
logic, not business logic.
"""
from contextlib import ExitStack
import pytest
from unittest.mock import MagicMock, patch
# ---------------------------------------------------------------------------
# Autouse fixture: suppress json_handler.log_operation
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _silence_json_handler():
"""Prevent log_operation from writing real JSON files during tests."""
with patch("aipass.ai_mail.apps.modules.dispatch.json_handler") as mock_jh:
mock_jh.log_operation.return_value = True
yield mock_jh
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
MOD = "aipass.ai_mail.apps.modules.dispatch"
def _mock_console(printed: list[str]) -> MagicMock:
"""Create a mock console that appends all print calls to *printed*."""
mc = MagicMock()
mc.print = lambda msg="", **kw: printed.append(str(msg))
return mc
# ===========================================================================
# print_help
# ===========================================================================
class TestPrintHelp:
"""Tests for dispatch.print_help."""
def test_print_help_contains_keywords(self, monkeypatch):
"""Help text contains expected keywords."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import print_help
print_help()
combined = " ".join(printed)
assert "dispatch" in combined.lower()
assert "status" in combined.lower()
assert "daemon" in combined.lower()
assert "wake" in combined.lower()
# ===========================================================================
# handle_command
# ===========================================================================
class TestHandleCommand:
"""Tests for the top-level handle_command router."""
def test_non_dispatch_command_returns_false(self):
"""A command that is not 'dispatch' returns False."""
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("email", ["inbox"])
assert result is False
def test_dispatch_no_args_calls_introspection(self, monkeypatch):
"""'dispatch' with no args calls print_introspection."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", [])
assert result is True
combined = " ".join(printed)
assert "dispatch Module" in combined
def test_dispatch_help_flag(self, monkeypatch):
"""'dispatch --help' prints help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["--help"])
assert result is True
combined = " ".join(printed)
assert "COMMANDS" in combined
def test_dispatch_h_flag(self, monkeypatch):
"""'dispatch -h' prints help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["-h"])
assert result is True
def test_dispatch_help_word(self, monkeypatch):
"""'dispatch help' prints help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["help"])
assert result is True
def test_dispatch_status_subcommand(self, monkeypatch):
"""'dispatch status' delegates to _orchestrate_status."""
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: [])
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["status"])
assert result is True
def test_dispatch_daemon_subcommand(self, monkeypatch):
"""'dispatch daemon' delegates to _orchestrate_daemon."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
monkeypatch.setattr(
f"{MOD}._orchestrate_daemon",
lambda: True,
)
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["daemon"])
assert result is True
def test_dispatch_wake_subcommand(self, monkeypatch):
"""'dispatch wake @branch' delegates to _orchestrate_wake."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
# No further args -> shows help
result = handle_command("dispatch", ["wake"])
assert result is True
def test_dispatch_at_target(self, monkeypatch):
"""'dispatch @target Subject Body' routes to _orchestrate_dispatch_send."""
monkeypatch.setattr(
f"{MOD}._orchestrate_dispatch_send",
lambda args: True,
)
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["@branch", "Subject", "Body"])
assert result is True
def test_dispatch_path_target(self, monkeypatch):
"""'dispatch /path Subject Body' routes to _orchestrate_dispatch_send."""
monkeypatch.setattr(
f"{MOD}._orchestrate_dispatch_send",
lambda args: True,
)
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["/some/path", "Subject", "Body"])
assert result is True
def test_dispatch_unknown_subcommand(self, monkeypatch):
"""Unknown subcommand prints error and returns False."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import handle_command
result = handle_command("dispatch", ["bogus"])
assert result is False
assert any("Unknown" in e for e in errors)
# ===========================================================================
# _orchestrate_status
# ===========================================================================
class TestOrchestrateStatus:
"""Tests for _orchestrate_status."""
def test_no_dispatches_prints_empty(self, monkeypatch):
"""No dispatches prints 'No dispatches recorded yet.'."""
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: [])
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
assert any("No dispatches" in p for p in printed)
def test_running_status_display(self, monkeypatch):
"""A spawned dispatch with a running PID shows RUNNING."""
dispatches = [
{
"branch": "@alpha",
"pid": 12345,
"timestamp": "2026-04-25T10:00:00",
"status": "spawned",
},
]
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(f"{MOD}.check_pid_status", lambda pid: "RUNNING")
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "5m ago")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "RUNNING" in combined
assert "@alpha" in combined
assert "Active: 1" in combined
def test_completed_status_display(self, monkeypatch):
"""A spawned dispatch with a completed PID shows COMPLETED."""
dispatches = [
{
"branch": "@beta",
"pid": 99999,
"timestamp": "2026-04-25T09:00:00",
"status": "spawned",
},
]
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(f"{MOD}.check_pid_status", lambda pid: "COMPLETED")
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "1h ago")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "COMPLETED" in combined
assert "Active: 0" in combined
def test_failed_status_display(self, monkeypatch):
"""A dispatch with status 'failed' shows FAILED."""
dispatches = [
{
"branch": "@gamma",
"pid": None,
"timestamp": "2026-04-25T08:00:00",
"status": "failed",
},
]
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "2h ago")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "FAILED" in combined
assert "NO PID" in combined
def test_unknown_status_display(self, monkeypatch):
"""A dispatch with unknown status shows yellow UNKNOWN."""
dispatches = [
{
"branch": "@delta",
"pid": None,
"timestamp": "2026-04-25T07:00:00",
"status": "weird",
},
]
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "3h ago")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "UNKNOWN" in combined
def test_multiple_dispatches_shows_active_count(self, monkeypatch):
"""Multiple dispatches shows correct active count."""
dispatches = [
{"branch": "@a", "pid": 100, "timestamp": "t1", "status": "spawned"},
{"branch": "@b", "pid": 200, "timestamp": "t2", "status": "spawned"},
{"branch": "@c", "pid": 300, "timestamp": "t3", "status": "spawned"},
]
pid_map = {100: "RUNNING", 200: "COMPLETED", 300: "RUNNING"}
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(
f"{MOD}.check_pid_status",
lambda pid: pid_map.get(pid, "UNKNOWN"),
)
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "0m")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "Active: 2" in combined
assert "Total: 3" in combined
def test_more_than_five_dispatches_shows_last_five(self, monkeypatch):
"""Only the last 5 dispatches are shown."""
dispatches = [
{
"branch": f"@b{i}",
"pid": None,
"timestamp": f"t{i}",
"status": "failed",
}
for i in range(8)
]
monkeypatch.setattr(f"{MOD}.load_dispatch_log", lambda: dispatches)
monkeypatch.setattr(f"{MOD}.calculate_age", lambda ts: "0m")
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_status
result = _orchestrate_status()
assert result is True
combined = " ".join(printed)
assert "Total: 5" in combined
# The first 3 (b0, b1, b2) should NOT appear
assert "@b0" not in combined
assert "@b1" not in combined
assert "@b2" not in combined
# The last 5 (b3..b7) should appear
assert "@b7" in combined
assert "@b3" in combined
# ===========================================================================
# _orchestrate_wake
# ===========================================================================
class TestOrchestrateWake:
"""Tests for _orchestrate_wake."""
def test_no_args_prints_help(self, monkeypatch):
"""No args prints wake help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake([])
assert result is True
combined = " ".join(printed)
assert "Wake" in combined
def test_help_flag_prints_help(self, monkeypatch):
"""--help prints wake help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["--help"])
assert result is True
combined = " ".join(printed)
assert "Wake" in combined
def test_h_flag_prints_help(self, monkeypatch):
"""-h prints wake help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["-h"])
assert result is True
def test_help_word_prints_help(self, monkeypatch):
"""'help' prints wake help and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["help"])
assert result is True
def test_missing_branch_after_flags_returns_false(self, monkeypatch):
"""Only flags (--fresh) without a branch returns False."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["--fresh"])
assert result is False
assert any("Missing" in e for e in errors)
def test_blocked_branch_shows_error(self, monkeypatch):
"""A blocked branch shows error and returns True."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=True,
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["@protected"])
assert result is True
assert any("protected" in e for e in errors)
def test_successful_wake(self, monkeypatch):
"""Successful wake prints status and returns True."""
mock_status = MagicMock()
mock_status.format.return_value = "WAKE OK: @branch woke up"
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
return_value=(mock_status, True),
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["@branch"])
assert result is True
combined = " ".join(printed)
assert "WAKE OK" in combined
def test_failed_wake_returns_false(self, monkeypatch):
"""Failed wake returns False (wake_branch returns success=False)."""
mock_status = MagicMock()
mock_status.format.return_value = "WAKE FAILED: spawn error"
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
return_value=(mock_status, False),
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
result = _orchestrate_wake(["@branch"])
assert result is False
def test_fresh_flag(self, monkeypatch):
"""--fresh flag is passed through to wake_branch."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Capture wake_branch call arguments."""
wake_calls.append({"branch": branch, "fresh": fresh, "sender": sender, "model": model})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
side_effect=mock_wake,
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
_orchestrate_wake(["--fresh", "@branch"])
assert len(wake_calls) == 1
assert wake_calls[0]["fresh"] is True
def test_model_flag(self, monkeypatch):
"""--model flag is passed through to wake_branch."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Track model argument passed to wake_branch."""
wake_calls.append({"branch": branch, "model": model})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
side_effect=mock_wake,
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
_orchestrate_wake(["--model", "opus", "@branch"])
assert len(wake_calls) == 1
assert wake_calls[0]["model"] == "opus"
def test_sender_flag(self, monkeypatch):
"""--sender flag is passed through to wake_branch."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Track sender argument passed to wake_branch."""
wake_calls.append({"branch": branch, "sender": sender})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
side_effect=mock_wake,
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
_orchestrate_wake(["--sender", "@custom", "@branch"])
assert len(wake_calls) == 1
assert wake_calls[0]["sender"] == "@custom"
def test_custom_message(self, monkeypatch):
"""A custom message after the branch is passed to wake_branch."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Track custom message argument passed to wake_branch."""
wake_calls.append({"branch": branch, "msg": msg})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.is_wake_blocked",
return_value=False,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch",
side_effect=mock_wake,
),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_wake
_orchestrate_wake(["@branch", "Check your inbox now"])
assert len(wake_calls) == 1
assert wake_calls[0]["msg"] == "Check your inbox now"
# ===========================================================================
# _orchestrate_dispatch_send
# ===========================================================================
def _send_patches(overrides: dict | None = None):
"""Return a dict of default patch targets for _orchestrate_dispatch_send.
Call with overrides to replace specific mocks.
"""
mock_status = MagicMock()
mock_status.format.return_value = "WAKE OK"
defaults = {
f"{MOD}.resolve_sender_info": MagicMock(return_value={"email_address": "@ai_mail"}),
f"{MOD}.prepend_dispatch_header": MagicMock(return_value="[DISPATCH] Body"),
f"{MOD}.send_to_single": MagicMock(return_value=(True, None)),
f"{MOD}.on_email_delivered": MagicMock(),
f"{MOD}.push_dashboard_update": MagicMock(),
f"{MOD}.get_current_user": MagicMock(return_value={"name": "test"}),
f"{MOD}.get_branch_by_email": MagicMock(return_value={"email": "@target"}),
f"{MOD}.update_central": MagicMock(),
f"{MOD}.create_email_file": MagicMock(),
f"{MOD}.load_email_file": MagicMock(),
f"{MOD}.deliver_email_to_branch": MagicMock(),
f"{MOD}.dispatch_send_error": MagicMock(),
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch": MagicMock(return_value=(mock_status, True)),
"aipass.trigger.apps.modules.core.trigger": MagicMock(),
}
if overrides:
defaults.update(overrides)
return defaults
class TestOrchestrateDispatchSend:
"""Tests for _orchestrate_dispatch_send."""
def test_too_few_args_shows_usage(self, monkeypatch):
"""Fewer than 3 args prints usage error and returns True."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_dispatch_send
result = _orchestrate_dispatch_send(["@target", "Subject"])
assert result is True
assert any("Usage" in e for e in errors)
def test_successful_send_and_wake(self, monkeypatch):
"""Successful send + wake returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches()
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is True
combined = " ".join(printed)
assert "sent" in combined.lower()
def test_send_failure_calls_dispatch_send_error(self, monkeypatch):
"""Send failure calls dispatch_send_error and returns False."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
dispatch_error_calls: list[tuple] = []
mock_dispatch_err = MagicMock(side_effect=lambda *a: dispatch_error_calls.append(a))
patches = _send_patches(
{
f"{MOD}.send_to_single": MagicMock(return_value=(False, "Branch not found")),
f"{MOD}.dispatch_send_error": mock_dispatch_err,
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is False
assert any("Send failed" in e for e in errors)
assert len(dispatch_error_calls) == 1
def test_send_ok_but_wake_failure_shows_warning(self, monkeypatch):
"""Send succeeds but wake fails -- returns True but shows error."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
mock_status = MagicMock()
mock_status.format.return_value = "WAKE FAILED"
patches = _send_patches(
{
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch": MagicMock(return_value=(mock_status, False)),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is True
assert any("wake failed" in e.lower() for e in errors)
def test_fresh_flag_passed_through(self, monkeypatch):
"""--fresh flag is passed to wake_branch as fresh=True."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Track fresh flag passed to wake_branch."""
wake_calls.append({"branch": branch, "fresh": fresh})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches(
{
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch": MagicMock(side_effect=mock_wake),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
_orchestrate_dispatch_send(["@target", "Subject", "Body", "--fresh"])
assert len(wake_calls) == 1
assert wake_calls[0]["fresh"] is True
def test_from_flag_passed_through(self, monkeypatch):
"""--from flag is passed to resolve_sender_info."""
sender_calls: list[str | None] = []
def tracking_resolve(from_branch, *args):
"""Track from_branch argument passed to resolve_sender_info."""
sender_calls.append(from_branch)
return {"email_address": "@custom"}
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches(
{
f"{MOD}.resolve_sender_info": MagicMock(side_effect=tracking_resolve),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
_orchestrate_dispatch_send(["@target", "Subject", "Body", "--from", "@custom_sender"])
assert len(sender_calls) == 1
assert sender_calls[0] == "@custom_sender"
def test_model_flag(self, monkeypatch):
"""--model flag is passed to wake_branch."""
wake_calls: list[dict] = []
mock_status = MagicMock()
mock_status.format.return_value = "OK"
def mock_wake(branch, msg=None, fresh=False, sender="@devpulse", model=None):
"""Track model argument passed to wake_branch in dispatch send."""
wake_calls.append({"model": model})
return (mock_status, True)
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches(
{
"aipass.ai_mail.apps.handlers.dispatch.wake.wake_branch": MagicMock(side_effect=mock_wake),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
_orchestrate_dispatch_send(["@target", "Subject", "Body", "--model", "sonnet"])
assert len(wake_calls) == 1
assert wake_calls[0]["model"] == "sonnet"
def test_no_memory_save_flag(self, monkeypatch):
"""--no-memory-save flag is passed to prepend_dispatch_header."""
header_calls: list[dict] = []
def tracking_header(body, no_memory_save=False):
"""Track no_memory_save flag passed to prepend_dispatch_header."""
header_calls.append({"no_memory_save": no_memory_save})
return f"[DISPATCH] {body}"
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches(
{
f"{MOD}.prepend_dispatch_header": MagicMock(side_effect=tracking_header),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
_orchestrate_dispatch_send(["@target", "Subject", "Body", "--no-memory-save"])
assert len(header_calls) == 1
assert header_calls[0]["no_memory_save"] is True
def test_trigger_fire_failure_does_not_fail(self, monkeypatch):
"""If trigger.fire raises, the send still succeeds."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
mock_trigger = MagicMock()
mock_trigger.fire.side_effect = RuntimeError("trigger broken")
patches = _send_patches(
{
"aipass.trigger.apps.modules.core.trigger": mock_trigger,
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is True
def test_send_phase_exception_returns_false(self, monkeypatch):
"""Exception during send phase returns False."""
errors: list[str] = []
monkeypatch.setattr(f"{MOD}.error", lambda msg: errors.append(msg))
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
patches = _send_patches(
{
f"{MOD}.resolve_sender_info": MagicMock(side_effect=RuntimeError("boom")),
}
)
with patch.multiple("", **{k: v for k, v in patches.items()}):
from aipass.ai_mail.apps.modules.dispatch import (
_orchestrate_dispatch_send,
)
result = _orchestrate_dispatch_send(["@target", "Subject", "Body"])
assert result is False
assert any("Send failed" in e for e in errors)
# ===========================================================================
# _orchestrate_daemon
# ===========================================================================
class TestOrchestrateDaemon:
"""Tests for _orchestrate_daemon."""
def test_calls_run_daemon(self, monkeypatch):
"""_orchestrate_daemon calls run_daemon and returns True."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
daemon_called: list[bool] = []
with patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.run_daemon",
side_effect=lambda: daemon_called.append(True),
):
from aipass.ai_mail.apps.modules.dispatch import _orchestrate_daemon
result = _orchestrate_daemon()
assert result is True
assert len(daemon_called) == 1
combined = " ".join(printed)
assert "daemon" in combined.lower()
# ===========================================================================
# print_introspection
# ===========================================================================
class TestPrintIntrospection:
"""Tests for print_introspection."""
def test_prints_module_info(self, monkeypatch):
"""print_introspection prints module info."""
printed: list[str] = []
monkeypatch.setattr(f"{MOD}.console", _mock_console(printed))
from aipass.ai_mail.apps.modules.dispatch import print_introspection
print_introspection()
combined = " ".join(printed)
assert "dispatch Module" in combined
assert "Connected Handlers" in combined
assert "status.py" in combined
assert "wake.py" in combined
assert "daemon.py" in combined
src/aipass/ai_mail/tests/test_email_module.py
+729
View File
@@ -1032,3 +1032,732 @@ class TestHandleCommand:
result = handle_command("inbox", ["help"])
assert result is True
assert any("Email Module" in p for p in printed)
# ###########################################################################
# NEW COVERAGE TESTS — appended for untested paths in email.py & email_send.py
# ###########################################################################
# ===========================================================================
# _resolve_branch_path RuntimeError fallback (email.py line 67-69)
# ===========================================================================
class TestResolveBranchPath:
"""Tests for _resolve_branch_path fallback behaviour."""
def test_runtime_error_fallback(self, monkeypatch):
"""When get_current_user raises RuntimeError, falls back to _AI_MAIL_DIR."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.get_current_user",
lambda: (_ for _ in ()).throw(RuntimeError("no branch")),
)
from aipass.ai_mail.apps.modules.email import _resolve_branch_path, _AI_MAIL_DIR
result = _resolve_branch_path()
assert result == _AI_MAIL_DIR
# ===========================================================================
# handle_inbox additional paths
# ===========================================================================
class TestHandleInboxExtended:
"""Extended tests for handle_inbox edge cases."""
def test_inbox_nonexistent_file(self, tmp_path, monkeypatch):
"""When inbox_file does not exist, prints 'empty' (line 156-158)."""
non_existent = tmp_path / ".ai_mail.local" / "inbox.json"
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.resolve_inbox_target",
lambda first_arg, repo_root, get_branch_fn, get_user_fn: (
True,
{
"inbox_file": non_existent,
"display_name": "TEST",
"target_branch": None,
"error": None,
},
),
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
from aipass.ai_mail.apps.modules.email import handle_inbox
result = handle_inbox([])
assert result is True
assert any("empty" in p.lower() for p in printed)
def test_inbox_with_target_branch_label(self, tmp_path, monkeypatch):
"""When resolve returns target_branch, label shows 'for @target (NAME)'."""
messages = [{"id": "m1", "status": "new", "subject": "Hello"}]
_write_inbox(tmp_path, messages=messages)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.resolve_inbox_target",
lambda first_arg, repo_root, get_branch_fn, get_user_fn: (
True,
{
"inbox_file": tmp_path / ".ai_mail.local" / "inbox.json",
"display_name": "ALPHA",
"target_branch": "@alpha",
"error": None,
},
),
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.load_inbox",
lambda f: {"messages": messages},
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.format_email_list_item",
lambda i, msg, show_unread=True: f"[{i}] {msg['subject']}",
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
from aipass.ai_mail.apps.modules.email import handle_inbox
result = handle_inbox(["@alpha"])
assert result is True
assert any("for @alpha (ALPHA)" in p for p in printed)
def test_inbox_broken_pipe(self, monkeypatch):
"""BrokenPipeError is caught and returns True (line 173-175)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.resolve_inbox_target",
lambda first_arg, repo_root, get_branch_fn, get_user_fn: (_ for _ in ()).throw(
BrokenPipeError("pipe closed")
),
)
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: None
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
from aipass.ai_mail.apps.modules.email import handle_inbox
result = handle_inbox([])
assert result is True
def test_inbox_generic_exception(self, monkeypatch):
"""Generic exception is caught and returns False (lines 176-179)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.resolve_inbox_target",
lambda first_arg, repo_root, get_branch_fn, get_user_fn: (_ for _ in ()).throw(ValueError("corrupt inbox")),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: None
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
from aipass.ai_mail.apps.modules.email import handle_inbox
result = handle_inbox([])
assert result is False
assert any("corrupt inbox" in e for e in errors)
# ===========================================================================
# handle_view additional paths
# ===========================================================================
class TestHandleViewExtended:
"""Extended tests for handle_view edge cases."""
def test_view_broken_pipe(self, tmp_path, monkeypatch):
"""BrokenPipeError is caught and returns True (line 217-219)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.mark_as_opened",
lambda bp, mid: (_ for _ in ()).throw(BrokenPipeError("pipe")),
)
from aipass.ai_mail.apps.modules.email import handle_view
result = handle_view(["some_id"])
assert result is True
def test_view_generic_exception(self, tmp_path, monkeypatch):
"""Generic exception is caught and returns True (lines 220-223)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.mark_as_opened",
lambda bp, mid: (_ for _ in ()).throw(RuntimeError("db error")),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_view
result = handle_view(["some_id"])
assert result is True
assert any("db error" in e for e in errors)
def test_view_latest_empty_inbox(self, tmp_path, monkeypatch):
"""'latest' with empty inbox returns True and prints error."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.load_inbox",
lambda f: {"messages": []},
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_view
result = handle_view(["latest"])
assert result is True
assert any("empty" in e.lower() for e in errors)
def test_view_latest_no_id_on_message(self, tmp_path, monkeypatch):
"""'latest' with message missing 'id' key returns True, prints error."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.load_inbox",
lambda f: {"messages": [{"subject": "no id here"}]},
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_view
result = handle_view(["latest"])
assert result is True
assert any("latest" in e.lower() or "could not" in e.lower() for e in errors)
# ===========================================================================
# handle_close additional paths
# ===========================================================================
class TestHandleCloseExtended:
"""Extended tests for handle_close edge cases."""
def test_close_batch_mixed_success_failure(self, tmp_path, monkeypatch):
"""Batch close with mixed results prints both success and error."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
results = [
("m1", True, "Closed m1"),
("m2", False, "Not found: m2"),
("m3", True, "Closed m3"),
]
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.batch_close",
lambda bp, ids, fn: (results, 2, 1),
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.batch_close_post_ops",
lambda bp, push_fn, central_fn, purge_fn: None,
)
printed: list[str] = []
errors: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_close
result = handle_close(["m1", "m2", "m3"])
assert result is True
assert any("Closed m1" in p for p in printed)
assert any("Not found: m2" in e for e in errors)
assert any("Closed m3" in p for p in printed)
assert any("Closed 2" in p and "failed 1" in p for p in printed)
def test_close_generic_exception(self, tmp_path, monkeypatch):
"""Generic exception is caught and returns True (lines 261-264)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: (_ for _ in ()).throw(RuntimeError("branch error")),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_close
result = handle_close(["m1"])
assert result is True
assert any("branch error" in e for e in errors)
def test_close_single_id_no_post_ops(self, tmp_path, monkeypatch):
"""Single ID close does NOT trigger post_ops."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: tmp_path,
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.batch_close",
lambda bp, ids, fn: ([("m1", True, "Closed m1")], 1, 0),
)
post_ops_called: list[bool] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.batch_close_post_ops",
lambda bp, push_fn, central_fn, purge_fn: post_ops_called.append(True),
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.error", lambda msg: None)
from aipass.ai_mail.apps.modules.email import handle_close
result = handle_close(["m1"])
assert result is True
assert len(post_ops_called) == 0
# ===========================================================================
# handle_reply generic exception (email.py lines 288-291)
# ===========================================================================
class TestHandleReplyExtended:
"""Extended tests for handle_reply edge cases."""
def test_reply_generic_exception(self, tmp_path, monkeypatch):
"""Generic exception is caught and returns True (lines 288-291)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: (_ for _ in ()).throw(OSError("disk fail")),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_reply
result = handle_reply(["msg1", "my reply"])
assert result is True
assert any("disk fail" in e for e in errors)
# ===========================================================================
# handle_sent generic exception (email.py lines 312-316)
# ===========================================================================
class TestHandleSentExtended:
"""Extended tests for handle_sent edge cases."""
def test_sent_generic_exception(self, monkeypatch):
"""Generic exception is caught and returns True (lines 312-316)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email._resolve_branch_path",
lambda: (_ for _ in ()).throw(RuntimeError("path error")),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email import handle_sent
result = handle_sent([])
assert result is True
assert any("path error" in e for e in errors)
# ===========================================================================
# print_introspection (email.py lines 361-402)
# ===========================================================================
class TestPrintIntrospection:
"""Tests for email.print_introspection."""
def test_print_introspection_outputs_module_info(self, monkeypatch):
"""print_introspection outputs module info including handler list."""
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg="", **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email.console", mock_console)
from aipass.ai_mail.apps.modules.email import print_introspection
print_introspection()
combined = "\n".join(printed)
assert "email Module" in combined
assert "Connected Handlers" in combined
assert "send.py" in combined
assert "inbox_ops.py" in combined
assert "json_handler.py" in combined
# ===========================================================================
# email_send.py: _delivery_callback
# ===========================================================================
class TestDeliveryCallback:
"""Tests for email_send._delivery_callback."""
def test_delivery_callback_calls_on_email_delivered(self, monkeypatch):
"""_delivery_callback delegates to on_email_delivered with correct args."""
delivered_args: list[dict] = []
def mock_on_delivered(
branch_path,
new_count,
opened_count,
total,
push_dashboard_fn=None,
update_central_fn=None,
):
"""Capture on_email_delivered arguments."""
delivered_args.append(
{
"branch_path": branch_path,
"new_count": new_count,
"opened_count": opened_count,
"total": total,
"push_dashboard_fn": push_dashboard_fn,
"update_central_fn": update_central_fn,
}
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.on_email_delivered",
mock_on_delivered,
)
from aipass.ai_mail.apps.modules.email_send import _delivery_callback
_delivery_callback("/some/path", 3, 2, 5)
assert len(delivered_args) == 1
assert delivered_args[0]["branch_path"] == "/some/path"
assert delivered_args[0]["new_count"] == 3
assert delivered_args[0]["opened_count"] == 2
assert delivered_args[0]["total"] == 5
assert delivered_args[0]["push_dashboard_fn"] is not None
# ===========================================================================
# email_send.py: _get_branch_info_fn
# ===========================================================================
class TestGetBranchInfoFn:
"""Tests for email_send._get_branch_info_fn."""
def test_get_branch_info_fn_success(self):
"""Returns function on success."""
from aipass.ai_mail.apps.modules.email_send import _get_branch_info_fn
result = _get_branch_info_fn()
assert result is None or callable(result)
def test_get_branch_info_fn_import_error(self, monkeypatch):
"""Returns None on ImportError."""
import builtins
original_import = builtins.__import__
def mock_import(name, *args, **kwargs):
"""Raise ImportError for branch_detection module."""
if "branch_detection" in name:
raise ImportError("no module")
return original_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", mock_import)
from aipass.ai_mail.apps.modules.email_send import _get_branch_info_fn
result = _get_branch_info_fn()
assert result is None
# ===========================================================================
# email_send.py: _send_direct BrokenPipeError & generic exception
# ===========================================================================
class TestSendDirectExtended:
"""Extended tests for email_send._send_direct edge cases."""
def test_send_direct_broken_pipe(self, monkeypatch):
"""BrokenPipeError is caught and returns True (line 187-189)."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.resolve_sender_info",
lambda fb, rr, amd, gbe, gcu: (_ for _ in ()).throw(BrokenPipeError("stdout closed")),
)
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: None
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None)
from aipass.ai_mail.apps.modules.email_send import _send_direct
result = _send_direct("@target", "Sub", "Msg")
assert result is True
def test_send_direct_generic_exception(self, monkeypatch):
"""Generic exception calls dispatch_send_error and returns False."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.resolve_sender_info",
lambda fb, rr, amd, gbe, gcu: (_ for _ in ()).throw(RuntimeError("send boom")),
)
dispatched_errors: list[tuple] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.dispatch_send_error",
lambda to, subj, err_msg, deliver_fn: dispatched_errors.append((to, subj, err_msg)),
)
errors: list[str] = []
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.error",
lambda msg: errors.append(msg),
)
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: None
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
from aipass.ai_mail.apps.modules.email_send import _send_direct
result = _send_direct("@target", "Sub", "Msg")
assert result is False
assert any("send boom" in e for e in errors)
assert len(dispatched_errors) == 1
assert dispatched_errors[0][0] == "@target"
def test_send_direct_broadcast_target(self, monkeypatch):
"""When to_branch is '@all', delegates to _send_broadcast."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.resolve_sender_info",
lambda fb, rr, amd, gbe, gcu: {
"email_address": "@ai_mail",
"display_name": "AI_MAIL",
"mailbox_path": "/tmp/mailbox",
},
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.get_all_branches",
lambda: [{"name": "A", "email": "@a"}],
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.send_to_broadcast",
lambda *a, **kw: (True, 1, 1, [("A", True, None)]),
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None)
from aipass.ai_mail.apps.modules.email_send import _send_direct
result = _send_direct("@all", "Hello", "World")
assert result is True
assert any("Broadcast" in p or "OK" in p for p in printed)
# ===========================================================================
# email_send.py: _fire_dispatch_trigger exception (line 201-202)
# ===========================================================================
class TestFireDispatchTrigger:
"""Tests for email_send._fire_dispatch_trigger exception handling."""
def test_fire_dispatch_trigger_exception_logged(self, monkeypatch):
"""Exception in trigger.fire is logged but does not propagate."""
mock_trigger = MagicMock()
mock_trigger.fire.side_effect = RuntimeError("trigger broken")
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.trigger", mock_trigger)
from aipass.ai_mail.apps.modules.email_send import _fire_dispatch_trigger
_fire_dispatch_trigger("@target", "Test Subject")
mock_trigger.fire.assert_called_once_with("email_dispatched", to="@target", subject="Test Subject")
# ===========================================================================
# email_send.py: _send_broadcast happy path & failure path
# ===========================================================================
class TestSendBroadcast:
"""Tests for email_send._send_broadcast."""
def test_send_broadcast_happy_path(self, monkeypatch):
"""Broadcast sends to all branches and reports success."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.get_all_branches",
lambda: [
{"name": "A", "email": "@a"},
{"name": "B", "email": "@b"},
],
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.send_to_broadcast",
lambda *a, **kw: (True, 2, 2, [("A", True, None), ("B", True, None)]),
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None)
from aipass.ai_mail.apps.modules.email_send import _send_broadcast
user_info = {
"email_address": "@ai_mail",
"display_name": "AI_MAIL",
"mailbox_path": "/tmp",
}
result = _send_broadcast("Subj", "Msg", user_info, False, False, None, None)
assert result is True
assert any("Broadcasting" in p for p in printed)
assert any("2/2" in p for p in printed)
def test_send_broadcast_failure_path(self, monkeypatch):
"""When send_to_broadcast returns string results (error), prints error."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.get_all_branches",
lambda: [{"name": "A", "email": "@a"}],
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.send_to_broadcast",
lambda *a, **kw: (False, 0, 1, "load failed"),
)
errors: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: None
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.error",
lambda msg: errors.append(msg),
)
from aipass.ai_mail.apps.modules.email_send import _send_broadcast
user_info = {
"email_address": "@ai_mail",
"display_name": "AI_MAIL",
"mailbox_path": "/tmp",
}
result = _send_broadcast("Subj", "Msg", user_info, False, False, None, None)
assert result is False
assert any("Failed to load" in e for e in errors)
# ===========================================================================
# email_send.py: print_introspection
# ===========================================================================
class TestEmailSendIntrospection:
"""Tests for email_send.print_introspection."""
def test_print_introspection_outputs_module_info(self, monkeypatch):
"""print_introspection prints function list and header."""
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg="", **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
from aipass.ai_mail.apps.modules.email_send import print_introspection
print_introspection()
combined = "\n".join(printed)
assert "EMAIL SEND ORCHESTRATION" in combined
assert "handle_send" in combined
assert "_send_direct" in combined
assert "_send_broadcast" in combined
assert "_delivery_callback" in combined
# ===========================================================================
# email_send.py: _send_interactive complete path (user provides input)
# ===========================================================================
class TestSendInteractiveExtended:
"""Extended tests for email_send._send_interactive."""
def test_send_interactive_complete_path(self, monkeypatch):
"""User provides input successfully, send proceeds."""
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.get_all_branches",
lambda: [{"name": "ALPHA", "email": "@alpha"}],
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.collect_interactive_input",
lambda branches: {
"to": "@alpha",
"subject": "Hi",
"message": "Hello there",
},
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.resolve_sender_info",
lambda fb, rr, amd, gbe, gcu: {
"email_address": "@ai_mail",
"display_name": "AI_MAIL",
"mailbox_path": "/tmp/mailbox",
},
)
monkeypatch.setattr(
"aipass.ai_mail.apps.modules.email_send.send_to_single",
lambda *a, **kw: (True, None),
)
printed: list[str] = []
mock_console = MagicMock()
mock_console.print = lambda msg, **kw: printed.append(str(msg))
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.console", mock_console)
monkeypatch.setattr("aipass.ai_mail.apps.modules.email_send.error", lambda msg: None)
from aipass.ai_mail.apps.modules.email_send import _send_interactive
result = _send_interactive()
assert result is True
assert any("@alpha" in p for p in printed)
assert any("sent" in p.lower() for p in printed)
src/aipass/ai_mail/tests/test_wake.py
+561 -6
View File
@@ -3,15 +3,17 @@
# Description: Tests for wake dispatch handler
# Version: 1.0.0
# Created: 2026-03-29
# Modified: 2026-03-29
# Modified: 2026-04-26
# =============================================
"""Tests for wake handler -- branch resolution, lock checking, PID checks, helpers."""
import json
import os
import subprocess
import pytest
from datetime import datetime, timedelta
from pathlib import Path as _Path
import aipass.ai_mail.apps.handlers.dispatch.wake as wake_mod
from aipass.ai_mail.apps.handlers.dispatch.wake import (
@@ -23,6 +25,13 @@ from aipass.ai_mail.apps.handlers.dispatch.wake import (
_find_claude_bin,
resolve_branch,
DispatchStatus,
MODEL_MAP,
DEFAULT_MODEL,
_acquire_lock,
_load_config,
_set_session_name,
_is_branch_occupied,
wake_branch,
)
@@ -411,7 +420,7 @@ def test_clean_zombies_subprocess_error(monkeypatch):
# --- Helpers ---------------------------------------------------------
_real_open = open
_REAL_OPEN = open
def _raise_process_lookup(pid, sig):
@@ -434,16 +443,14 @@ def _fake_open_factory(real_status_path, mapping):
def _fake_open(path, *args, **kwargs):
path_str = str(path)
if path_str in mapping:
return _real_open(mapping[path_str], *args, **kwargs)
return _real_open(path, *args, **kwargs)
return _REAL_OPEN(mapping[path_str], *args, **kwargs)
return _REAL_OPEN(path, *args, **kwargs)
return _fake_open
# --- Model flag tests ---------------------------------------------------
from aipass.ai_mail.apps.handlers.dispatch.wake import MODEL_MAP, DEFAULT_MODEL
def test_model_map_has_expected_entries():
"""MODEL_MAP should contain sonnet, opus, haiku shorthand mappings."""
@@ -624,3 +631,551 @@ class TestWakeBranchSpawnEnv:
assert captured_envs, "Popen was not called"
env = captured_envs[0]
assert local_bin in env.get("PATH", ""), f"~/.local/bin not in spawn_env PATH: {env.get('PATH', '')}"
# ─── NEW LINE COVERAGE TESTS ──────────────────────────────────
# --- _acquire_lock tests -----------------------------------------------
class TestAcquireLock:
"""Tests for _acquire_lock() — atomic lock file creation."""
def test_successful_lock_creation(self, tmp_path):
"""Successful lock writes pid, timestamp, and branch to JSON file."""
ok, msg = _acquire_lock(tmp_path, 12345)
assert ok is True
assert msg == "Lock acquired"
lock_file = tmp_path / ".ai_mail.local" / ".dispatch.lock"
assert lock_file.exists()
data = json.loads(lock_file.read_text(encoding="utf-8"))
assert data["pid"] == 12345
assert "timestamp" in data
assert data["branch"] == str(tmp_path)
def test_file_exists_error(self, tmp_path):
"""FileExistsError returns (False, 'Lock file already exists')."""
lock_dir = tmp_path / ".ai_mail.local"
lock_dir.mkdir(parents=True)
lock_file = lock_dir / ".dispatch.lock"
lock_file.write_text("{}", encoding="utf-8")
ok, msg = _acquire_lock(tmp_path, 999)
assert ok is False
assert msg == "Lock file already exists"
def test_os_error(self, tmp_path, monkeypatch):
"""OSError returns (False, error message)."""
original_os_open = os.open
def _fail_open(path, flags, *args, **kwargs):
if ".dispatch.lock" in str(path):
raise OSError("disk full")
return original_os_open(path, flags, *args, **kwargs)
monkeypatch.setattr(os, "open", _fail_open)
ok, msg = _acquire_lock(tmp_path, 999)
assert ok is False
assert "Lock failed:" in msg
assert "disk full" in msg
# --- _load_config tests -------------------------------------------------
class TestLoadConfig:
"""Tests for _load_config() — safety config loading with defaults."""
def test_no_config_file_returns_defaults(self, tmp_path, monkeypatch):
"""Missing config file returns default dict."""
monkeypatch.setattr(wake_mod, "CONFIG_FILE", tmp_path / "nonexistent.json")
result = _load_config()
assert result == {"max_turns_per_wake": 100}
def test_partial_config_fills_defaults(self, tmp_path, monkeypatch):
"""Config file without max_turns_per_wake gets default filled in."""
config_file = tmp_path / "safety_config.json"
config_file.write_text(json.dumps({"other_key": "value"}), encoding="utf-8")
monkeypatch.setattr(wake_mod, "CONFIG_FILE", config_file)
result = _load_config()
assert result["max_turns_per_wake"] == 100
assert result["other_key"] == "value"
def test_full_config_returned(self, tmp_path, monkeypatch):
"""Config file with all keys returned as-is."""
config_file = tmp_path / "safety_config.json"
config_file.write_text(json.dumps({"max_turns_per_wake": 50}), encoding="utf-8")
monkeypatch.setattr(wake_mod, "CONFIG_FILE", config_file)
result = _load_config()
assert result["max_turns_per_wake"] == 50
# --- _set_session_name tests --------------------------------------------
class TestSetSessionName:
"""Tests for _set_session_name() — writes custom-title to Claude session JSONL."""
def test_success_appends_entry(self, tmp_path, monkeypatch):
"""Creates expected JSON entry in the most recent session JSONL."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
session_file = projects_dir / "abc123.jsonl"
session_file.write_text("", encoding="utf-8")
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.Path.expanduser",
lambda self: tmp_path / ".claude" / "projects" if str(self).endswith("projects") else self,
)
# We need to mock expanduser properly — override the whole projects_dir lookup
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is True
content = session_file.read_text(encoding="utf-8")
entry = json.loads(content.strip())
assert entry["type"] == "custom-title"
assert entry["customTitle"] == "TEST-dispatched"
assert entry["sessionId"] == "abc123"
def test_no_projects_dir_returns_false(self, tmp_path, monkeypatch):
"""Returns False when ~/.claude/projects/{encoded} does not exist."""
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
def test_no_jsonl_files_returns_false(self, tmp_path, monkeypatch):
"""Returns False when projects dir exists but has no .jsonl files."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
def test_os_error_on_write_returns_false(self, tmp_path, monkeypatch):
"""Returns False when write to JSONL file raises OSError."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
session_file = projects_dir / "abc123.jsonl"
session_file.write_text("", encoding="utf-8")
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
def _fail_open(path, *args, **kwargs):
path_str = str(path)
if path_str.endswith(".jsonl") and "a" in args:
raise OSError("permission denied")
return _REAL_OPEN(path, *args, **kwargs)
monkeypatch.setattr("builtins.open", _fail_open)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
# --- _is_branch_occupied tests ------------------------------------------
class TestIsBranchOccupied:
"""Tests for _is_branch_occupied() — checks for interactive Claude sessions."""
def test_no_claude_processes(self, monkeypatch):
"""pgrep returns non-zero (no claude processes) -> not occupied."""
class FakeResult:
returncode = 1
stdout = ""
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: FakeResult())
assert _is_branch_occupied(_Path("/some/branch")) is False
def test_claude_in_different_dir(self, tmp_path, monkeypatch):
"""Claude running in a different directory -> not occupied."""
monkeypatch.setattr("sys.platform", "linux")
class FakeResult:
returncode = 0
stdout = "100\n"
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: FakeResult())
monkeypatch.setattr(os, "readlink", lambda p: "/some/other/dir")
assert _is_branch_occupied(tmp_path) is False
def test_claude_in_same_dir_interactive(self, tmp_path, monkeypatch):
"""Claude in same dir with interactive session -> occupied."""
monkeypatch.setattr("sys.platform", "linux")
resolved = str(tmp_path.resolve())
class FakeResult:
returncode = 0
stdout = "100\n"
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: FakeResult())
monkeypatch.setattr(os, "readlink", lambda p: resolved)
# _read_session_type returns "interactive" for this PID
monkeypatch.setattr(wake_mod, "_read_session_type", lambda pid_str: "interactive")
assert _is_branch_occupied(tmp_path) is True
def test_claude_in_same_dir_daemon_not_blocking(self, tmp_path, monkeypatch):
"""Claude in same dir with daemon session -> not blocking."""
monkeypatch.setattr("sys.platform", "linux")
resolved = str(tmp_path.resolve())
class FakeResult:
returncode = 0
stdout = "100\n"
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: FakeResult())
monkeypatch.setattr(os, "readlink", lambda p: resolved)
monkeypatch.setattr(wake_mod, "_read_session_type", lambda pid_str: "daemon")
assert _is_branch_occupied(tmp_path) is False
def test_pgrep_subprocess_failure_returns_false(self, monkeypatch):
"""subprocess failure returns False."""
def _fail(*a, **kw):
raise subprocess.SubprocessError("pgrep failed")
monkeypatch.setattr(subprocess, "run", _fail)
assert _is_branch_occupied(_Path("/some/branch")) is False
def test_readlink_oserror_continues(self, tmp_path, monkeypatch):
"""OSError on readlink is caught, continues to next PID."""
monkeypatch.setattr("sys.platform", "linux")
class FakeResult:
returncode = 0
stdout = "100\n200\n"
monkeypatch.setattr(subprocess, "run", lambda *a, **kw: FakeResult())
def _fail_readlink(p):
raise OSError("no such file")
monkeypatch.setattr(os, "readlink", _fail_readlink)
assert _is_branch_occupied(tmp_path) is False
# --- wake_branch integration tests -------------------------------------
def _make_wake_fixtures(tmp_path, monkeypatch):
"""Helper: set up branch directory, registry, and monkeypatched module constants."""
branch_path = tmp_path / "src" / "aipass" / "testbranch"
branch_path.mkdir(parents=True)
(branch_path / ".ai_mail.local").mkdir()
registry_file = tmp_path / "AIPASS_REGISTRY.json"
registry_file.write_text(
json.dumps({"branches": [{"name": "TESTBRANCH", "email": "@testbranch", "path": str(branch_path)}]}),
encoding="utf-8",
)
monkeypatch.setattr(wake_mod, "_REPO_ROOT", tmp_path)
monkeypatch.setattr(wake_mod, "BRANCH_REGISTRY", registry_file)
monkeypatch.setattr(wake_mod, "PAUSE_FILE", tmp_path / ".aipass" / "autonomous_pause")
monkeypatch.setattr(wake_mod, "CONFIG_FILE", tmp_path / "safety_config.json")
monkeypatch.setattr(wake_mod, "MONITOR_SCRIPT", tmp_path / "dispatch_monitor.py")
(tmp_path / "dispatch_monitor.py").touch()
return branch_path
def _patch_wake_deps(monkeypatch, **overrides):
"""Monkeypatch all wake_branch dependencies with sane defaults; override as needed."""
defaults = {
"_check_lock": lambda p: None,
"_clean_zombies": lambda: 0,
"_is_branch_occupied": lambda p: False,
"_acquire_lock": lambda p, pid: (True, "ok"),
"_check_pid_alive": lambda pid: True,
}
defaults.update(overrides)
for attr, val in defaults.items():
monkeypatch.setattr(wake_mod, attr, val)
monkeypatch.setattr("aipass.ai_mail.apps.handlers.dispatch.wake.time.sleep", lambda _: None)
class _FakeProc:
"""Minimal stand-in for subprocess.Popen return value."""
def __init__(self, pid: int = 55555):
self.pid = pid
class TestWakeBranch:
"""Tests for wake_branch() — all remaining code paths."""
# --- early exits ---
def test_auto_pause_file_blocks(self, tmp_path, monkeypatch):
"""auto=True with PAUSE_FILE existing returns failure."""
_make_wake_fixtures(tmp_path, monkeypatch)
pause = tmp_path / ".aipass" / "autonomous_pause"
pause.parent.mkdir(parents=True, exist_ok=True)
pause.touch()
status, ok = wake_branch("@testbranch", auto=True)
assert ok is False
assert any(s[0] == "fail" and "pause" in s[1] for s in status.steps)
def test_resolve_fails(self, tmp_path, monkeypatch):
"""Branch not found returns failure."""
_make_wake_fixtures(tmp_path, monkeypatch)
status, ok = wake_branch("@nonexistent")
assert ok is False
assert any(s[0] == "fail" and "resolve" in s[1] for s in status.steps)
def test_zombie_check_warns_but_continues(self, tmp_path, monkeypatch):
"""Zombie detected adds warning but dispatch continues."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch, _clean_zombies=lambda: 2)
monkeypatch.setattr("subprocess.Popen", lambda *a, **kw: _FakeProc())
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch")
assert ok is True
assert any(s[0] == "warn" and "zombie" in s[2].lower() for s in status.steps)
# --- lock exists ---
def test_lock_exists_auto_true_fails(self, tmp_path, monkeypatch):
"""Lock active + auto=True returns failure."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(
monkeypatch,
_check_lock=lambda p: {"pid": 111, "timestamp": "2026-01-01T00:00:00"},
_clean_zombies=lambda: 0,
)
status, ok = wake_branch("@testbranch", auto=True)
assert ok is False
assert any(s[0] == "fail" and "lock" in s[1] for s in status.steps)
def test_lock_exists_auto_false_delivers_to_inbox(self, tmp_path, monkeypatch):
"""Lock active + auto=False returns info + True (routed to inbox)."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(
monkeypatch,
_check_lock=lambda p: {"pid": 111, "timestamp": "2026-01-01T00:00:00"},
_clean_zombies=lambda: 0,
)
status, ok = wake_branch("@testbranch", auto=False)
assert ok is True
assert any(s[0] == "info" and "delivery" in s[1] for s in status.steps)
# --- branch occupied ---
def test_branch_occupied_blocks(self, tmp_path, monkeypatch):
"""Interactive session running -> blocked."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch, _is_branch_occupied=lambda p: True)
status, ok = wake_branch("@testbranch")
assert ok is False
assert any(s[0] == "fail" and "blocked" in s[1] for s in status.steps)
# --- fresh vs resume ---
def test_fresh_true_no_continue_flag(self, tmp_path, monkeypatch):
"""fresh=True -> claude_cmd does NOT include '-c' flag."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
captured_cmds: list = []
def fake_popen(cmd, **kwargs):
captured_cmds.append(cmd)
return _FakeProc()
monkeypatch.setattr("subprocess.Popen", fake_popen)
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch", fresh=True)
assert ok is True
# The claude subcommand is embedded after "--" in monitor_cmd
assert captured_cmds
cmd = captured_cmds[0]
# Everything after "--" is the claude command
sep_idx = cmd.index("--")
claude_part = cmd[sep_idx + 1 :]
assert "-c" not in claude_part
def test_fresh_false_has_continue_flag(self, tmp_path, monkeypatch):
"""fresh=False -> claude_cmd includes '-c' flag."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
captured_cmds: list = []
def fake_popen(cmd, **kwargs):
captured_cmds.append(cmd)
return _FakeProc()
monkeypatch.setattr("subprocess.Popen", fake_popen)
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch", fresh=False)
assert ok is True
assert captured_cmds
cmd = captured_cmds[0]
sep_idx = cmd.index("--")
claude_part = cmd[sep_idx + 1 :]
assert "-c" in claude_part
# --- custom message ---
def test_custom_message_sets_prompt(self, tmp_path, monkeypatch):
"""custom_message uses 'Hi. <message>' instead of DEFAULT_PROMPT."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
captured_cmds: list = []
def fake_popen(cmd, **kwargs):
captured_cmds.append(cmd)
return _FakeProc()
monkeypatch.setattr("subprocess.Popen", fake_popen)
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch", custom_message="Run the audit")
assert ok is True
assert captured_cmds
cmd = captured_cmds[0]
sep_idx = cmd.index("--")
claude_part = cmd[sep_idx + 1 :]
# Find the prompt argument (follows -p)
p_idx = claude_part.index("-p")
prompt = claude_part[p_idx + 1]
assert prompt.startswith("Hi. Run the audit")
# --- spawn errors ---
def test_spawn_file_not_found(self, tmp_path, monkeypatch):
"""FileNotFoundError during Popen -> fail step."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
def _fail_popen(*a, **kw):
raise FileNotFoundError("python not found")
monkeypatch.setattr("subprocess.Popen", _fail_popen)
status, ok = wake_branch("@testbranch")
assert ok is False
assert any(s[0] == "fail" and "spawn" in s[1] for s in status.steps)
def test_spawn_generic_exception(self, tmp_path, monkeypatch):
"""Generic exception during Popen -> fail step with class name."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
def _fail_popen(*a, **kw):
raise RuntimeError("something broke")
monkeypatch.setattr("subprocess.Popen", _fail_popen)
status, ok = wake_branch("@testbranch")
assert ok is False
assert any(s[0] == "fail" and "RuntimeError" in s[2] for s in status.steps)
# --- post-spawn: lock acquisition failure ---
def test_lock_acquisition_fails_after_spawn_warns(self, tmp_path, monkeypatch):
"""Lock fails after spawn -> warn step but still succeeds."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch, _acquire_lock=lambda p, pid: (False, "Lock file already exists"))
monkeypatch.setattr("subprocess.Popen", lambda *a, **kw: _FakeProc())
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch")
assert ok is True
assert any(s[0] == "warn" and "lock-acquire" in s[1] for s in status.steps)
# --- alive check fails ---
def test_alive_check_fails_cleans_up_lock(self, tmp_path, monkeypatch):
"""Agent dies immediately -> cleans up lock, returns False."""
branch_path = _make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch, _check_pid_alive=lambda pid: False)
monkeypatch.setattr("subprocess.Popen", lambda *a, **kw: _FakeProc())
# Create the lock file so we can verify it gets cleaned up
lock_dir = branch_path / ".ai_mail.local"
lock_dir.mkdir(parents=True, exist_ok=True)
lock_file = lock_dir / ".dispatch.lock"
lock_file.write_text("{}", encoding="utf-8")
status, ok = wake_branch("@testbranch")
assert ok is False
assert any(s[0] == "fail" and "alive" in s[1] for s in status.steps)
# Lock file should be cleaned up
assert not lock_file.exists()
# --- successful full path ---
def test_successful_full_path(self, tmp_path, monkeypatch):
"""All steps OK -> returns (status, True) with all expected steps."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
monkeypatch.setattr("subprocess.Popen", lambda *a, **kw: _FakeProc())
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
lambda *a, **kw: None,
raising=False,
)
status, ok = wake_branch("@testbranch")
assert ok is True
assert status.success is True
labels = [s[1] for s in status.steps]
assert "resolve" in labels
assert "pre-flight" in labels
assert "lock" in labels
assert "occupancy" in labels
assert "spawn" in labels
assert "lock-acquire" in labels
assert "alive" in labels
# --- notification failure doesn't break success ---
def test_notification_failure_does_not_break_success(self, tmp_path, monkeypatch):
"""Exception in send_notification is caught — overall success unchanged."""
_make_wake_fixtures(tmp_path, monkeypatch)
_patch_wake_deps(monkeypatch)
monkeypatch.setattr("subprocess.Popen", lambda *a, **kw: _FakeProc())
def _fail_notify(*a, **kw):
raise RuntimeError("dbus not found")
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.notify.send_notification",
_fail_notify,
raising=False,
)
status, ok = wake_branch("@testbranch")
assert ok is True
src/aipass/drone/tests/test_cli_routing.py
+768
View File
@@ -18,7 +18,9 @@ Covers the 2 missing CLI routing items:
- print_help (CR-007)
"""
import json
import sys
from pathlib import Path
from unittest.mock import patch
import pytest
@@ -49,3 +51,769 @@ def test_short_help() -> None: # CR-002
with patch.object(sys, "argv", ["drone", "-h"]):
result = main()
assert result == 0, "drone -h must return exit code 0"
# ===========================================================================
# main() dispatch — version, help, introspection
# ===========================================================================
_DRONE = "aipass.drone.apps.drone"
class TestMainVersion:
"""drone --version and -V flags."""
def test_version_long_flag(self) -> None:
"""--version prints version and returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "--version"]):
result = main()
assert result == 0
def test_version_short_flag(self) -> None:
"""-V prints version and returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "-V"]):
result = main()
assert result == 0
class TestMainHelp:
"""drone --help, -h, and help command."""
def test_help_long_flag(self) -> None:
"""--help returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "--help"]):
result = main()
assert result == 0
def test_help_word(self) -> None:
"""bare 'help' returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "help"]):
result = main()
assert result == 0
class TestMainNoArgs:
"""drone with no args shows introspection."""
def test_no_args_introspection(self) -> None:
"""No args calls print_introspection and returns 0."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone"]),
patch(f"{_DRONE}.print_introspection"),
):
result = main()
assert result == 0
def test_no_args_registry_error(self) -> None:
"""RegistryError during introspection returns 1."""
from aipass.drone.apps.drone import main
from aipass.drone.apps.modules import RegistryError
with (
patch.object(sys, "argv", ["drone"]),
patch(
f"{_DRONE}.print_introspection",
side_effect=RegistryError("no registry"),
),
):
result = main()
assert result == 1
# ===========================================================================
# main() dispatch — built-in commands
# ===========================================================================
class TestMainSystems:
"""drone systems command."""
def test_systems_success(self) -> None:
"""systems delegates to _handle_systems and returns 0."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "systems"]),
patch(f"{_DRONE}._handle_systems", return_value=0) as mock_sys,
):
result = main()
assert result == 0
mock_sys.assert_called_once()
def test_systems_registry_error(self) -> None:
"""RegistryError in systems returns 1."""
from aipass.drone.apps.drone import main
from aipass.drone.apps.modules import RegistryError
with (
patch.object(sys, "argv", ["drone", "systems"]),
patch(
f"{_DRONE}._handle_systems",
side_effect=RegistryError("broken"),
),
):
result = main()
assert result == 1
def test_systems_unexpected_error(self) -> None:
"""Unexpected Exception in systems returns 1."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "systems"]),
patch(
f"{_DRONE}._handle_systems",
side_effect=RuntimeError("boom"),
),
):
result = main()
assert result == 1
class TestMainScan:
"""drone scan command."""
def test_scan_no_target(self) -> None:
"""scan with no target returns 1."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "scan"]):
result = main()
assert result == 1
def test_scan_success(self) -> None:
"""scan with target delegates to scan module."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "scan", "@seedgo"]),
patch(
"aipass.drone.apps.modules.scan.scan",
return_value=[{"name": "audit"}],
),
):
result = main()
assert result == 0
def test_scan_failure(self) -> None:
"""scan returning None means failure -> exit 1."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "scan", "@seedgo"]),
patch("aipass.drone.apps.modules.scan.scan", return_value=None),
):
result = main()
assert result == 1
class TestMainActivate:
"""drone activate command."""
def test_activate_no_target(self) -> None:
"""activate with no target shows help and returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "activate"]):
result = main()
assert result == 0
def test_activate_help_flag(self) -> None:
"""activate --help shows help and returns 0."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "activate", "--help"]):
result = main()
assert result == 0
def test_activate_with_target(self) -> None:
"""activate with target delegates to _handle_activate."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "activate", "@seedgo"]),
patch(f"{_DRONE}._handle_activate", return_value=0) as mock_act,
):
result = main()
assert result == 0
mock_act.assert_called_once_with("@seedgo")
class TestMainList:
"""drone list command."""
def test_list_delegates(self) -> None:
"""list delegates to _handle_list."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "list"]),
patch(f"{_DRONE}._handle_list", return_value=0) as mock_list,
):
result = main()
assert result == 0
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."""
def test_remove_no_name(self) -> None:
"""remove with no name returns 1."""
from aipass.drone.apps.drone import main
with patch.object(sys, "argv", ["drone", "remove"]):
result = main()
assert result == 1
def test_remove_with_name(self) -> None:
"""remove with name delegates to _handle_remove."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "remove", "audit"]),
patch(f"{_DRONE}._handle_remove", return_value=0) as mock_rm,
):
result = main()
assert result == 0
mock_rm.assert_called_once_with("audit")
class TestMainAtTarget:
"""drone @target routing."""
def test_at_target_delegates(self) -> None:
"""@target routes to _handle_target."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "@seedgo", "audit"]),
patch(f"{_DRONE}._handle_target", return_value=0) as mock_tgt,
):
result = main()
assert result == 0
mock_tgt.assert_called_once_with(["@seedgo", "audit"])
class TestMainModuleRouting:
"""drone bare module name routing."""
def test_discovered_module_bool_true(self) -> None:
"""Discovered module returning True yields exit 0."""
from aipass.drone.apps.drone import main
mock_mod = type(sys)("fake_mod")
mock_mod.handle_command = lambda cmd, args: True
with (
patch.object(sys, "argv", ["drone", "config", "list"]),
patch(
f"{_DRONE}._discover_modules",
return_value=[("config", "Config module")],
),
patch(
f"{_DRONE}.importlib.import_module",
return_value=mock_mod,
),
):
result = main()
assert result == 0
def test_discovered_module_bool_false(self) -> None:
"""Discovered module returning False yields exit 1."""
from aipass.drone.apps.drone import main
mock_mod = type(sys)("fake_mod")
mock_mod.handle_command = lambda cmd, args: False
with (
patch.object(sys, "argv", ["drone", "config", "broken"]),
patch(
f"{_DRONE}._discover_modules",
return_value=[("config", "Config module")],
),
patch(
f"{_DRONE}.importlib.import_module",
return_value=mock_mod,
),
):
result = main()
assert result == 1
def test_discovered_module_dict_result(self) -> None:
"""Discovered module returning dict uses stdout/stderr/exit_code."""
from aipass.drone.apps.drone import main
mock_mod = type(sys)("fake_mod")
mock_mod.handle_command = lambda cmd, args: {
"stdout": "output",
"stderr": "",
"exit_code": 0,
}
with (
patch.object(sys, "argv", ["drone", "config", "list"]),
patch(
f"{_DRONE}._discover_modules",
return_value=[("config", "Config module")],
),
patch(
f"{_DRONE}.importlib.import_module",
return_value=mock_mod,
),
):
result = main()
assert result == 0
def test_discovered_module_exception(self) -> None:
"""Module raising exception returns 1."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "config", "list"]),
patch(
f"{_DRONE}._discover_modules",
return_value=[("config", "Config module")],
),
patch(
f"{_DRONE}.importlib.import_module",
side_effect=ImportError("nope"),
),
):
result = main()
assert result == 1
class TestMainCustomCommand:
"""drone custom command matching fallback."""
def test_custom_command_matched(self) -> None:
"""Custom command matched returns its result."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "audit", "aipass"]),
patch(f"{_DRONE}._discover_modules", return_value=[]),
patch(f"{_DRONE}._handle_custom_command", return_value=0),
):
result = main()
assert result == 0
def test_custom_command_not_matched(self) -> None:
"""Unmatched command falls through to unknown."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "nonexistent_cmd"]),
patch(f"{_DRONE}._discover_modules", return_value=[]),
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
patch(
"aipass.drone.apps.modules.resolver.branch_exists",
return_value=False,
),
):
result = main()
assert result == 1
class TestMainUnknownCommand:
"""drone unknown command handling with branch hint."""
def test_unknown_bare_branch_name(self) -> None:
"""Bare branch name shows @ prefix hint."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "seedgo"]),
patch(f"{_DRONE}._discover_modules", return_value=[]),
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
patch(
"aipass.drone.apps.modules.resolver.branch_exists",
return_value=True,
),
):
result = main()
assert result == 1
def test_unknown_branch_check_fails(self) -> None:
"""Exception in branch_exists doesn't crash -- still returns 1."""
from aipass.drone.apps.drone import main
with (
patch.object(sys, "argv", ["drone", "broken_cmd"]),
patch(f"{_DRONE}._discover_modules", return_value=[]),
patch(f"{_DRONE}._handle_custom_command", return_value=-1),
patch(
"aipass.drone.apps.modules.resolver.branch_exists",
side_effect=Exception("boom"),
),
):
result = main()
assert result == 1
# ===========================================================================
# _handle_systems() paths
# ===========================================================================
class TestHandleSystems:
"""_handle_systems() logic paths."""
def test_no_registry(self) -> None:
"""No registry in CWD tree returns 0 with message."""
from aipass.drone.apps.drone import _handle_systems
with patch(f"{_DRONE}._cwd_has_registry", return_value=False):
result = _handle_systems()
assert result == 0
def test_with_branches_and_modules(self) -> None:
"""Normal case with modules and branches."""
from aipass.drone.apps.drone import _handle_systems
branches = [
{"name": "drone", "profile": "library", "description": "Router"},
{"name": "myapp", "profile": "agent"},
]
with (
patch(f"{_DRONE}._cwd_has_registry", return_value=True),
patch(f"{_DRONE}.get_all_branches", return_value=branches),
patch(f"{_DRONE}.list_modules", return_value=["git"]),
patch(
f"{_DRONE}.get_module_info",
return_value=type("I", (), {"description": "Git ops"})(),
),
):
result = _handle_systems()
assert result == 0
def test_aipass_home_hint(self) -> None:
"""Shows hint when AIPASS_HOME not set and drone not in branches."""
import os
from aipass.drone.apps.drone import _handle_systems
with (
patch(f"{_DRONE}._cwd_has_registry", return_value=True),
patch(f"{_DRONE}.get_all_branches", return_value=[]),
patch(f"{_DRONE}.list_modules", return_value=[]),
patch.dict(os.environ, {}, clear=False),
):
os.environ.pop("AIPASS_HOME", None)
result = _handle_systems()
assert result == 0
# ===========================================================================
# _handle_module() paths
# ===========================================================================
class TestHandleModule:
"""_handle_module() introspection, help, and command routing."""
def test_no_args_introspection(self) -> None:
"""No args shows introspection text."""
from aipass.drone.apps.drone import _handle_module
with patch(
f"{_DRONE}.get_module_introspective",
return_value="Module info",
):
result = _handle_module("git", [])
assert result == 0
def test_no_args_no_introspection(self) -> None:
"""No args with empty introspection text shows fallback."""
from aipass.drone.apps.drone import _handle_module
with patch(f"{_DRONE}.get_module_introspective", return_value=""):
result = _handle_module("git", [])
assert result == 0
def test_help_flag(self) -> None:
"""--help shows help text."""
from aipass.drone.apps.drone import _handle_module
with patch(f"{_DRONE}.get_module_help", return_value="Help text"):
result = _handle_module("git", ["--help"])
assert result == 0
def test_help_no_text(self) -> None:
"""--help with no text shows fallback."""
from aipass.drone.apps.drone import _handle_module
with patch(f"{_DRONE}.get_module_help", return_value=""):
result = _handle_module("git", ["--help"])
assert result == 0
def test_command_routing(self) -> None:
"""Command routes through route_module_command."""
from aipass.drone.apps.drone import _handle_module
with patch(
f"{_DRONE}.route_module_command",
return_value={"stdout": "ok", "stderr": "", "exit_code": 0},
):
result = _handle_module("git", ["status"])
assert result == 0
def test_command_import_error(self) -> None:
"""ImportError during module command returns 1."""
from aipass.drone.apps.drone import _handle_module
with patch(
f"{_DRONE}.route_module_command",
side_effect=ImportError("missing"),
):
result = _handle_module("git", ["status"])
assert result == 1
# ===========================================================================
# _handle_target() paths
# ===========================================================================
class TestHandleTarget:
"""_handle_target() routing for @branch commands."""
def test_module_route(self) -> None:
"""Module target (non-interactive) routes to _handle_module."""
from aipass.drone.apps.drone import _handle_target
with (
patch(f"{_DRONE}.is_module", return_value=True),
patch(f"{_DRONE}._handle_module", return_value=0) as mock_hm,
):
result = _handle_target(["@git", "status"])
assert result == 0
mock_hm.assert_called_once_with("git", ["status"])
def test_no_args_introspection(self) -> None:
"""@target with no args routes via route_command for introspection."""
from aipass.drone.apps.drone import _handle_target
from aipass.drone.apps.handlers.executor import CommandResult
mock_result = CommandResult(
stdout="introspection",
stderr="",
exit_code=0,
branch="seedgo",
command="",
)
with (
patch(f"{_DRONE}.is_module", return_value=False),
patch(f"{_DRONE}.route_command", return_value=mock_result),
):
result = _handle_target(["@seedgo"])
assert result == 0
def test_help_flag(self) -> None:
"""@target --help routes via get_help."""
from aipass.drone.apps.drone import _handle_target
mock_help = type("H", (), {"text": "Help text"})()
with (
patch(f"{_DRONE}.is_module", return_value=False),
patch(f"{_DRONE}.get_help", return_value=mock_help),
):
result = _handle_target(["@seedgo", "--help"])
assert result == 0
def test_command_routing(self) -> None:
"""@target command routes via route_command."""
from aipass.drone.apps.drone import _handle_target
from aipass.drone.apps.handlers.executor import CommandResult
mock_result = CommandResult(
stdout="output",
stderr="",
exit_code=0,
branch="seedgo",
command="audit",
)
with (
patch(f"{_DRONE}.is_module", return_value=False),
patch(f"{_DRONE}.route_command", return_value=mock_result),
):
result = _handle_target(["@seedgo", "audit", "aipass"])
assert result == 0
def test_branch_not_found_module_fallback(self) -> None:
"""BranchNotFoundError for a module falls back to _handle_module."""
from aipass.drone.apps.drone import _handle_target
from aipass.drone.apps.modules import BranchNotFoundError
with (
patch(f"{_DRONE}.is_module", side_effect=[False, True]),
patch(
f"{_DRONE}.route_command",
side_effect=BranchNotFoundError("not found"),
),
patch(f"{_DRONE}._handle_module", return_value=0) as mock_hm,
):
result = _handle_target(["@seedgo", "audit"])
assert result == 0
mock_hm.assert_called_once()
# ===========================================================================
# _handle_custom_command() paths
# ===========================================================================
class TestHandleCustomCommand:
"""_handle_custom_command() matching and routing."""
_MATCH = "aipass.drone.apps.modules.commands.match"
def test_no_match(self) -> None:
"""No match returns -1 sentinel."""
from aipass.drone.apps.drone import _handle_custom_command
with patch(self._MATCH, return_value=None):
result = _handle_custom_command(["unknown"])
assert result == -1
def test_matched_routes_success(self) -> None:
"""Matched command routes and returns exit code."""
from aipass.drone.apps.drone import _handle_custom_command
from aipass.drone.apps.handlers.executor import CommandResult
cmd_data = {
"target": "@seedgo",
"command": "audit",
"args": ["aipass"],
}
mock_result = CommandResult(
stdout="ok",
stderr="",
exit_code=0,
branch="seedgo",
command="audit",
)
with (
patch(self._MATCH, return_value=(cmd_data, [])),
patch(f"{_DRONE}.route_command", return_value=mock_result),
):
result = _handle_custom_command(["audit"])
assert result == 0
# ===========================================================================
# Helper functions
# ===========================================================================
class TestReadInboxMessageId:
"""_read_inbox_message_id() edge cases."""
def test_valid_index(self, tmp_path: Path) -> None:
"""Returns message ID for valid index."""
inbox = tmp_path / "inbox.json"
inbox.write_text(
json.dumps({"messages": [{"id": "abc123"}, {"id": "def456"}]}),
encoding="utf-8",
)
from aipass.drone.apps.drone import _read_inbox_message_id
assert _read_inbox_message_id(inbox, 1) == "abc123"
assert _read_inbox_message_id(inbox, 2) == "def456"
def test_out_of_range(self, tmp_path: Path) -> None:
"""Returns None for out-of-range index."""
inbox = tmp_path / "inbox.json"
inbox.write_text(
json.dumps({"messages": [{"id": "abc"}]}),
encoding="utf-8",
)
from aipass.drone.apps.drone import _read_inbox_message_id
assert _read_inbox_message_id(inbox, 5) is None
assert _read_inbox_message_id(inbox, 0) is None
def test_corrupt_file(self, tmp_path: Path) -> None:
"""Returns None for corrupt inbox."""
inbox = tmp_path / "inbox.json"
inbox.write_text("{bad json", encoding="utf-8")
from aipass.drone.apps.drone import _read_inbox_message_id
assert _read_inbox_message_id(inbox, 1) is None
class TestDiscoverModules:
"""_discover_modules() auto-discovery."""
def test_discovers_modules_with_handle_command(self) -> None:
"""Modules with handle_command are discovered."""
from aipass.drone.apps.drone import _discover_modules
result = _discover_modules()
names = [m[0] for m in result]
assert "git_module" in names
assert "resolver" in names
def test_skips_private_files(self) -> None:
"""Files starting with _ are skipped."""
from aipass.drone.apps.drone import _discover_modules
result = _discover_modules()
names = [m[0] for m in result]
assert "__init__" not in names
class TestCliEntryPoint:
"""cli.py entry point."""
def test_cli_main_calls_drone_main(self) -> None:
"""cli.main() calls drone main and exits."""
from aipass.drone.cli import main as cli_main
with (
patch("aipass.drone.cli._drone_main", return_value=0),
pytest.raises(SystemExit) as exc_info,
):
cli_main()
assert exc_info.value.code == 0
src/aipass/drone/tests/test_registry.py
+234
View File
@@ -0,0 +1,234 @@
# =================== AIPass ====================
# Name: test_registry.py
# Description: Tests for the registry module orchestrator
# Version: 1.0.0
# Created: 2026-04-26
# Modified: 2026-04-26
# =============================================
"""Tests for the registry module orchestrator (apps/modules/registry.py).
Covers:
- print_introspection() output
- print_help() output
- handle_command() dispatch: load, branches, lookup, help, introspection, unknown
"""
from unittest.mock import patch
import pytest
_REG = "aipass.drone.apps.modules.registry"
# ===========================================================================
# print_introspection
# ===========================================================================
class TestPrintIntrospection:
"""print_introspection() output."""
def test_prints_module_info(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Prints registry module info to stdout."""
from aipass.drone.apps.modules.registry import print_introspection
print_introspection()
captured = capsys.readouterr()
assert "registry" in captured.out.lower()
assert "handler" in captured.out.lower()
def test_fallback_console(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Falls back to rich.Console when CLI console unavailable."""
import importlib
import sys
import aipass.drone.apps.modules.registry as reg_mod
saved = sys.modules.pop("aipass.cli.apps.modules.display", None)
sys.modules["aipass.cli.apps.modules.display"] = None # type: ignore[assignment]
try:
importlib.reload(reg_mod)
reg_mod.print_introspection()
captured = capsys.readouterr()
assert "registry" in captured.out.lower()
finally:
if saved is not None:
sys.modules["aipass.cli.apps.modules.display"] = saved
else:
sys.modules.pop("aipass.cli.apps.modules.display", None)
importlib.reload(reg_mod)
# ===========================================================================
# print_help
# ===========================================================================
class TestPrintHelp:
"""print_help() output."""
def test_prints_help(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Prints help text with command list."""
from aipass.drone.apps.modules.registry import print_help
print_help()
captured = capsys.readouterr()
assert "load" in captured.out
assert "branches" in captured.out
assert "lookup" in captured.out
# ===========================================================================
# handle_command — introspection and help
# ===========================================================================
class TestHandleCommandIntrospection:
"""handle_command() introspection and help paths."""
def test_no_command_no_args_introspection(self) -> None:
"""No command + no args triggers introspection."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.print_introspection") as mock_intro:
result = handle_command()
assert result is True
mock_intro.assert_called_once()
def test_help_flag_command(self) -> None:
"""--help as command triggers print_help."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.print_help") as mock_help:
result = handle_command("--help")
assert result is True
mock_help.assert_called_once()
def test_h_flag_command(self) -> None:
"""-h as command triggers print_help."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.print_help") as mock_help:
result = handle_command("-h")
assert result is True
mock_help.assert_called_once()
def test_help_in_args(self) -> None:
"""--help in args triggers print_help."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.print_help") as mock_help:
result = handle_command("load", ["--help"])
assert result is True
mock_help.assert_called_once()
# ===========================================================================
# handle_command — load
# ===========================================================================
class TestHandleCommandLoad:
"""handle_command('load') path."""
def test_load_success(self) -> None:
"""load returns True and calls load_registry."""
from aipass.drone.apps.modules.registry import handle_command
mock_registry = {"branches": {"drone": {}, "seedgo": {}}}
with patch(f"{_REG}.load_registry", return_value=mock_registry):
result = handle_command("load", [])
assert result is True
def test_load_empty_registry(self) -> None:
"""load with empty registry still returns True."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.load_registry", return_value={}):
result = handle_command("load", [])
assert result is True
# ===========================================================================
# handle_command — branches
# ===========================================================================
class TestHandleCommandBranches:
"""handle_command('branches') path."""
def test_branches_no_filter(self) -> None:
"""branches with no args lists all branches."""
from aipass.drone.apps.modules.registry import handle_command
mock_branches = [{"name": "drone"}, {"name": "seedgo"}]
with patch(f"{_REG}.get_all_branches", return_value=mock_branches) as mock_gab:
result = handle_command("branches", [])
assert result is True
mock_gab.assert_called_once_with(branch_type=None)
def test_branches_with_type_filter(self) -> None:
"""branches with type arg filters by type."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.get_all_branches", return_value=[]) as mock_gab:
result = handle_command("branches", ["library"])
assert result is True
mock_gab.assert_called_once_with(branch_type="library")
# ===========================================================================
# handle_command — lookup
# ===========================================================================
class TestHandleCommandLookup:
"""handle_command('lookup') path."""
def test_lookup_no_args(self) -> None:
"""lookup with no args returns False."""
from aipass.drone.apps.modules.registry import handle_command
result = handle_command("lookup", [])
assert result is False
def test_lookup_found(self) -> None:
"""lookup with existing branch returns True."""
from aipass.drone.apps.modules.registry import handle_command
mock_branch = {"name": "drone", "profile": "library"}
with patch(f"{_REG}.get_branch_by_name", return_value=mock_branch):
result = handle_command("lookup", ["drone"])
assert result is True
def test_lookup_not_found(self) -> None:
"""lookup with missing branch returns False."""
from aipass.drone.apps.modules.registry import handle_command
with patch(f"{_REG}.get_branch_by_name", return_value=None):
result = handle_command("lookup", ["ghost"])
assert result is False
# ===========================================================================
# handle_command — unknown
# ===========================================================================
class TestHandleCommandUnknown:
"""handle_command() unknown command path."""
def test_unknown_command(self) -> None:
"""Unknown command returns False."""
from aipass.drone.apps.modules.registry import handle_command
result = handle_command("nonexistent", ["arg"])
assert result is False
def test_none_command_with_args(self) -> None:
"""None command with args (no --help) falls through to unknown."""
from aipass.drone.apps.modules.registry import handle_command
result = handle_command(None, ["some_arg"])
assert result is False
src/aipass/seedgo/.seedgo/bypass.json
+10
View File
@@ -259,6 +259,16 @@
"file": "tests/test_hook_bridge.py",
"standard": "help_text",
"reason": "python3 references appear in test data strings (hook command fixtures), not user-facing help text."
},
{
"file": "tests/test_coverage_audit.py",
"standard": "architecture",
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
},
{
"file": "tests/test_coverage_audit.py",
"standard": "encapsulation",
"reason": "Unit tests must import handlers directly to test them in isolation. Same pattern as test_checkers_batch5.py."
}
],
"notes": {
src/aipass/seedgo/apps/handlers/bypass/bypass_handler.py
+8 -3
View File
@@ -170,13 +170,18 @@ def load_bypass_rules(branch_path: str) -> List[Dict[str, Any]]:
try:
if bypass_file.exists():
with open(bypass_file, "r", encoding="utf-8") as f:
config = json.load(f)
content = bypass_file.read_text(encoding="utf-8").strip()
if not content:
logger.warning("[bypass_handler] Empty bypass.json at %s — skipping", bypass_file)
return []
config = json.loads(content)
rules = config.get("bypass", [])
json_handler.log_operation("bypass_rules_loaded", {"branch": branch_path, "count": len(rules)})
return rules
except json.JSONDecodeError as e:
logger.warning("[bypass_handler] Corrupt bypass.json at %s: %s — skipping", bypass_file, e)
except Exception as e:
logger.error(f"[bypass_handler] Error loading bypass rules: {e}")
logger.error("[bypass_handler] Error loading bypass rules: %s", e)
return []
src/aipass/seedgo/tests/test_coverage_audit.py
+5 -3
View File
@@ -1085,8 +1085,10 @@ class TestCollectPyFiles:
"""Files matching ignore patterns are excluded."""
import sys
# Use a unique pattern that will NOT collide with the pytest tmp_path
# directory name (which includes the test function name).
mock_ign = sys.modules["aipass.seedgo.apps.handlers.bypass"].ignore_handler
mock_ign.get_audit_ignore_patterns.return_value = ["test_"]
mock_ign.get_audit_ignore_patterns.return_value = ["xskip_"]
from aipass.seedgo.apps.handlers.audit.branch_audit import (
_collect_py_files,
@@ -1095,11 +1097,11 @@ class TestCollectPyFiles:
apps_dir = tmp_path / "apps"
apps_dir.mkdir()
(apps_dir / "module.py").write_text("pass", encoding="utf-8")
(apps_dir / "test_module.py").write_text("pass", encoding="utf-8")
(apps_dir / "xskip_bad.py").write_text("pass", encoding="utf-8")
result = _collect_py_files(tmp_path)
names = [f["name"] for f in result]
assert "module.py" in names
assert "test_module.py" not in names
assert "xskip_bad.py" not in names
class TestExtractBranchLevelViolations:
src/aipass/spawn/templates/builder/.spawn/.template_registry.json
+2 -2
View File
@@ -155,7 +155,7 @@
"content_hash": "a4cf0a8e3b4f",
"has_branch_placeholder": false
},
"f026": {
"f015": {
"path": "apps/modules/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",
@@ -263,7 +263,7 @@
"content_hash": "28e9ae373563",
"has_branch_placeholder": false
},
"f015": {
"f026": {
"path": "apps/plugins/__init__.py",
"name": "__init__.py",
"content_hash": "e3b0c44298fc",
src/aipass/trigger/tests/test_log_watcher.py
+654
View File
@@ -614,3 +614,657 @@ class TestInitializePositions:
watcher = lw.BranchLogWatcher()
watcher.initialize_positions()
assert watcher.log_positions[str(log_file)] == log_file.stat().st_size
# ---------------------------------------------------------------------------
# Tests -- _load_seen_hashes
# ---------------------------------------------------------------------------
class TestLoadSeenHashes:
"""Tests for _load_seen_hashes persistence."""
def test_loads_from_existing_file(self, tmp_path):
"""Loads hashes from a valid trigger_data.json file."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"seen_error_hashes": ["aaa", "bbb"]}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
lw._seen_error_hashes = set()
lw._load_seen_hashes()
assert lw._seen_error_hashes == {"aaa", "bbb"}
def test_handles_missing_file(self, tmp_path):
"""Missing file leaves _seen_error_hashes unchanged (no crash)."""
lw = _import_log_watcher()
lw.TRIGGER_DATA_FILE = tmp_path / "nonexistent.json"
lw._seen_error_hashes = {"existing"}
lw._load_seen_hashes()
assert lw._seen_error_hashes == {"existing"}
def test_handles_corrupt_json(self, tmp_path):
"""Corrupt JSON resets _seen_error_hashes to empty set."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text("{invalid json", encoding="utf-8")
lw.TRIGGER_DATA_FILE = data_file
lw._seen_error_hashes = {"leftovers"}
lw._load_seen_hashes()
assert lw._seen_error_hashes == set()
def test_handles_missing_key(self, tmp_path):
"""File exists but has no seen_error_hashes key -- loads empty."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(json.dumps({"other_key": 1}), encoding="utf-8")
lw.TRIGGER_DATA_FILE = data_file
lw._seen_error_hashes = {"old"}
lw._load_seen_hashes()
assert lw._seen_error_hashes == set()
# ---------------------------------------------------------------------------
# Tests -- _save_seen_hashes
# ---------------------------------------------------------------------------
class TestSaveSeenHashes:
"""Tests for _save_seen_hashes persistence."""
def test_saves_to_new_file(self, tmp_path):
"""Creates trigger_data.json when it does not exist yet."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
lw.TRIGGER_DATA_FILE = data_file
lw._seen_error_hashes = {"hash1", "hash2"}
lw._save_seen_hashes()
written = json.loads(data_file.read_text(encoding="utf-8"))
assert set(written["seen_error_hashes"]) == {"hash1", "hash2"}
def test_merges_with_existing_data(self, tmp_path):
"""Preserves other keys already in trigger_data.json."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"log_positions": {"/a.log": 100}}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
lw._seen_error_hashes = {"x"}
lw._save_seen_hashes()
written = json.loads(data_file.read_text(encoding="utf-8"))
assert written["log_positions"] == {"/a.log": 100}
assert written["seen_error_hashes"] == ["x"]
def test_handles_write_error(self, tmp_path):
"""Write failure logs warning but does not raise."""
lw = _import_log_watcher()
bad_path = tmp_path / "nope" / "nope" / "trigger_data.json"
lw.TRIGGER_DATA_FILE = bad_path
lw._seen_error_hashes = {"z"}
with patch.object(lw, "atomic_write_json", side_effect=PermissionError("denied")):
lw._save_seen_hashes()
# ---------------------------------------------------------------------------
# Tests -- _load_log_positions
# ---------------------------------------------------------------------------
class TestLoadLogPositions:
"""Tests for _load_log_positions persistence."""
def test_loads_positions_from_file(self, tmp_path):
"""Returns positions dict when trigger_data.json has log_positions."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"log_positions": {"/a.log": 42, "/b.log": 99}}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
result = lw._load_log_positions()
assert result == {"/a.log": 42, "/b.log": 99}
def test_returns_empty_for_missing_file(self, tmp_path):
"""Returns empty dict when file does not exist."""
lw = _import_log_watcher()
lw.TRIGGER_DATA_FILE = tmp_path / "nonexistent.json"
assert lw._load_log_positions() == {}
def test_returns_empty_for_corrupt_json(self, tmp_path):
"""Returns empty dict when file has corrupt JSON."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text("not json!", encoding="utf-8")
lw.TRIGGER_DATA_FILE = data_file
assert lw._load_log_positions() == {}
def test_returns_empty_when_positions_not_dict(self, tmp_path):
"""Returns empty dict when log_positions is not a dict."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"log_positions": "not_a_dict"}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
assert lw._load_log_positions() == {}
def test_coerces_values_to_int(self, tmp_path):
"""String position values are coerced to int."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"log_positions": {"/x.log": "123"}}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
result = lw._load_log_positions()
assert result["/x.log"] == 123
assert isinstance(result["/x.log"], int)
# ---------------------------------------------------------------------------
# Tests -- _save_log_positions
# ---------------------------------------------------------------------------
class TestSaveLogPositions:
"""Tests for _save_log_positions persistence."""
def test_saves_positions_to_new_file(self, tmp_path):
"""Creates file with log_positions key."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
lw.TRIGGER_DATA_FILE = data_file
lw._save_log_positions({"/a.log": 50})
written = json.loads(data_file.read_text(encoding="utf-8"))
assert written["log_positions"] == {"/a.log": 50}
def test_merges_with_existing_data(self, tmp_path):
"""Preserves other keys in trigger_data.json."""
lw = _import_log_watcher()
data_file = tmp_path / "trigger_data.json"
data_file.write_text(
json.dumps({"seen_error_hashes": ["abc"]}),
encoding="utf-8",
)
lw.TRIGGER_DATA_FILE = data_file
lw._save_log_positions({"/b.log": 77})
written = json.loads(data_file.read_text(encoding="utf-8"))
assert written["seen_error_hashes"] == ["abc"]
assert written["log_positions"] == {"/b.log": 77}
def test_handles_write_error(self, tmp_path):
"""Write failure logs warning but does not raise."""
lw = _import_log_watcher()
lw.TRIGGER_DATA_FILE = tmp_path / "trigger_data.json"
with patch.object(lw, "atomic_write_json", side_effect=OSError("disk full")):
lw._save_log_positions({"/c.log": 10})
# ---------------------------------------------------------------------------
# Tests -- _is_stale_entry (additional format coverage)
# ---------------------------------------------------------------------------
class TestIsStaleEntryFormats:
"""Additional format coverage for _is_stale_entry."""
def test_iso_format_with_microseconds_fresh(self):
"""ISO format with microseconds: T separator and dot microseconds."""
lw = _import_log_watcher()
recent = datetime.now() - timedelta(seconds=5)
ts = recent.strftime("%Y-%m-%dT%H:%M:%S.%f")
assert lw._is_stale_entry(ts) is False
def test_iso_format_simple_stale(self):
"""ISO format without microseconds, stale timestamp."""
lw = _import_log_watcher()
old = datetime.now() - timedelta(seconds=600)
ts = old.strftime("%Y-%m-%dT%H:%M:%S")
assert lw._is_stale_entry(ts) is True
def test_simple_format_no_microseconds_fresh(self):
"""Simple YYYY-MM-DD HH:MM:SS format, fresh."""
lw = _import_log_watcher()
recent = datetime.now() - timedelta(seconds=2)
ts = recent.strftime("%Y-%m-%d %H:%M:%S")
assert lw._is_stale_entry(ts) is False
def test_whitespace_stripped(self):
"""Leading/trailing whitespace is stripped before parsing."""
lw = _import_log_watcher()
recent = datetime.now() - timedelta(seconds=5)
ts = " " + recent.strftime("%Y-%m-%d %H:%M:%S.%f") + " "
assert lw._is_stale_entry(ts) is False
def test_empty_string_returns_true(self):
"""Empty string is unparseable and treated as stale."""
lw = _import_log_watcher()
assert lw._is_stale_entry("") is True
# ---------------------------------------------------------------------------
# Tests -- _detect_branch_from_path (additional edge cases)
# ---------------------------------------------------------------------------
class TestDetectBranchFromPathEdgeCases:
"""Additional edge cases for _detect_branch_from_path."""
def test_pycache_directory_ignored(self):
"""__pycache__ after aipass/ is not treated as a branch."""
lw = _import_log_watcher()
path = str(Path("/home/user/src") / "aipass" / "__pycache__" / "logs" / "something.log")
assert lw._detect_branch_from_path(path) == "UNKNOWN"
def test_system_logs_unknown_file(self):
"""Unknown file in system_logs returns UNKNOWN."""
lw = _import_log_watcher()
path = str(lw.SYSTEM_LOGS_DIR / "completely_random.log")
assert lw._detect_branch_from_path(path) == "UNKNOWN"
def test_multiple_aipass_segments(self):
"""First valid aipass/branch/logs/ match wins."""
lw = _import_log_watcher()
path = str(Path("/src") / "aipass" / "trigger" / "logs" / "inner.log")
assert lw._detect_branch_from_path(path) == "TRIGGER"
def test_aipass_without_logs_subdir(self):
"""aipass/branch without /logs/ segment returns UNKNOWN."""
lw = _import_log_watcher()
path = str(Path("/src") / "aipass" / "drone" / "core.log")
assert lw._detect_branch_from_path(path) == "UNKNOWN"
def test_system_logs_ai_mail_prefix(self):
"""Multi-word prefix (ai_mail) is matched correctly."""
lw = _import_log_watcher()
path = str(lw.SYSTEM_LOGS_DIR / "ai_mail_delivery.log")
assert lw._detect_branch_from_path(path) == "AI_MAIL"
# ---------------------------------------------------------------------------
# Tests -- _parse_prax_log_line (additional edge cases)
# ---------------------------------------------------------------------------
class TestParsePraxLogLineEdgeCases:
"""Additional edge cases for _parse_prax_log_line."""
def test_dash_format_critical(self):
"""Dash format with CRITICAL level is accepted."""
lw = _import_log_watcher()
line = "2026-04-26 10:00:00,100 - core - CRITICAL - System down"
result = lw._parse_prax_log_line(line)
assert result is not None
assert result["level"] == "CRITICAL"
assert result["module"] == "core"
assert result["message"] == "System down"
def test_dash_format_info_returns_none(self):
"""Dash format with INFO level returns None."""
lw = _import_log_watcher()
line = "2026-04-26 10:00:00,100 - core - INFO - All is well"
assert lw._parse_prax_log_line(line) is None
def test_pipe_format_too_few_parts(self):
"""Pipe format with fewer than 4 parts returns None."""
lw = _import_log_watcher()
line = "2026-04-26 10:00:00 | only_two_parts"
assert lw._parse_prax_log_line(line) is None
def test_dash_format_too_few_parts(self):
"""Dash format with fewer than 4 parts returns None."""
lw = _import_log_watcher()
line = "2026-04-26 - module_only"
assert lw._parse_prax_log_line(line) is None
def test_pipe_format_warning_level_returns_none(self):
"""Pipe format with WARNING level (not error) returns None."""
lw = _import_log_watcher()
line = "2026-04-26 10:00:00 | mod | WARNING | caution"
assert lw._parse_prax_log_line(line) is None
def test_dash_format_debug_level_returns_none(self):
"""Dash format with DEBUG level returns None."""
lw = _import_log_watcher()
line = "2026-04-26 10:00:00,100 - mod - DEBUG - tracing"
assert lw._parse_prax_log_line(line) is None
# ---------------------------------------------------------------------------
# Tests -- BranchLogWatcher._should_process (additional edge cases)
# ---------------------------------------------------------------------------
class TestShouldProcessEdgeCases:
"""Additional edge cases for BranchLogWatcher._should_process."""
def test_excluded_file_case_insensitive(self):
"""Exclusion matching is case-insensitive."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
path = str(Path("/src") / "aipass" / "flow" / "logs" / "DISPATCH.LOG")
assert watcher._should_process(path) is False
def test_non_log_extension_py(self):
""".py file is rejected."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
path = str(Path("/src") / "aipass" / "flow" / "logs" / "handler.py")
assert watcher._should_process(path) is False
def test_excluded_trigger_log_watcher(self):
"""trigger_log_watcher.log is excluded (self-referential)."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
path = str(Path("/src") / "aipass" / "trigger" / "logs" / "trigger_log_watcher.log")
assert watcher._should_process(path) is False
def test_excluded_medic_suppressed(self):
"""medic_suppressed.log is excluded."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
path = str(Path("/src") / "aipass" / "flow" / "logs" / "medic_suppressed.log")
assert watcher._should_process(path) is False
# ---------------------------------------------------------------------------
# Tests -- BranchLogWatcher._read_new_lines (deeper coverage)
# ---------------------------------------------------------------------------
class TestReadNewLinesDeeper:
"""Deeper coverage for BranchLogWatcher._read_new_lines."""
def test_no_read_when_size_unchanged(self, tmp_path):
"""When file size equals last position, no reading occurs."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
log_file = tmp_path / "unchanged.log"
log_file.write_text("content\n", encoding="utf-8")
file_path = str(log_file)
current_size = log_file.stat().st_size
watcher.log_positions[file_path] = current_size
watcher._process_log_line = MagicMock()
with patch.object(lw, "_save_log_positions"):
watcher._read_new_lines(file_path)
watcher._process_log_line.assert_not_called()
assert watcher.log_positions[file_path] == current_size
def test_position_save_interval_triggers_save(self, tmp_path):
"""_save_log_positions is called when counter hits interval."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
log_file = tmp_path / "interval.log"
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
log_file.write_text(f"{now} | mod | ERROR | fail\n", encoding="utf-8")
file_path = str(log_file)
watcher.log_positions[file_path] = 0
watcher._position_save_counter = watcher._POSITION_SAVE_INTERVAL - 1
with patch.object(lw, "_save_log_positions") as mock_save:
watcher._read_new_lines(file_path)
mock_save.assert_called_once()
assert watcher._position_save_counter == 0
def test_position_save_interval_not_reached(self, tmp_path):
"""_save_log_positions is NOT called when counter is below interval."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
log_file = tmp_path / "notsaved.log"
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
log_file.write_text(f"{now} | mod | ERROR | fail\n", encoding="utf-8")
file_path = str(log_file)
watcher.log_positions[file_path] = 0
watcher._position_save_counter = 0
with patch.object(lw, "_save_log_positions") as mock_save:
watcher._read_new_lines(file_path)
mock_save.assert_not_called()
def test_blank_lines_are_skipped(self, tmp_path):
"""Blank lines in new content do not trigger _process_log_line."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
log_file = tmp_path / "blanks.log"
log_file.write_text("\n\n\n", encoding="utf-8")
file_path = str(log_file)
watcher.log_positions[file_path] = 0
watcher._process_log_line = MagicMock()
with patch.object(lw, "_save_log_positions"):
watcher._read_new_lines(file_path)
watcher._process_log_line.assert_not_called()
def test_file_truncated_resets_position(self, tmp_path):
"""When file is smaller than stored position, resets to 0."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
log_file = tmp_path / "truncated.log"
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
log_file.write_text(f"{now} | mod | ERROR | after rotation\n", encoding="utf-8")
file_path = str(log_file)
watcher.log_positions[file_path] = 99999
watcher._process_log_line = MagicMock()
with patch.object(lw, "_save_log_positions"):
watcher._read_new_lines(file_path)
watcher._process_log_line.assert_called_once()
# ---------------------------------------------------------------------------
# Tests -- BranchLogWatcher._process_log_line (deeper coverage)
# ---------------------------------------------------------------------------
_BRANCH_LOG_PATH = str(Path("/src") / "aipass" / "flow" / "logs" / "flow.log")
class TestProcessLogLineDeeper:
"""Deeper coverage for BranchLogWatcher._process_log_line."""
def _make_error_line(self, message: str = "Something broke") -> str:
"""Build a fresh ERROR line with current timestamp."""
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
return f"{now} | test_mod | ERROR | {message}"
def test_non_error_line_returns_early(self):
"""INFO-level line parsed as None, no event fired."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
watcher._process_log_line(f"{now} | mod | INFO | Fine", "/a.log")
fire.assert_not_called()
def test_semantic_exclusion_error_hash(self):
"""Line containing 'error_hash' in message is skipped."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
line = self._make_error_line("Processed error error_hash=abc123")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_not_called()
def test_semantic_exclusion_fingerprint(self):
"""Line containing 'fingerprint' in message is skipped."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
line = self._make_error_line("Error with fingerprint=xyz789")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_not_called()
def test_semantic_exclusion_registry_id(self):
"""Line containing 'registry_id' in message is skipped."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
line = self._make_error_line("Logged with registry_id=r001")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_not_called()
def test_stale_entry_skipped(self):
"""Line with old timestamp is skipped."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
old = (datetime.now() - timedelta(seconds=600)).strftime("%Y-%m-%d %H:%M:%S.%f")
line = f"{old} | mod | ERROR | Old error"
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_not_called()
def test_registry_path_fires_event_with_registry_data(self):
"""Registry available: fires event with registry metadata."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
lw._REGISTRY_AVAILABLE = True
mock_report = MagicMock(
return_value={
"is_new": True,
"count": 1,
"id": "reg123",
"fingerprint": "fp456",
"first_seen": "2026-04-26",
"last_seen": "2026-04-26",
}
)
lw.registry_report = mock_report
watcher = lw.BranchLogWatcher()
line = self._make_error_line("DB connection lost")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_called_once()
call_kwargs = fire.call_args[1]
assert call_kwargs["branch"] == "FLOW"
assert call_kwargs["message"] == "DB connection lost"
assert call_kwargs["registry_id"] == "reg123"
assert call_kwargs["fingerprint"] == "fp456"
assert call_kwargs["count"] == 1
def test_registry_path_fire_event_none_logs_warning(self):
"""Registry available but _fire_event is None: logs warning."""
lw = _import_log_watcher()
lw._fire_event = None
lw._REGISTRY_AVAILABLE = True
mock_report = MagicMock(
return_value={
"is_new": True,
"count": 1,
"id": "reg999",
}
)
lw.registry_report = mock_report
watcher = lw.BranchLogWatcher()
line = self._make_error_line("No callback set")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
def test_registry_report_exception_falls_to_fallback(self):
"""Registry report raises: falls through to fallback path."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
lw._REGISTRY_AVAILABLE = True
lw.registry_report = MagicMock(side_effect=RuntimeError("registry down"))
with patch.dict(
sys.modules,
{"aipass.trigger.apps.handlers.error_registry": None},
):
watcher = lw.BranchLogWatcher()
lw._fallback_error_counts.clear()
line = self._make_error_line("Fallback triggered")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_called_once()
call_kwargs = fire.call_args[1]
assert call_kwargs["message"] == "Fallback triggered"
assert call_kwargs["count"] == 1
def test_fallback_lazy_import_succeeds(self):
"""Fallback path: lazy import succeeds, fires event."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
lw._REGISTRY_AVAILABLE = False
watcher = lw.BranchLogWatcher()
line = self._make_error_line("Lazy import works")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
fire.assert_called_once()
call_kwargs = fire.call_args[1]
assert call_kwargs["message"] == "Lazy import works"
def test_local_count_tracking_increments(self):
"""Local count path: repeated errors increment counter."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
lw._REGISTRY_AVAILABLE = True
lw.registry_report = MagicMock(side_effect=RuntimeError("registry down"))
lw._fallback_error_counts.clear()
with patch.dict(
sys.modules,
{"aipass.trigger.apps.handlers.error_registry": None},
):
watcher = lw.BranchLogWatcher()
line = self._make_error_line("Repeated failure")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
watcher._process_log_line(line, _BRANCH_LOG_PATH)
assert fire.call_count == 2
second_call = fire.call_args_list[1][1]
assert second_call["count"] == 2
def test_local_count_fire_event_none_logs_warning(self):
"""Local count with _fire_event=None: logs warning, no crash."""
lw = _import_log_watcher()
lw._fire_event = None
lw._REGISTRY_AVAILABLE = True
lw.registry_report = MagicMock(side_effect=RuntimeError("registry down"))
lw._fallback_error_counts.clear()
with patch.dict(
sys.modules,
{"aipass.trigger.apps.handlers.error_registry": None},
):
watcher = lw.BranchLogWatcher()
line = self._make_error_line("No callback at all")
watcher._process_log_line(line, _BRANCH_LOG_PATH)
def test_outer_exception_handler_catches_unexpected(self):
"""Outer try/except catches unexpected errors without raising."""
lw = _import_log_watcher()
fire = MagicMock()
lw.set_event_callback(fire)
watcher = lw.BranchLogWatcher()
with patch.object(
lw,
"_parse_prax_log_line",
side_effect=TypeError("boom"),
):
watcher._process_log_line("any line", "/any/path.log")
fire.assert_not_called()