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

feat(trigger): test(trigger): improve line coverage on log watcher + entry point

Browse Source 
Co-Authored-By: @trigger <trigger@aipass>
This commit is contained in:
AIOSAI
2026-04-26 07:56:22 -07:00
co-authored by @trigger
parent 8601265d75
commit 87fb6c8359
4 changed files with 741 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/trigger/.seedgo/bypass.json
+32
View File
@@ -391,6 +391,38 @@
"standard": "silent_catch",
"lines": [233],
"reason": "Timestamp format parsing loop — tries multiple datetime formats, ValueError is expected control flow on non-matching formats."
},
{
"file": "tests/test_log_watcher_service.py",
"standard": "architecture",
"pattern": "3-layer structure",
"reason": "Test file — tests/ is the standard location for unit tests, not part of the apps/modules/handlers source tree."
},
{
"file": "tests/test_log_watcher.py",
"standard": "architecture",
"pattern": "3-layer structure",
"reason": "Test file — tests/ is the standard location for unit tests, not part of the apps/modules/handlers source tree."
},
{
"file": "tests/test_log_watcher.py",
"standard": "encapsulation",
"lines": [82],
"pattern": "handler imported directly",
"reason": "Test helper _import_log_watcher() must import the handler module directly to test it. All trigger test files follow this pattern."
},
{
"file": "tests/test_log_watcher.py",
"standard": "log_structure",
"lines": [133, 310, 316],
"pattern": "hardcoded log paths",
"reason": "Test assertions use hardcoded log path strings as fixture data to verify path-detection logic. These are not production paths."
},
{
"file": "tests/test_trigger_entry.py",
"standard": "architecture",
"pattern": "3-layer structure",
"reason": "Test file — tests/ is the standard location for unit tests, not part of the apps/modules/handlers source tree."
}
],
"notes": {
src/aipass/trigger/tests/test_log_watcher.py
+10
View File
@@ -8,6 +8,7 @@
# Modified: 2026-04-03
# =============================================
import json
import sys
import hashlib
from datetime import datetime, timedelta
@@ -491,6 +492,7 @@ class TestOnModified:
"""Tests for BranchLogWatcher.on_modified."""
def test_skips_directory_events(self):
"""Directory events are ignored by on_modified."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
watcher._read_new_lines = MagicMock()
@@ -501,6 +503,7 @@ class TestOnModified:
watcher._read_new_lines.assert_not_called()
def test_skips_excluded_files(self):
"""Files that fail _should_process are not read."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
watcher._should_process = MagicMock(return_value=False)
@@ -512,6 +515,7 @@ class TestOnModified:
watcher._read_new_lines.assert_not_called()
def test_processes_valid_file(self):
"""Valid log file triggers _read_new_lines with correct path."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
watcher._should_process = MagicMock(return_value=True)
@@ -523,6 +527,7 @@ class TestOnModified:
watcher._read_new_lines.assert_called_once_with("/some/branch/logs/core.log")
def test_handles_read_exception(self):
"""IOError during _read_new_lines is handled without raising."""
lw = _import_log_watcher()
watcher = lw.BranchLogWatcher()
watcher._should_process = MagicMock(return_value=True)
@@ -542,6 +547,7 @@ class TestInitializePositions:
"""Tests for BranchLogWatcher.initialize_positions."""
def test_snaps_to_eof_when_no_persisted(self, tmp_path):
"""Without persisted positions, snaps all log files to EOF."""
lw = _import_log_watcher()
lw.AIPASS_PKG_ROOT = tmp_path / "aipass"
lw.SYSTEM_LOGS_DIR = tmp_path / "system_logs"
@@ -555,6 +561,7 @@ class TestInitializePositions:
assert watcher.log_positions[str(log_file)] == log_file.stat().st_size
def test_uses_persisted_position_when_valid(self, tmp_path):
"""Restores a persisted position that is within current file size."""
lw = _import_log_watcher()
lw.AIPASS_PKG_ROOT = tmp_path / "aipass"
lw.SYSTEM_LOGS_DIR = tmp_path / "system_logs"
@@ -569,6 +576,7 @@ class TestInitializePositions:
assert watcher.log_positions[str(log_file)] == saved_pos
def test_snaps_to_eof_when_persisted_beyond_size(self, tmp_path):
"""Resets to EOF when persisted position exceeds current file size."""
lw = _import_log_watcher()
lw.AIPASS_PKG_ROOT = tmp_path / "aipass"
lw.SYSTEM_LOGS_DIR = tmp_path / "system_logs"
@@ -582,6 +590,7 @@ class TestInitializePositions:
assert watcher.log_positions[str(log_file)] == log_file.stat().st_size
def test_skips_branches_without_logs_dir(self, tmp_path):
"""Branch directories without a logs/ subdirectory are skipped."""
lw = _import_log_watcher()
lw.AIPASS_PKG_ROOT = tmp_path / "aipass"
lw.SYSTEM_LOGS_DIR = tmp_path / "system_logs"
@@ -592,6 +601,7 @@ class TestInitializePositions:
assert len(watcher.log_positions) == 0
def test_initializes_system_logs(self, tmp_path):
"""System log files are initialized to EOF during position setup."""
lw = _import_log_watcher()
lw.AIPASS_PKG_ROOT = tmp_path / "aipass"
(tmp_path / "aipass").mkdir(parents=True)
src/aipass/trigger/tests/test_log_watcher_service.py
+286
View File
@@ -0,0 +1,286 @@
# =================== AIPass ====================
# Name: test_log_watcher_service.py
# Description: Tests for the log watcher service entry point
# Version: 1.0.0
# Created: 2026-04-26
# Modified: 2026-04-26
# =============================================
"""Tests for log_watcher_service — the persistent service entry point."""
import signal
import sys
import threading
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_infrastructure(monkeypatch: pytest.MonkeyPatch) -> None:
"""Mock heavy watcher module imports before log_watcher_service loads."""
# Mock branch_log_events module
mock_branch = MagicMock()
mock_branch.start = MagicMock(return_value=True)
mock_branch.stop = MagicMock()
monkeypatch.setitem(
sys.modules,
"aipass.trigger.apps.modules.branch_log_events",
mock_branch,
)
# Mock log_events module
mock_system = MagicMock()
mock_system.start = MagicMock(return_value=True)
mock_system.stop = MagicMock()
monkeypatch.setitem(
sys.modules,
"aipass.trigger.apps.modules.log_events",
mock_system,
)
# Force re-import so mocks take effect
monkeypatch.delitem(
sys.modules,
"aipass.trigger.apps.log_watcher_service",
raising=False,
)
def _import_module():
"""Import log_watcher_service fresh (after mocks are in place)."""
import aipass.trigger.apps.log_watcher_service as mod
return mod
# ---------------------------------------------------------------------------
# Tests -- print_introspection
# ---------------------------------------------------------------------------
class TestPrintIntrospection:
"""Tests for print_introspection output."""
def test_prints_module_name(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Output includes the module name."""
mod = _import_module()
mod.print_introspection()
captured = capsys.readouterr()
assert "log_watcher_service" in captured.out
def test_prints_description(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Output includes the service description."""
mod = _import_module()
mod.print_introspection()
captured = capsys.readouterr()
assert "systemd service" in captured.out
# ---------------------------------------------------------------------------
# Tests -- main
# ---------------------------------------------------------------------------
class TestMain:
"""Tests for main() entry point."""
def test_both_watchers_start(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When both watchers start successfully, prints both names."""
mod = _import_module()
# Make stop_event.wait() return immediately
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
# Both start functions return True (already the default from fixture)
mod.start_branch_watcher = MagicMock(return_value=True)
mod.start_system_watcher = MagicMock(return_value=True)
captured: list[str] = []
monkeypatch.setattr("builtins.print", lambda *a, **kw: captured.append(str(a)))
mod.main()
output = " ".join(captured)
assert "branch" in output
assert "system" in output
assert "Stopped" in output
def test_only_branch_watcher_starts(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When only branch watcher starts, prints only 'branch'."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
mod.start_branch_watcher = MagicMock(return_value=True)
mod.start_system_watcher = MagicMock(return_value=None)
captured: list[str] = []
monkeypatch.setattr("builtins.print", lambda *a, **kw: captured.append(str(a)))
mod.main()
output = " ".join(captured)
assert "branch" in output
assert "system" not in output.replace("Stopped", "").split("Running")[1] if "Running" in output else True
def test_only_system_watcher_starts(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When only system watcher starts, prints only 'system'."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
mod.start_branch_watcher = MagicMock(return_value=None)
mod.start_system_watcher = MagicMock(return_value=True)
captured: list[str] = []
monkeypatch.setattr("builtins.print", lambda *a, **kw: captured.append(str(a)))
mod.main()
output = " ".join(captured)
assert "system" in output
def test_both_fail_exits_1(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When both watchers fail, exits with code 1."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
mod.start_branch_watcher = MagicMock(return_value=None)
mod.start_system_watcher = MagicMock(return_value=None)
with pytest.raises(SystemExit) as exc_info:
mod.main()
assert exc_info.value.code == 1
def test_calls_stop_on_shutdown(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""After stop_event is set, both stop functions are called."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
mod.start_branch_watcher = MagicMock(return_value=True)
mod.start_system_watcher = MagicMock(return_value=True)
mock_stop_branch = MagicMock()
mock_stop_system = MagicMock()
mod.stop_branch_watcher = mock_stop_branch
mod.stop_system_watcher = mock_stop_system
# Suppress print output
monkeypatch.setattr("builtins.print", lambda *a, **kw: None)
mod.main()
mock_stop_branch.assert_called_once()
mock_stop_system.assert_called_once()
def test_signal_handler_sets_stop_event(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The shutdown closure sets the stop_event when invoked."""
mod = _import_module()
# Use a real Event but do NOT pre-set it; we will trigger it via
# the signal handler that main() installs.
real_event = threading.Event()
monkeypatch.setattr(threading, "Event", lambda: real_event)
# Capture the signal handler that main() registers (no-op stub
# avoids calling real signal.signal which fails in threads).
installed_handlers: dict[int, object] = {}
def capture_signal(signum: int, handler: object) -> object:
"""Record installed signal handler for later inspection."""
installed_handlers[signum] = handler
return signal.SIG_DFL
monkeypatch.setattr(signal, "signal", capture_signal)
mod.start_branch_watcher = MagicMock(return_value=True)
mod.start_system_watcher = MagicMock(return_value=True)
# Suppress print output
monkeypatch.setattr("builtins.print", lambda *a, **kw: None)
# main() is called on the main thread; capture_signal is a no-op so
# signal registration is safe. The pre-set Event makes wait() return
# immediately, but we need it NOT set yet so we can trigger via handler.
# Instead, run main() in a background thread.
t = threading.Thread(target=mod.main, daemon=True)
t.start()
# Give it a moment to register signals and start waiting
import time
time.sleep(0.05)
# Invoke the captured SIGTERM handler
assert signal.SIGTERM in installed_handlers
handler = installed_handlers[signal.SIGTERM]
handler(signal.SIGTERM, None) # type: ignore[operator]
# The event should now be set, unblocking main()
assert real_event.is_set()
t.join(timeout=2)
assert not t.is_alive()
def test_registers_both_signal_handlers(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""main() registers handlers for both SIGTERM and SIGINT."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
installed_signals: list[int] = []
original_signal = signal.signal
def capture_signal(signum: int, handler: object) -> object:
"""Record which signal numbers are registered."""
installed_signals.append(signum)
return original_signal(signum, signal.SIG_DFL)
monkeypatch.setattr(signal, "signal", capture_signal)
mod.start_branch_watcher = MagicMock(return_value=True)
mod.start_system_watcher = MagicMock(return_value=True)
monkeypatch.setattr("builtins.print", lambda *a, **kw: None)
mod.main()
assert signal.SIGTERM in installed_signals
assert signal.SIGINT in installed_signals
def test_both_fail_prints_stderr(self, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
"""When both watchers fail, error message goes to stderr."""
mod = _import_module()
pre_set = threading.Event()
pre_set.set()
monkeypatch.setattr(threading, "Event", lambda: pre_set)
mod.start_branch_watcher = MagicMock(return_value=False)
mod.start_system_watcher = MagicMock(return_value=False)
with pytest.raises(SystemExit):
mod.main()
captured = capsys.readouterr()
assert "Both watchers failed" in captured.err
src/aipass/trigger/tests/test_trigger_entry.py
+413
View File
@@ -0,0 +1,413 @@
# =================== AIPass ====================
# Name: test_trigger_entry.py
# Description: Tests for trigger.py CLI entry point — line coverage
# Version: 1.0.0
# Created: 2026-04-26
# Modified: 2026-04-26
# =============================================
"""Tests for the trigger.py CLI entry point (discover, route, main)."""
import sys
from pathlib import Path
from unittest.mock import MagicMock
import pytest
# ---------------------------------------------------------------------------
# Infrastructure mocks — isolate from real prax / cli
# ---------------------------------------------------------------------------
_mock_console = MagicMock()
_mock_header = MagicMock()
_mock_error = MagicMock()
_mock_logger = MagicMock()
@pytest.fixture(autouse=True)
def _mock_infrastructure(monkeypatch: pytest.MonkeyPatch) -> None:
"""Replace heavy infrastructure imports with lightweight mocks."""
# Reset call counts between tests
_mock_console.reset_mock()
_mock_header.reset_mock()
_mock_error.reset_mock()
_mock_logger.reset_mock()
# ---- prax logger ----
prax_logger_mod = MagicMock()
prax_logger_mod.system_logger = _mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", MagicMock())
monkeypatch.setitem(sys.modules, "aipass.prax.apps", MagicMock())
monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules", MagicMock())
monkeypatch.setitem(sys.modules, "aipass.prax.apps.modules.logger", prax_logger_mod)
# ---- cli ----
mock_cli = MagicMock()
mock_cli.console = _mock_console
mock_cli.header = _mock_header
mock_cli.error = _mock_error
monkeypatch.setitem(sys.modules, "aipass.cli", MagicMock())
monkeypatch.setitem(sys.modules, "aipass.cli.apps", MagicMock())
monkeypatch.setitem(sys.modules, "aipass.cli.apps.modules", mock_cli)
# ---- force re-import so the module picks up our mocks ----
monkeypatch.delitem(sys.modules, "aipass.trigger.apps.trigger", raising=False)
def _import_trigger():
"""Import trigger.py fresh (after infrastructure mocks are in place)."""
import aipass.trigger.apps.trigger as mod
return mod
# ===================================================================
# discover_modules
# ===================================================================
class TestDiscoverModules:
"""Cover discover_modules() paths."""
def test_discovers_modules_with_handle_command(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Module .py with handle_command() is discovered."""
mod_file = tmp_path / "good_mod.py"
mod_file.write_text(
"def handle_command(command, args):\n return False\n",
encoding="utf-8",
)
trigger = _import_trigger()
monkeypatch.setattr(trigger, "MODULES_DIR", tmp_path)
# We need importlib to actually find the module, so patch import_module
fake_module = MagicMock()
fake_module.handle_command = MagicMock()
monkeypatch.setattr(trigger.importlib, "import_module", lambda name: fake_module)
result = trigger.discover_modules()
assert fake_module in result
def test_skips_underscore_files(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Files starting with _ (e.g. __init__.py) are skipped."""
(tmp_path / "__init__.py").write_text("# init", encoding="utf-8")
(tmp_path / "_private.py").write_text("# private", encoding="utf-8")
trigger = _import_trigger()
monkeypatch.setattr(trigger, "MODULES_DIR", tmp_path)
import_called = False
def _no_import(name: str):
nonlocal import_called
import_called = True
monkeypatch.setattr(trigger.importlib, "import_module", _no_import)
result = trigger.discover_modules()
assert result == []
assert not import_called
def test_skips_modules_without_handle_command(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Module lacking handle_command() is silently skipped."""
(tmp_path / "no_handler.py").write_text("x = 1\n", encoding="utf-8")
trigger = _import_trigger()
monkeypatch.setattr(trigger, "MODULES_DIR", tmp_path)
fake_module = MagicMock(spec=[]) # spec=[] means NO attributes
monkeypatch.setattr(trigger.importlib, "import_module", lambda name: fake_module)
result = trigger.discover_modules()
assert result == []
def test_handles_import_error(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Module that raises on import is skipped and logged."""
(tmp_path / "bad_mod.py").write_text("raise RuntimeError('boom')\n", encoding="utf-8")
trigger = _import_trigger()
monkeypatch.setattr(trigger, "MODULES_DIR", tmp_path)
monkeypatch.setattr(
trigger.importlib,
"import_module",
MagicMock(side_effect=ImportError("boom")),
)
result = trigger.discover_modules()
assert result == []
_mock_logger.error.assert_called()
def test_returns_empty_when_dir_missing(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
"""Non-existent MODULES_DIR returns empty list and warns."""
trigger = _import_trigger()
monkeypatch.setattr(trigger, "MODULES_DIR", tmp_path / "nope")
result = trigger.discover_modules()
assert result == []
_mock_logger.warning.assert_called()
# ===================================================================
# route_command
# ===================================================================
class TestRouteCommand:
"""Cover route_command() paths."""
def test_routes_to_first_matching_module(self) -> None:
"""Module returning True from handle_command is accepted."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.handle_command.return_value = True
assert trigger.route_command("fire", [], [mock_mod]) is True
mock_mod.handle_command.assert_called_once_with("fire", [])
def test_returns_false_when_no_module_handles(self) -> None:
"""All modules return False so route_command returns False."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.handle_command.return_value = False
assert trigger.route_command("bogus", [], [mock_mod]) is False
def test_handles_module_exception(self) -> None:
"""Exception in handle_command is caught and logged."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.__name__ = "aipass.trigger.apps.modules.broken"
mock_mod.handle_command.side_effect = RuntimeError("boom")
assert trigger.route_command("fire", [], [mock_mod]) is False
_mock_logger.error.assert_called()
def test_stops_after_first_handler(self) -> None:
"""Only the first module that returns True is used."""
trigger = _import_trigger()
mod_a = MagicMock()
mod_a.handle_command.return_value = True
mod_b = MagicMock()
mod_b.handle_command.return_value = True
assert trigger.route_command("cmd", ["a"], [mod_a, mod_b]) is True
mod_a.handle_command.assert_called_once()
mod_b.handle_command.assert_not_called()
def test_tries_next_module_on_false(self) -> None:
"""When first module returns False, second is tried."""
trigger = _import_trigger()
mod_a = MagicMock()
mod_a.handle_command.return_value = False
mod_b = MagicMock()
mod_b.handle_command.return_value = True
assert trigger.route_command("cmd", [], [mod_a, mod_b]) is True
mod_a.handle_command.assert_called_once()
mod_b.handle_command.assert_called_once()
def test_empty_modules_returns_false(self) -> None:
"""Empty module list means nothing can handle the command."""
trigger = _import_trigger()
assert trigger.route_command("anything", [], []) is False
# ===================================================================
# print_introspection
# ===================================================================
class TestPrintIntrospection:
"""Cover print_introspection() paths."""
def test_prints_module_info_with_doc(self) -> None:
"""Module with __doc__ gets its first line as description."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.__name__ = "aipass.trigger.apps.modules.fire"
mock_mod.__doc__ = "Fire all the things\nSecond line ignored"
trigger.print_introspection([mock_mod])
# Verify console.print was called with the module name
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "fire" in joined
def test_prints_module_info_without_doc(self) -> None:
"""Module with __doc__=None shows 'No description'."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.__name__ = "aipass.trigger.apps.modules.silent"
mock_mod.__doc__ = None
trigger.print_introspection([mock_mod])
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "No description" in joined
def test_prints_no_modules_message(self) -> None:
"""Empty module list shows 'No modules discovered'."""
trigger = _import_trigger()
trigger.print_introspection([])
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "No modules discovered" in joined
# ===================================================================
# print_help
# ===================================================================
class TestPrintHelp:
"""Cover print_help() paths."""
def test_prints_help_with_modules_and_doc(self) -> None:
"""Help output includes module name and docstring first line."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.__name__ = "aipass.trigger.apps.modules.status"
mock_mod.__doc__ = "Show status information\nDetails"
trigger.print_help([mock_mod])
_mock_header.assert_called()
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "status" in joined
def test_prints_help_with_modules_no_doc(self) -> None:
"""Help output shows 'No description' when module lacks docstring."""
trigger = _import_trigger()
mock_mod = MagicMock()
mock_mod.__name__ = "aipass.trigger.apps.modules.quiet"
mock_mod.__doc__ = None
trigger.print_help([mock_mod])
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "No description" in joined
def test_prints_help_no_modules(self) -> None:
"""Help output shows 'No modules discovered' for empty list."""
trigger = _import_trigger()
trigger.print_help([])
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "No modules discovered" in joined
# ===================================================================
# main
# ===================================================================
class TestMain:
"""Cover main() paths."""
def test_no_args_shows_introspection(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""No CLI args triggers print_introspection and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
# print_introspection prints "No modules discovered"
calls = [str(c) for c in _mock_console.print.call_args_list]
joined = " ".join(calls)
assert "No modules discovered" in joined
def test_version_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The --version flag prints version string and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "--version"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
_mock_console.print.assert_any_call("TRIGGER v2.2.0")
def test_version_short_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The -V short flag prints version string and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "-V"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
_mock_console.print.assert_any_call("TRIGGER v2.2.0")
def test_help_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The --help flag calls print_help and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "--help"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
_mock_header.assert_called()
def test_help_short_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The -h short flag calls print_help and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "-h"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
def test_help_word(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""The bare 'help' command calls print_help and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "help"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 0
def test_valid_command_routes(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Known command is routed to the matching module and returns 0."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "fire", "startup"])
mock_mod = MagicMock()
mock_mod.handle_command.return_value = True
monkeypatch.setattr(trigger, "discover_modules", lambda: [mock_mod])
result = trigger.main()
assert result == 0
mock_mod.handle_command.assert_called_once_with("fire", ["startup"])
def test_valid_command_no_extra_args(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Command with no trailing args passes empty list."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "fire"])
mock_mod = MagicMock()
mock_mod.handle_command.return_value = True
monkeypatch.setattr(trigger, "discover_modules", lambda: [mock_mod])
result = trigger.main()
assert result == 0
mock_mod.handle_command.assert_called_once_with("fire", [])
def test_unknown_command_returns_1(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Unrecognised command calls error() and returns 1."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "bogus"])
mock_mod = MagicMock()
mock_mod.handle_command.return_value = False
monkeypatch.setattr(trigger, "discover_modules", lambda: [mock_mod])
result = trigger.main()
assert result == 1
_mock_error.assert_called_once()
def test_unknown_command_error_message(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Error call includes the unknown command name and suggestion."""
trigger = _import_trigger()
monkeypatch.setattr(sys, "argv", ["trigger", "xyzzy"])
monkeypatch.setattr(trigger, "discover_modules", lambda: [])
result = trigger.main()
assert result == 1
args, kwargs = _mock_error.call_args
assert "xyzzy" in args[0]
assert "suggestion" in kwargs