#661 exit-code foundation: @cli resolve_exit + error() auto-trip (inert) + @seedgo output_routing checker + devpulse reference adoption

- @cli: process failure-flag + resolve_exit(handled)->0/1/2; error() auto-trips the flag; inert until a branch adopts it; 10 tests, seedgo 100%
- @seedgo: new output_routing standard (39th checker) flags user-facing status output bypassing cli helpers; 254-site per-branch migration checklist; precise (0 FP, dogfooded on devpulse); 1178 tests
- devpulse: first adopter — main()->reset_command_state()+resolve_exit(); feedback module+handlers migrated raw-red/logger.error->error(); exit 2/0/1 verified; 380 tests green; seedgo 100%
- CHANGELOG updated. Fleet migration (remaining 13 branches) to follow. Tracked in DPLAN-0236.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HeaAmr3oj2ZexB6ng311Vj
This commit is contained in:
AIOSAI
2026-07-09 19:38:17 -07:00
co-authored by Claude Opus 4.8
parent 1edea552bf
commit d2d1adca0a
13 changed files with 978 additions and 14 deletions
+17
View File
@@ -11,6 +11,23 @@ PyPI version — not the changelog header.
## [2026-07-09]
### Added
- **Exit-code contract foundation — failing commands can now exit non-zero
(issue #661, in progress).** CLI error paths printed an error but returned exit
`0`, so `$?`-checking callers (core to running `drone` as a subprocess) were
told success on failure. The dispatch contract was a 2-state bool (`handled` /
`not-mine`) with no way to say "handled *and* failed". `@cli` now exposes a
process-level failure flag + `resolve_exit(handled)` (→ `0`/`1`/`2`), and
`error()` auto-trips the flag — so any failure routed through `error()` gets a
correct non-zero exit with zero per-site edits, and it can't regress. Inert
until a branch's `main()` adopts it. `@seedgo` added an `output_routing`
standard (39th checker) flagging user-facing status output that bypasses the
cli helpers — 254 sites across 14 branches, the migration checklist. `devpulse`
is the first adopter (`main()`→`resolve_exit`, feedback migrated to `error()`,
exit `2`/`0`/`1` verified, 100% seedgo). Fleet migration to follow.
(built by @cli + @seedgo)
### Fixed
- **macOS session lock-out: the boot wrapper can now see tmux sessions on
+8
View File
@@ -26,6 +26,9 @@ from aipass.cli.apps.modules.display import console, err_console
# Display functions
from aipass.cli.apps.modules.display import header, success, error, warning, fatal, section
# Exit-code failure-flag API
from aipass.cli.apps.modules.display import mark_command_failed, command_failed, reset_command_state, resolve_exit
# Operation templates
from aipass.cli.apps.modules.templates import operation_start, operation_complete
@@ -40,6 +43,11 @@ __all__ = [
"warning",
"fatal",
"section",
# Exit-code failure-flag API
"mark_command_failed",
"command_failed",
"reset_command_state",
"resolve_exit",
# Templates
"operation_start",
"operation_complete",
+35
View File
@@ -48,6 +48,36 @@ err_console = Console(stderr=True, force_terminal=sys.stderr.isatty()) # Stderr
_TRIGGER = None
_TRIGGER_LOADED = False
# Process-level command failure flag — mutable container avoids global statement
_CMD_STATE = {"failed": False}
def mark_command_failed() -> None:
"""Set the process-level failure flag (called automatically by error())."""
_CMD_STATE["failed"] = True
def command_failed() -> bool:
"""Return whether mark_command_failed() has been called since last reset."""
return _CMD_STATE["failed"]
def reset_command_state() -> None:
"""Reset the failure flag to False (for tests and main() entry)."""
_CMD_STATE["failed"] = False
def resolve_exit(handled: bool) -> int:
"""Map handled/failed state to an exit code.
Returns 1 if not handled, 2 if handled but failed, 0 otherwise.
"""
if not handled:
return 1
if _CMD_STATE["failed"]:
return 2
return 0
# ============================================================================
# MODULE PATTERN FUNCTIONS (SEEDGO compliant)
@@ -341,6 +371,7 @@ def error(message: str, suggestion: str | None = None) -> None:
Example:
error('Branch not found', suggestion='Check branch name spelling')
"""
mark_command_failed()
err_console.print(f"❌ [red bold]{message}[/red bold]")
if suggestion:
err_console.print(f" [yellow]→ Try: {suggestion}[/yellow]")
@@ -411,6 +442,10 @@ __all__ = [
"warning",
"fatal",
"section",
"mark_command_failed",
"command_failed",
"reset_command_state",
"resolve_exit",
]
# ============================================================================
+59
View File
@@ -522,3 +522,62 @@ class TestInfrastructureMocking:
module_key = "aipass.cli.apps.modules.display"
assert module_key in sys.modules
assert sys.modules[module_key] is display
# =============================================================================
# Exit-code failure-flag tests
# =============================================================================
class TestCommandState:
"""Verify the process-level failure flag and resolve_exit truth table."""
def setup_method(self):
display.reset_command_state()
def teardown_method(self):
display.reset_command_state()
def test_initial_state_is_not_failed(self):
assert display.command_failed() is False
def test_mark_command_failed_sets_flag(self):
display.mark_command_failed()
assert display.command_failed() is True
def test_reset_command_state_clears_flag(self):
display.mark_command_failed()
display.reset_command_state()
assert display.command_failed() is False
def test_resolve_exit_not_handled(self):
assert display.resolve_exit(handled=False) == 1
def test_resolve_exit_handled_ok(self):
assert display.resolve_exit(handled=True) == 0
def test_resolve_exit_handled_failed(self):
display.mark_command_failed()
assert display.resolve_exit(handled=True) == 2
def test_resolve_exit_not_handled_ignores_flag(self):
display.mark_command_failed()
assert display.resolve_exit(handled=False) == 1
def test_error_trips_failure_flag(self):
cons, _ = _make_capture_console()
with patch.object(display, "err_console", cons):
display.error("something broke")
assert display.command_failed() is True
def test_warning_does_not_trip_flag(self):
cons, _ = _make_capture_console()
with patch.object(display, "err_console", cons):
display.warning("just a warning")
assert display.command_failed() is False
def test_success_does_not_trip_flag(self):
cons, _ = _make_capture_console()
with patch.object(display, "CONSOLE", cons):
display.success("all good")
assert display.command_failed() is False
@@ -79,6 +79,11 @@ drone @flow create . "Subject" aplan # APLAN (FPLAN/DPLAN
drone @flow list open # active plans
```
# Dispatch — in-flight comms
- **Steer a working agent with `email` (no wake), NOT `dispatch`.** `dispatch` = send **+ wake** (hand NEW work to a sleeping agent). An agent already running is awake, so `drone @ai_mail email @target "Subject" "Msg"` reaches it mid-task via its hook — no re-wake, no interrupt. Forgot something / need to correct a brief / add context → **email it in-flight**, don't re-dispatch. (`drone @ai_mail --help`)
- **No backticks in dispatch/email body strings** — bash runs `` `word` `` as command-substitution and silently eats it. Use single quotes or plain text (hit this live: a backtick'd word vanished from a brief).
# Watchdog
Devpulse module. After dispatch, arm as a background task — it polls the dispatch lock and exits when the agent finishes. Resolves @target → branch path → `.ai_mail.local/.dispatch.lock`. Default timeout 1800s; `drone @devpulse watchdog --help` for the full reference.
+3 -2
View File
@@ -35,7 +35,7 @@ if sys.platform == "win32":
_reconfigure(encoding="utf-8", errors="replace")
from aipass.prax import logger
from aipass.cli.apps.modules import err_console
from aipass.cli.apps.modules import err_console, resolve_exit, reset_command_state
console = err_console
@@ -161,13 +161,14 @@ def _handle_command(command: str, args: list) -> bool:
def main():
"""Main entry point - routes commands or shows help."""
reset_command_state()
args = sys.argv[1:]
if len(args) == 0:
print_introspection()
return 0
return 0 if _handle_command(args[0], args[1:]) else 1
return resolve_exit(_handle_command(args[0], args[1:]))
if __name__ == "__main__":
@@ -25,7 +25,7 @@ from aipass.devpulse.apps.handlers.feedback.storage import (
generate_id,
)
from aipass.cli.apps.modules import err_console
from aipass.cli.apps.modules import err_console, error
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -133,7 +133,7 @@ def reply_to(msg_id: str, body: str) -> bool:
break
if msg is None:
console.print(f"[red]Message {msg_id} not found.[/red]")
error(f"Message {msg_id} not found.")
return False
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
@@ -17,7 +17,7 @@ from rich.table import Table
from aipass.devpulse.apps.handlers.feedback.storage import load_inbox, save_inbox
from aipass.cli.apps.modules import err_console
from aipass.cli.apps.modules import err_console, error
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -67,7 +67,7 @@ def view_message(msg_id: str) -> None:
msg = _find_message(messages, msg_id)
if msg is None:
console.print(f"[red]Message {msg_id} not found.[/red]")
error(f"Message {msg_id} not found.")
return
# Mark as read
@@ -105,7 +105,7 @@ def clear_message(msg_id: str) -> None:
msg = _find_message(messages, msg_id)
if msg is None:
console.print(f"[red]Message {msg_id} not found.[/red]")
error(f"Message {msg_id} not found.")
return
was_unread = not msg.get("read")
+7 -7
View File
@@ -27,7 +27,7 @@ from aipass.devpulse.apps.handlers.feedback.compose import (
)
from aipass.prax import logger
from aipass.cli.apps.modules import err_console
from aipass.cli.apps.modules import err_console, error
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -93,14 +93,14 @@ def handle_command(command: str, args: list[str]) -> bool:
if subcommand == "view":
if not sub_args:
console.print("[red]Usage: feedback view <id>[/red]")
error("Usage: feedback view <id>")
return True
view_message(sub_args[0])
return True
if subcommand == "reply":
if len(sub_args) < 2:
console.print('[red]Usage: feedback reply <id> "message"[/red]')
error('Usage: feedback reply <id> "message"')
return True
msg_id = sub_args[0]
body = " ".join(sub_args[1:])
@@ -112,7 +112,7 @@ def handle_command(command: str, args: list[str]) -> bool:
if subcommand == "clear":
if not sub_args:
logger.error("Usage: feedback clear <id> | feedback clear --all")
error("Usage: feedback clear <id> | feedback clear --all")
return True
if sub_args[0] == "--all":
clear_all_read()
@@ -120,8 +120,8 @@ def handle_command(command: str, args: list[str]) -> bool:
clear_message(sub_args[0])
return True
console.print(f"[red]Unknown feedback subcommand: {subcommand}[/red]")
console.print("Use [bold]feedback --help[/bold] for usage.")
logger.warning("[feedback] unknown subcommand: %s", subcommand)
error(f"Unknown feedback subcommand: {subcommand}", suggestion="Use 'feedback --help' for usage")
return True
@@ -138,7 +138,7 @@ def _handle_send(args: list[str]) -> bool:
bool: Always True (command was handled).
"""
if len(args) < 2:
console.print('[red]Usage: feedback send "subject" "body"[/red]')
error('Usage: feedback send "subject" "body"')
console.print("[dim]Tip: from_branch is auto-detected or pass as first arg.[/dim]")
return True
@@ -0,0 +1,87 @@
# Output Routing Standard
**Status:** Active
**Date:** 2026-07-09
---
## What This Standard Is
User-facing error, success, and warning output must route through `@cli`'s semantic helpers (`error()`, `success()`, `warning()`) instead of raw `console.print()` with status markup or emojis.
## Why It Matters
1. **Consistent formatting** — all agents display errors, successes, and warnings the same way.
2. **Exit-code correctness** — `error()` carries the GitHub #661 failure-flag fix. Raw `console.print("[red]...")` bypasses it, causing error paths to exit 0.
3. **Stderr routing** — `error()` and `warning()` write to stderr; raw `console.print()` writes to stdout.
## What the Checker Scans For
Detects `console.print()` or `err_console.print()` calls containing status indicators:
- `[red]` or `[bold red]` markup (error-style output)
- Status emojis: `❌ ✅ ✓ ✗ ✘ ⚠ ✔`
- `[green]` paired with check emojis (`✓ ✔ ✅`)
- `[yellow]` paired with warning indicators (`⚠`, "warning", "WARN", "FAIL")
### Exclusions
- Lines inside docstrings (triple-quoted regions)
- Comment lines (`# ...`)
- `__init__.py` files
- Test files (`test_*.py`, `*_test.py`, `conftest.py`)
- `console.print()` with no status markup (tables, panels, informational text)
- Non-status color like `[cyan]`, `[dim]`, `[blue]`
## Code Examples
### Violation
```python
console.print(f"[red]Error: {msg}[/red]")
console.print("[bold red]Failed to process[/bold red]")
console.print(f"[green]✓[/green] Task complete")
console.print(f"❌ Something went wrong")
```
### Fix
```python
from aipass.cli.apps.modules import error, success, warning
error(f"Error: {msg}")
error("Failed to process")
success("Task complete")
error("Something went wrong")
```
## Scoring
- Single check per file: pass (0 violations) or fail (any violations)
- Score: 100 if passed, 0 if failed
- Threshold: score >= 75 to pass overall
- Line-level bypass filtering is supported
## Bypass
Add an entry to `.seedgo/bypass.json`:
```json
{"standard": "output_routing", "file": "path/to/file.py"}
```
Or bypass specific lines:
```json
{"standard": "output_routing", "file": "file.py", "lines": [42, 78]}
```
## Audit Scope
`AUDIT_SCOPE = all_files` — runs against every `.py` file in the branch. Skips `__init__.py` and test files.
## Reference
- Checker: `output_routing_check.py`
- Standards pack: seedgo standards (output_routing)
- Related: GitHub #661 (error paths exit 0)
@@ -0,0 +1,222 @@
# =================== AIPass ====================
# Name: output_routing_check.py
# Description: Output Routing Standards Checker Handler
# Version: 1.0.0
# Created: 2026-07-09
# Modified: 2026-07-09
# =============================================
"""
Output Routing Standards Checker Handler
Detects user-facing error/success/warning output that bypasses @cli's
semantic helpers (error(), success(), warning()) by using raw
console.print() with status markup or status emojis.
"""
import re
import sys
from pathlib import Path
from typing import Dict
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed
if sys.stdout and hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
if sys.stderr and hasattr(sys.stderr, "reconfigure"):
sys.stderr.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
AUDIT_SCOPE = "all_files"
_TEST_FILE_RE = re.compile(r"^(test_.+|.+_test|conftest)\.py$")
# console.print( or err_console.print( at any indentation
_CONSOLE_PRINT_RE = re.compile(r"(?:console|err_console)\.print\(")
# Status color markup — error indicators
_RED_MARKUP_RE = re.compile(r"\[(?:bold\s+)?red(?:\s+bold)?\]")
# Status color markup — warning indicators
_YELLOW_STATUS_RE = re.compile(r"\[yellow\].*(?:⚠|[Ww]arning|WARN|FAIL)")
# Status emojis: ❌ ✅ ✓ ✗ ✘ ⚠ ✔
_STATUS_EMOJI_RE = re.compile(r"[❌✅✓✗✘⚠✔]")
# Green check pattern: [green] followed by check emoji
_GREEN_CHECK_RE = re.compile(r"\[green\].*[✓✔✅]")
def _is_status_console_print(code: str) -> bool:
if not _CONSOLE_PRINT_RE.search(code):
return False
if _RED_MARKUP_RE.search(code):
return True
if _STATUS_EMOJI_RE.search(code):
return True
if _YELLOW_STATUS_RE.search(code):
return True
if _GREEN_CHECK_RE.search(code):
return True
return False
def _scan_file(file_path: Path) -> tuple[list[int], str | None]:
try:
source = file_path.read_text(encoding="utf-8", errors="ignore")
except OSError as exc:
logger.info("Cannot read %s: %s", file_path, exc)
return [], f"cannot read: {exc}"
lines = source.splitlines()
hit_lines: list[int] = []
in_docstring = False
docstring_char: str | None = None
for lineno, line in enumerate(lines, start=1):
stripped = line.strip()
for tq in ('"""', "'''"):
count = line.count(tq)
if count == 0:
continue
if not in_docstring:
in_docstring = True
docstring_char = tq
if count >= 2:
in_docstring = False
docstring_char = None
elif docstring_char == tq:
in_docstring = False
docstring_char = None
if in_docstring:
continue
if stripped.startswith("#"):
continue
code_part = line.split("#")[0]
if _is_status_console_print(code_part):
hit_lines.append(lineno)
return hit_lines, None
def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"""Check a Python file for user-facing output bypassing @cli helpers."""
path = Path(module_path)
if is_bypassed(module_path, "output_routing", bypass_rules=bypass_rules):
return {
"passed": True,
"checks": [
{
"name": "Bypassed",
"passed": True,
"message": "Standard bypassed via .seedgo/bypass.json",
}
],
"score": 100,
"standard": "OUTPUT_ROUTING",
}
if path.name == "__init__.py":
return {
"passed": True,
"checks": [
{
"name": "Output routing",
"passed": True,
"message": "__init__.py skipped",
}
],
"score": 100,
"standard": "OUTPUT_ROUTING",
}
if _TEST_FILE_RE.match(path.name):
return {
"passed": True,
"checks": [
{
"name": "Output routing",
"passed": True,
"message": "Test file skipped",
}
],
"score": 100,
"standard": "OUTPUT_ROUTING",
}
if not path.exists():
return {
"passed": False,
"checks": [
{
"name": "File exists",
"passed": False,
"message": f"File not found: {module_path}",
}
],
"score": 0,
"standard": "OUTPUT_ROUTING",
}
hit_lines, error = _scan_file(path)
if error is not None:
return {
"passed": False,
"checks": [
{
"name": "File readable",
"passed": False,
"message": f"Error reading file: {error}",
}
],
"score": 0,
"standard": "OUTPUT_ROUTING",
}
non_bypassed = [ln for ln in hit_lines if not is_bypassed(module_path, "output_routing", ln, bypass_rules)]
checks: list[Dict] = []
if not non_bypassed:
checks.append(
{
"name": "Output routing",
"passed": True,
"message": "All user-facing status output uses @cli helpers",
}
)
else:
sample = ", ".join(str(ln) for ln in non_bypassed[:5])
suffix = f" (and {len(non_bypassed) - 5} more)" if len(non_bypassed) > 5 else ""
checks.append(
{
"name": "Output routing",
"passed": False,
"message": f"{len(non_bypassed)} raw status output(s) on lines {sample}{suffix}",
}
)
passed_checks = sum(1 for c in checks if c["passed"])
total_checks = len(checks)
score = int(passed_checks / total_checks * 100) if total_checks > 0 else 0
overall_passed = score >= 75
json_handler.log_operation(
"check_completed",
{"file": str(module_path), "score": score, "standard": "output_routing"},
)
return {
"passed": overall_passed,
"checks": checks,
"score": score,
"standard": "OUTPUT_ROUTING",
}
@@ -0,0 +1,106 @@
# =================== AIPass ====================
# Name: output_routing_content.py
# Description: Output Routing Standards Content Handler
# Version: 1.0.0
# Created: 2026-07-09
# Modified: 2026-07-09
# =============================================
"""
Output Routing Standards Content Handler
Provides formatted Output Routing standards content.
Module orchestrates, handler implements.
"""
import sys
from aipass.seedgo.apps.handlers.json import json_handler
if sys.stdout and hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
if sys.stderr and hasattr(sys.stderr, "reconfigure"):
sys.stderr.reconfigure(encoding="utf-8") # type: ignore[attr-defined]
def get_output_routing_standards() -> str:
"""Return formatted output_routing standards content with Rich markup
Returns:
str: Formatted standards text with Rich styling
"""
lines = [
"[bold cyan]CORE PRINCIPLE:[/bold cyan]",
" User-facing error, success, and warning output MUST route through",
" @cli's semantic helpers — [dim]error()[/dim], [dim]success()[/dim],",
" [dim]warning()[/dim] — not raw [red]console.print()[/red] with",
" status markup or emojis.",
"",
"[bold cyan]WHY IT MATTERS:[/bold cyan]",
" 1. [yellow]Consistent formatting[/yellow] across all agents",
" 2. [yellow]Exit-code correctness[/yellow] — error() carries the #661",
" failure-flag fix; raw markup bypasses it (errors exit 0)",
" 3. [yellow]Stderr routing[/yellow] — error()/warning() write to stderr;",
" raw console.print() writes to stdout",
"",
"[bold cyan]WHAT IT CHECKS:[/bold cyan]",
" Scans every .py file for [dim]console.print()[/dim] or",
" [dim]err_console.print()[/dim] calls containing status indicators:",
"",
" [red]Flagged patterns:[/red]",
' - [dim]console.print(f"[red]Error: ...[/red]")[/dim] → use error()',
' - [dim]console.print("[bold red]Failed[/bold red]")[/dim] → use error()',
' - [dim]console.print("[green]...[/green]")[/dim] with check emojis → use success()',
' - [dim]console.print("...")[/dim] with status emojis → use helpers',
"",
" [green]NOT flagged (legitimate Rich usage):[/green]",
" - [dim]console.print(table)[/dim] — Rich Table objects",
" - [dim]console.print(Panel(...))[/dim] — decorative panels",
' - [dim]console.print(f"[cyan]Info...[/cyan]")[/dim] — non-status color',
' - [dim]console.print(f"[dim]...[/dim]")[/dim] — decorative formatting',
" - Lines inside docstrings, comments, test files",
"",
"[bold cyan]VIOLATIONS:[/bold cyan]",
"",
" [red]Bad — raw status output:[/red]",
' [dim]console.print(f"[red]Error: {msg}[/red]")[/dim]',
' [dim]console.print("[green]...[/green] Done")[/dim]',
' [dim]console.print("... Failed")[/dim]',
"",
" [green]Good — use @cli helpers:[/green]",
" [dim]from aipass.cli.apps.modules import error, success, warning[/dim]",
' [dim]error(f"Error: {msg}")[/dim]',
' [dim]success("Done")[/dim]',
' [dim]warning("Check configuration")[/dim]',
"",
"[bold cyan]HOW TO FIX:[/bold cyan]",
" 1. Import the helpers: [dim]from aipass.cli.apps.modules import error, success, warning[/dim]",
' 2. Replace [dim]console.print(f"[red]...")[/dim] with [dim]error(msg)[/dim]',
" 3. Replace status-emoji prints with [dim]success(msg)[/dim] or [dim]warning(msg)[/dim]",
" 4. Keep [dim]console.print()[/dim] for tables, panels, and non-status output",
"",
"[yellow]SCOPE:[/yellow]",
" AUDIT_SCOPE = [bold]all_files[/bold]",
" Runs against every .py file in the branch.",
" Skips __init__.py and test files (test_*.py, *_test.py, conftest.py).",
"",
"[bold cyan]SCORING:[/bold cyan]",
" Single check per file: [green]pass[/green] (0 violations) or [red]fail[/red]",
" Score: 100 if passed, 0 if failed",
" Threshold: score >= 75 to pass overall",
" Line-level bypass filtering is supported.",
"",
"[bold cyan]BYPASS:[/bold cyan]",
" Add an entry to [dim].seedgo/bypass.json[/dim]:",
' [dim]{"standard": "output_routing", "file": "path/to/file.py"}[/dim]',
" Or bypass specific lines:",
' [dim]{"standard": "output_routing", "file": "file.py", "lines": [42]}[/dim]',
"",
"[bold cyan]REFERENCE:[/bold cyan]",
" [dim]See: seedgo standards pack (output_routing)[/dim]",
" [dim]Checker: output_routing_check.py[/dim]",
" [dim]Related: GitHub #661 (error paths exit 0)[/dim]",
]
json_handler.log_operation("standard_content_queried", {"standard": "output_routing"})
return "\n".join(lines)
@@ -0,0 +1,424 @@
"""Tests for output_routing_check.py."""
# =================== META ====================
# Name: test_output_routing.py
# Description: Unit tests for output_routing_check
# Version: 1.0.0
# Created: 2026-07-09
# Modified: 2026-07-09
# =============================================
import pytest
from unittest.mock import MagicMock
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_infrastructure(monkeypatch):
"""Mock heavy infrastructure imports for standards checkers."""
import sys
mock_logger = MagicMock()
mock_json_handler = MagicMock()
mock_json_handler.log_operation = MagicMock(return_value=True)
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed as real_is_bypassed
bypass_pkg = MagicMock()
bypass_utils = MagicMock()
bypass_utils.is_bypassed = real_is_bypassed
bypass_pkg.utils = bypass_utils
bypass_ignore = MagicMock()
bypass_ignore.get_template_ignore_patterns = MagicMock(return_value=[])
bypass_pkg.ignore_handler = bypass_ignore
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass", bypass_pkg)
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass.utils", bypass_utils)
monkeypatch.setitem(
sys.modules,
"aipass.seedgo.apps.handlers.bypass.ignore_handler",
bypass_ignore,
)
for mod_name in [
"aipass.seedgo.apps.handlers.aipass_standards.output_routing_check",
]:
monkeypatch.delitem(sys.modules, mod_name, raising=False)
# ===========================================================================
# 1. _is_status_console_print — detection logic
# ===========================================================================
class TestIsStatusConsolePrint:
"""Tests for the _is_status_console_print helper."""
def test_red_markup_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print(f"[red]Error: {msg}[/red]")')
def test_bold_red_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("[bold red]Failed[/bold red]")')
def test_red_bold_order_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("[red bold]Error[/red bold]")')
def test_status_emoji_cross_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("❌ Something failed")')
def test_status_emoji_check_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("✅ Done")')
def test_status_emoji_checkmark_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("✓ Complete")')
def test_status_emoji_cross_mark_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("✗ Failed")')
def test_green_check_pattern_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("[green]✓[/green] Done")')
def test_yellow_warning_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('console.print("[yellow]⚠ Warning: check config[/yellow]")')
def test_err_console_detected(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert _is_status_console_print('err_console.print(f"[red]Error[/red]")')
def test_plain_console_print_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('console.print("Hello world")')
def test_cyan_markup_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('console.print(f"[cyan]Info: {msg}[/cyan]")')
def test_dim_markup_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('console.print(f"[dim]{details}[/dim]")')
def test_table_object_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print("console.print(table)")
def test_panel_object_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print("console.print(Panel(title))")
def test_no_console_print_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('logger.info("[red]error[/red]")')
def test_green_without_check_emoji_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('console.print("[green]name[/green]")')
def test_yellow_without_warning_not_flagged(self):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import (
_is_status_console_print,
)
assert not _is_status_console_print('console.print("[yellow]note[/yellow]")')
# ===========================================================================
# 2. _scan_file — file scanning
# ===========================================================================
class TestScanFile:
"""Tests for the _scan_file helper."""
def test_detects_red_markup(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text('console.print(f"[red]Error: {e}[/red]")\n', encoding="utf-8")
lines, err = _scan_file(f)
assert err is None
assert lines == [1]
def test_skips_docstrings(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text(
'"""\nconsole.print("[red]error[/red]")\n"""\npass\n',
encoding="utf-8",
)
lines, err = _scan_file(f)
assert err is None
assert lines == []
def test_skips_comments(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text('# console.print("[red]error[/red]")\n', encoding="utf-8")
lines, err = _scan_file(f)
assert err is None
assert lines == []
def test_skips_inline_comment(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text('x = 1 # console.print("[red]error[/red]")\n', encoding="utf-8")
lines, err = _scan_file(f)
assert err is None
assert lines == []
def test_detects_multiple_lines(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text(
'x = 1\nconsole.print("[red]a[/red]")\ny = 2\nconsole.print("✅ done")\n',
encoding="utf-8",
)
lines, err = _scan_file(f)
assert err is None
assert lines == [2, 4]
def test_clean_file(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "test.py"
f.write_text('console.print("[cyan]info[/cyan]")\nprint("hello")\n', encoding="utf-8")
lines, err = _scan_file(f)
assert err is None
assert lines == []
def test_unreadable_file(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import _scan_file
f = tmp_path / "missing.py"
lines, err = _scan_file(f)
assert err is not None
assert lines == []
# ===========================================================================
# 3. check_module — full checker
# ===========================================================================
class TestCheckModule:
"""Tests for check_module."""
def test_clean_file_passes(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "clean.py"
f.write_text('from aipass.cli.apps.modules import error\nerror("fail")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
assert result["score"] == 100
assert result["standard"] == "OUTPUT_ROUTING"
def test_violation_detected(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "bad.py"
f.write_text('console.print(f"[red]Error: {e}[/red]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is False
assert result["score"] == 0
def test_init_py_skipped(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "__init__.py"
f.write_text('console.print("[red]error[/red]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
assert result["score"] == 100
def test_test_file_skipped(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "test_something.py"
f.write_text('console.print("[red]error[/red]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
assert result["score"] == 100
def test_conftest_skipped(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "conftest.py"
f.write_text('console.print("[red]error[/red]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
assert result["score"] == 100
def test_bypass_returns_100(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "bypassed.py"
f.write_text('console.print("[red]error[/red]")\n', encoding="utf-8")
bypass = [{"standard": "output_routing", "file": str(f)}]
result = check_module(str(f), bypass_rules=bypass)
assert result["passed"] is True
assert result["score"] == 100
def test_missing_file(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
result = check_module(str(tmp_path / "no_such.py"))
assert result["passed"] is False
assert result["score"] == 0
def test_line_bypass_all_lines_pass(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "partial.py"
f.write_text(
'console.print("[red]a[/red]")\nconsole.print("[red]b[/red]")\n',
encoding="utf-8",
)
bypass = [{"standard": "output_routing", "file": "partial.py", "lines": [1, 2]}]
result = check_module(str(f), bypass_rules=bypass)
assert result["passed"] is True
def test_violation_message_shows_lines(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "multi.py"
f.write_text('console.print("[red]a[/red]")\nconsole.print("✅ b")\n', encoding="utf-8")
result = check_module(str(f))
assert "2 raw" in result["checks"][0]["message"]
assert "1, 2" in result["checks"][0]["message"]
def test_more_than_five_violations_truncated(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "many.py"
lines = [f'console.print("[red]err{i}[/red]")\n' for i in range(8)]
f.write_text("".join(lines), encoding="utf-8")
result = check_module(str(f))
assert "and 3 more" in result["checks"][0]["message"]
# ===========================================================================
# 4. False-positive avoidance
# ===========================================================================
class TestFalsePositiveAvoidance:
"""Verify that legitimate patterns are NOT flagged."""
def test_table_print_not_flagged(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "tables.py"
f.write_text("console.print(table)\nconsole.print(Panel(content))\n", encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
def test_blue_markup_not_flagged(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "blue.py"
f.write_text('console.print("[blue]Processing...[/blue]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
def test_empty_console_print_not_flagged(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "blank.py"
f.write_text('console.print("")\nconsole.print()\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
def test_docstring_with_markup_not_flagged(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "docs.py"
content = '"""\nconsole.print("[red]error[/red]")\n"""\ndef foo(): pass\n'
f.write_text(content, encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True
def test_green_text_without_emoji_not_flagged(self, tmp_path):
from aipass.seedgo.apps.handlers.aipass_standards.output_routing_check import check_module
f = tmp_path / "green.py"
f.write_text('console.print("[green]branch_name[/green]")\n', encoding="utf-8")
result = check_module(str(f))
assert result["passed"] is True