feat(task-logger): strip ANSI escape codes from logs and extend coverage (#1411)
* auto-claude: subtask-1-1 - Add strip_ansi_codes() utility function to task_lo * auto-claude: subtask-1-2 - Apply ANSI sanitization to TaskLogger.log_with_detail() and TaskLogger.tool_end() Changes: - Import strip_ansi_codes from utils in logger.py - Apply strip_ansi_codes to detail parameter in log_with_detail() - Apply strip_ansi_codes to stored_detail in tool_end() - Fix circular import in utils.py using TYPE_CHECKING Co-Authored-By: Claude <noreply@anthropic.com> * auto-claude: subtask-1-3 - Apply ANSI sanitization to StreamingLogCapture.pro - Import strip_ansi_codes from utils module - Apply sanitization to process_text() method before logging - Ensures ANSI escape codes are removed from streaming text output Co-Authored-By: Claude <noreply@anthropic.com> * feat(task-logger): strip ANSI escape codes from logs and extend coverage Backend (Python): - Extend CSI pattern to support private mode parameters (?<>=) - Add None handling to strip_ansi_codes() function - Add sanitization to TaskLogger.log() method (critical gap fix) - Export strip_ansi_codes in task_logger public API - Add 25 comprehensive unit tests for strip_ansi_codes() Frontend (TypeScript): - Extend CSI pattern to support private mode parameters (?<>=) - Apply ANSI sanitization to merge preview error messages in Kanban This ensures clean display of task logs and error messages in the UI by removing terminal color/formatting escape sequences. * fix(task_logger): resolve cyclic import issue Move strip_ansi_codes import from module-level to local imports in logger.py to avoid cyclic import when __init__.py imports both modules. Also use lazy import in get_task_logger() to avoid cyclic import at module level. Fixes NameError in test_planner_session_does_not_trigger_post_session_processing_on_retry * fix(tests): add missing os import in test_task_logger.py Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * fix(task_logger): sanitize content parameter in log_with_detail() - Apply ANSI stripping to both content and detail in log_with_detail() - Remove unused Path import from test file - Add test for content sanitization in log_with_detail() Fixes issue identified by CodeRabbit review where log_with_detail() only sanitized detail but not content, allowing ANSI codes to leak into stored logs and UI components. Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * fix(task_logger): sanitize result and content in tool_end() - Apply ANSI stripping to display_result and content in tool_end() - Add test for result and content sanitization in tool_end() Fixes CodeRabbit review comment where tool_end() only sanitized detail but not result/content, allowing ANSI codes to leak into the stored log content and UI components. Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * refactor: consolidate duplicate inline imports and remove redundant assert - Consolidate 3 inline imports of strip_ansi_codes in tool_end() to single import - Consolidate 2 inline imports in log_with_detail() to single import - Remove redundant 'assert True' in test_public_api_exports Addressed CodeRabbit review comments about code quality. Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * fix: scope strip_ansi_codes usage within import block Move content and detail sanitization inside the import block to resolve CodeQL false positive about potentially uninitialized local variable. The import and all usages are now within the same conditional scope. Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * fix: address CodeRabbit review comments - Add lazy import of TaskLogger in update_task_logger_path() to avoid potential NameError when accessing TaskLogger.LOG_FILE - Sanitize text before checking for empty in capture.process_text() to avoid logging blank entries when input contains only ANSI codes These were false positives in practice but improve code clarity and static analysis results. Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> * fix: sanitize message, tool_input, and subphase parameters in logging methods Apply strip_ansi_codes() to: - start_phase() message parameter - end_phase() message parameter - tool_start() tool_input parameter - start_subphase() subphase parameter This ensures consistency with other logging methods (log, log_with_detail, tool_end) which all sanitize input before storage. Addresses Auto Claude PR Review findings: - 🟡 [31465b234445] start_phase() message not sanitized - 🟡 [18192bb9d19d] end_phase() message not sanitized - 🟡 [16aa996a0d8d] tool_start() tool_input not sanitized - 🟡 [b39df9833b80] start_subphase() subphase not sanitized * refactor: extract ANSI utilities to separate module to fix cyclic import Create new task_logger/ansi.py module containing strip_ansi_codes() and related ANSI patterns, removing the dependency on logger.py. This resolves CodeQL cyclic import warnings: - utils.py imports TaskLogger (TYPE_CHECKING only) - logger.py imports strip_ansi_codes from utils - New ansi.py has no dependencies on other task_logger modules Changes: - Create apps/backend/task_logger/ansi.py with strip_ansi_codes() - Update __init__.py to import strip_ansi_codes directly from .ansi - Update logger.py to import from .ansi instead of .utils (7 locations) - Update capture.py to import from .ansi instead of .utils - Update tests to import from task_logger.ansi directly - Remove duplicate ANSI code patterns from utils.py All 27 tests pass. * fix: address CodeRabbit review comments 1. Expand ANSI_CSI_PATTERN to match full ANSI/VT100 CSI final-byte range - Old pattern: \x1b\[[?><=0-9;]*[A-Za-z] - New pattern: \x1b\[[0-?]*[ -/]*[@-~] - Now strips bracketed paste sequences (\x1b[200~, \x1b[201~) and other CSI sequences with non-letter final bytes 2. Print sanitized phase_message instead of raw message in start_phase/end_phase - Prevents ANSI codes from appearing in console output - Keeps console output consistent with stored log content 3. Add test for bracketed paste sequences (test_csi_bracketed_paste) All 28 tests pass. * fix: address CodeRabbit review comments 1. Use platform abstraction in subprocess-spawn test - Replace process.platform === 'win32' with isWindows() from platform module - Follows coding guidelines for cross-platform checks 2. Add trailing newline to test_task_logger.py for POSIX compliance All tests pass (28 Python tests, 14 TypeScript tests). * fix: resolve all Auto Claude PR Review findings Backend: - Move strip_ansi_codes to module-level import in logger.py - Removes 7 duplicate inline imports, keeping only 1 at top of file Frontend: - Update ANSI_CSI_PATTERN to match backend regex: /\x1b\[[0-?]*[ -/]*[@-~]/g - Change final byte pattern from [A-Za-z] to [@-~] for full CSI coverage - Add [ -/]* for intermediate bytes handling - Add tests for bracketed paste sequences and private mode parameters This resolves 4 remaining findings: - 🔵 [LOW] Duplicate inline imports - FIXED - 🟡 [MEDIUM] Frontend regex missing CSI final bytes - FIXED - 🔵 [LOW] Frontend regex missing intermediate bytes - FIXED - 🔵 [LOW] Frontend missing bracketed paste tests - FIXED All tests pass: 28 Python + 36 TypeScript = 64 total. * fix: sanitize before truncation to avoid partial ANSI remnants Truncating before sanitizing can cut ANSI escape sequences in half, leaving stray control characters that the regex won't remove. Changes: - Sanitize display_result before truncation (300 char limit) - Sanitize stored_detail before truncation (10KB limit) - Use original detail length for truncation message Fixes CodeRabbit review comment about tool_end() sanitization order. All 28 tests pass. * refactor: address CodeRabbit nitpick comments Backend: - Remove redundant outer guard "if content or detail:" in log_with_detail() - Remove redundant sanitization of content in tool_end() (already sanitized) - Reduces nesting and removes unnecessary conditional checks Frontend: - Standardize all subprocess test timeouts to 30000ms for Windows CI - Provides consistent margin for slower Windows CI environment All tests pass: 28 Python + 14 TypeScript = 42 total. * refactor: remove redundant conditions in start_phase and end_phase Since phase_message always has a fallback default value, it can never be falsy. The if phase_message: guards are unnecessary. Simplifies code by: - Removing redundant condition before sanitization - Removing redundant condition before print All 28 tests pass. * fix: address Auto Claude review LOW severity findings 1. Fix truncation message to report sanitized length - Use sanitized_len for truncation message instead of unsanitized detail length - Ensures reported length matches visible content 2. Fix misleading comment in utils.py - Update comment to reflect that ANSI functions are in ansi.py module - Clarifies architectural decision rather than implying an import All 28 tests pass. * fix: apply ANSI sanitization to all subprocess error paths Apply stripAnsiCodes() to merge and create PR error output for consistency with the preview handler. * refactor: remove unused variable original_len (dead code) Remove the unused original_len variable that was leftover from the previous ANSI sanitization fix. --------- Signed-off-by: StillKnotKnown <stillknotknown@users.noreply.github.com> Co-authored-by: StillKnotKnown <stillknotknown@users.noreply.github.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Andy <119136210+AndyMik90@users.noreply.github.com>
This commit is contained in:
@@ -14,6 +14,8 @@ Key features:
|
||||
|
||||
# Export models
|
||||
# Export streaming capture
|
||||
# Export utility functions
|
||||
from .ansi import strip_ansi_codes
|
||||
from .capture import StreamingLogCapture
|
||||
|
||||
# Export main logger
|
||||
@@ -22,9 +24,11 @@ from .models import LogEntry, LogEntryType, LogPhase, PhaseLog
|
||||
|
||||
# Export storage utilities
|
||||
from .storage import get_active_phase, load_task_logs
|
||||
|
||||
# Export utility functions
|
||||
from .utils import clear_task_logger, get_task_logger, update_task_logger_path
|
||||
from .utils import (
|
||||
clear_task_logger,
|
||||
get_task_logger,
|
||||
update_task_logger_path,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Models
|
||||
@@ -41,6 +45,7 @@ __all__ = [
|
||||
"get_task_logger",
|
||||
"clear_task_logger",
|
||||
"update_task_logger_path",
|
||||
"strip_ansi_codes",
|
||||
# Streaming capture
|
||||
"StreamingLogCapture",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
"""
|
||||
ANSI escape code utilities for task logging.
|
||||
|
||||
This module contains functions for stripping ANSI escape codes from strings.
|
||||
It has no dependencies on other task_logger modules to avoid cyclic imports.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
# ANSI escape code patterns
|
||||
# ANSI CSI (Control Sequence Introducer) escape sequence pattern.
|
||||
# Matches the full ANSI/VT100 CSI form: ESC [ parameter bytes (0-?) intermediate bytes ( -/) final bytes (@-~)
|
||||
# Parameter bytes: 0x30-0x3F (digits 0-9, :;<=>?)
|
||||
# Intermediate bytes: 0x20-0x2F (space and !"#$%&'()*+,-./)
|
||||
# Final bytes: 0x40-0x7E (@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\]^_`abcdefghijklmnopqrstuvwxyz{|}~)
|
||||
# Examples: \x1b[31m (red), \x1b[?25l (hide cursor), \x1b[200~ (bracketed paste start)
|
||||
ANSI_CSI_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
|
||||
|
||||
# OSC (Operating System Command) escape sequences with BEL (bell) terminator
|
||||
# Matches: \x1b] ... \x07
|
||||
ANSI_OSC_BEL_PATTERN = re.compile(r"\x1b\][^\x07]*\x07")
|
||||
|
||||
# OSC (Operating System Command) escape sequences with ST (string terminator)
|
||||
# Matches: \x1b] ... \x1b\
|
||||
ANSI_OSC_ST_PATTERN = re.compile(r"\x1b\][^\x1b]*\x1b\\")
|
||||
|
||||
|
||||
def strip_ansi_codes(text: str | None) -> str:
|
||||
"""
|
||||
Removes ANSI escape codes from a string.
|
||||
|
||||
These sequences are used for terminal coloring/formatting but appear
|
||||
as raw text in logs and UI components.
|
||||
|
||||
Args:
|
||||
text: The string potentially containing ANSI escape codes, or None
|
||||
|
||||
Returns:
|
||||
The string with all ANSI escape sequences removed, or empty string if input is None
|
||||
|
||||
Example:
|
||||
>>> strip_ansi_codes('\\x1b[90m[21:40:22.196]\\x1b[0m \\x1b[36m[DEBUG]\\x1b[0m')
|
||||
'[21:40:22.196] [DEBUG]'
|
||||
"""
|
||||
if not text:
|
||||
return ""
|
||||
|
||||
# Remove all ANSI escape sequences
|
||||
result = ANSI_CSI_PATTERN.sub("", text)
|
||||
result = ANSI_OSC_BEL_PATTERN.sub("", result)
|
||||
result = ANSI_OSC_ST_PATTERN.sub("", result)
|
||||
|
||||
return result
|
||||
@@ -2,6 +2,7 @@
|
||||
Streaming log capture for agent sessions.
|
||||
"""
|
||||
|
||||
from .ansi import strip_ansi_codes
|
||||
from .logger import TaskLogger
|
||||
from .models import LogPhase
|
||||
|
||||
@@ -36,8 +37,10 @@ class StreamingLogCapture:
|
||||
|
||||
def process_text(self, text: str) -> None:
|
||||
"""Process text output from the agent."""
|
||||
if text.strip():
|
||||
self.logger.log(text, phase=self.phase)
|
||||
# Remove ANSI escape codes before logging
|
||||
sanitized_text = strip_ansi_codes(text)
|
||||
if sanitized_text.strip():
|
||||
self.logger.log(sanitized_text, phase=self.phase)
|
||||
|
||||
def process_tool_start(self, tool_name: str, tool_input: str | None = None) -> None:
|
||||
"""Process tool start."""
|
||||
|
||||
@@ -7,6 +7,7 @@ from pathlib import Path
|
||||
|
||||
from core.debug import debug, debug_error, debug_info, debug_success, is_debug_enabled
|
||||
|
||||
from .ansi import strip_ansi_codes
|
||||
from .models import LogEntry, LogEntryType, LogPhase
|
||||
from .storage import LogStorage
|
||||
from .streaming import emit_marker
|
||||
@@ -160,6 +161,7 @@ class TaskLogger:
|
||||
|
||||
# Add phase start entry
|
||||
phase_message = message or f"Starting {phase_key} phase"
|
||||
phase_message = strip_ansi_codes(phase_message)
|
||||
entry = LogEntry(
|
||||
timestamp=self._timestamp(),
|
||||
type=LogEntryType.PHASE_START.value,
|
||||
@@ -172,9 +174,8 @@ class TaskLogger:
|
||||
# Debug log (when DEBUG=true)
|
||||
self._debug_log(phase_message, LogEntryType.PHASE_START, phase_key)
|
||||
|
||||
# Also print the message
|
||||
if message:
|
||||
print(message, flush=True)
|
||||
# Also print the message (sanitized)
|
||||
print(phase_message, flush=True)
|
||||
|
||||
def end_phase(
|
||||
self, phase: LogPhase, success: bool = True, message: str | None = None
|
||||
@@ -203,6 +204,8 @@ class TaskLogger:
|
||||
phase_message = (
|
||||
message or f"{'Completed' if success else 'Failed'} {phase_key} phase"
|
||||
)
|
||||
phase_message = strip_ansi_codes(phase_message)
|
||||
|
||||
entry = LogEntry(
|
||||
timestamp=self._timestamp(),
|
||||
type=LogEntryType.PHASE_END.value,
|
||||
@@ -216,8 +219,8 @@ class TaskLogger:
|
||||
entry_type = LogEntryType.SUCCESS if success else LogEntryType.ERROR
|
||||
self._debug_log(phase_message, entry_type, phase_key)
|
||||
|
||||
if message:
|
||||
print(message, flush=True)
|
||||
# Print the message (sanitized)
|
||||
print(phase_message, flush=True)
|
||||
|
||||
if phase == self.current_phase:
|
||||
self.current_phase = None
|
||||
@@ -240,6 +243,10 @@ class TaskLogger:
|
||||
phase: Optional phase override (uses current_phase if not specified)
|
||||
print_to_console: Whether to also print to stdout (default True)
|
||||
"""
|
||||
# Sanitize content to remove ANSI escape codes before storage
|
||||
if content:
|
||||
content = strip_ansi_codes(content)
|
||||
|
||||
phase_key = (phase or self.current_phase or LogPhase.CODING).value
|
||||
|
||||
entry = LogEntry(
|
||||
@@ -307,6 +314,13 @@ class TaskLogger:
|
||||
"""
|
||||
phase_key = (phase or self.current_phase or LogPhase.CODING).value
|
||||
|
||||
# Sanitize content and detail before storage
|
||||
if content:
|
||||
content = strip_ansi_codes(content)
|
||||
|
||||
if detail:
|
||||
detail = strip_ansi_codes(detail)
|
||||
|
||||
entry = LogEntry(
|
||||
timestamp=self._timestamp(),
|
||||
type=entry_type.value,
|
||||
@@ -363,6 +377,10 @@ class TaskLogger:
|
||||
"""
|
||||
phase_key = (phase or self.current_phase or LogPhase.CODING).value
|
||||
|
||||
# Sanitize subphase before use
|
||||
if subphase:
|
||||
subphase = strip_ansi_codes(subphase)
|
||||
|
||||
entry = LogEntry(
|
||||
timestamp=self._timestamp(),
|
||||
type=LogEntryType.INFO.value,
|
||||
@@ -406,6 +424,10 @@ class TaskLogger:
|
||||
"""
|
||||
phase_key = (phase or self.current_phase or LogPhase.CODING).value
|
||||
|
||||
# Sanitize tool_input before use
|
||||
if tool_input:
|
||||
tool_input = strip_ansi_codes(tool_input)
|
||||
|
||||
# Truncate long inputs for display (increased limit to avoid hiding critical info)
|
||||
display_input = tool_input
|
||||
if display_input and len(display_input) > 300:
|
||||
@@ -462,8 +484,8 @@ class TaskLogger:
|
||||
"""
|
||||
phase_key = (phase or self.current_phase or LogPhase.CODING).value
|
||||
|
||||
# Truncate long results for display (increased limit to avoid hiding critical info)
|
||||
display_result = result
|
||||
# Sanitize before truncation to avoid cutting ANSI sequences mid-stream
|
||||
display_result = strip_ansi_codes(result) if result else None
|
||||
if display_result and len(display_result) > 300:
|
||||
display_result = display_result[:297] + "..."
|
||||
|
||||
@@ -472,12 +494,13 @@ class TaskLogger:
|
||||
if display_result:
|
||||
content += f": {display_result}"
|
||||
|
||||
# Truncate detail for storage (max 10KB to avoid bloating JSON)
|
||||
stored_detail = detail
|
||||
# Sanitize before truncating detail
|
||||
stored_detail = strip_ansi_codes(detail) if detail else None
|
||||
if stored_detail and len(stored_detail) > 10240:
|
||||
sanitized_len = len(stored_detail)
|
||||
stored_detail = (
|
||||
stored_detail[:10240]
|
||||
+ f"\n\n... [truncated - full output was {len(detail)} chars]"
|
||||
+ f"\n\n... [truncated - full output was {sanitized_len} chars]"
|
||||
)
|
||||
|
||||
entry = LogEntry(
|
||||
|
||||
@@ -3,16 +3,21 @@ Utility functions for task logging.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
# ANSI functions are in separate ansi.py module to avoid cyclic imports
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .logger import TaskLogger
|
||||
|
||||
from .logger import TaskLogger
|
||||
|
||||
# Global logger instance for easy access
|
||||
_current_logger: TaskLogger | None = None
|
||||
_current_logger: "TaskLogger | None" = None
|
||||
|
||||
|
||||
def get_task_logger(
|
||||
spec_dir: Path | None = None, emit_markers: bool = True
|
||||
) -> TaskLogger | None:
|
||||
) -> "TaskLogger | None":
|
||||
"""
|
||||
Get or create a task logger for the given spec directory.
|
||||
|
||||
@@ -29,6 +34,9 @@ def get_task_logger(
|
||||
return _current_logger
|
||||
|
||||
if _current_logger is None or _current_logger.spec_dir != spec_dir:
|
||||
# Lazy import to avoid cyclic import
|
||||
from .logger import TaskLogger
|
||||
|
||||
_current_logger = TaskLogger(spec_dir, emit_markers)
|
||||
|
||||
return _current_logger
|
||||
@@ -55,6 +63,9 @@ def update_task_logger_path(new_spec_dir: Path) -> None:
|
||||
if _current_logger is None:
|
||||
return
|
||||
|
||||
# Lazy import to avoid cyclic import
|
||||
from .logger import TaskLogger
|
||||
|
||||
# Update the logger's internal paths
|
||||
_current_logger.spec_dir = Path(new_spec_dir)
|
||||
_current_logger.log_file = _current_logger.spec_dir / TaskLogger.LOG_FILE
|
||||
|
||||
@@ -12,6 +12,7 @@ import { mkdirSync, rmSync, existsSync, writeFileSync, mkdtempSync } from 'fs';
|
||||
import { tmpdir } from 'os';
|
||||
import path from 'path';
|
||||
import { findPythonCommand, parsePythonCommand } from '../../main/python-detector';
|
||||
import { isWindows } from '../../main/platform';
|
||||
|
||||
// Test directories - use secure temp directory with random suffix
|
||||
let TEST_DIR: string;
|
||||
@@ -361,7 +362,7 @@ describe('Subprocess Spawn Integration', () => {
|
||||
|
||||
expect(result).toBe(true);
|
||||
// On Windows, kill() is called without arguments; on Unix, kill('SIGTERM') is used
|
||||
if (process.platform === 'win32') {
|
||||
if (isWindows()) {
|
||||
expect(mockProcess.kill).toHaveBeenCalled();
|
||||
} else {
|
||||
expect(mockProcess.kill).toHaveBeenCalledWith('SIGTERM');
|
||||
|
||||
@@ -21,6 +21,7 @@ import {
|
||||
import { persistPlanStatus, updateTaskMetadataPrUrl } from './plan-file-utils';
|
||||
import { getIsolatedGitEnv, detectWorktreeBranch, refreshGitIndex } from '../../utils/git-isolation';
|
||||
import { killProcessGracefully } from '../../platform';
|
||||
import { stripAnsiCodes } from '../../../shared/utils/ansi-sanitizer';
|
||||
|
||||
// Regex pattern for validating git branch names
|
||||
const GIT_BRANCH_REGEX = /^[a-zA-Z0-9][a-zA-Z0-9._/-]*[a-zA-Z0-9]$|^[a-zA-Z0-9]$/;
|
||||
@@ -2328,7 +2329,9 @@ export function registerWorktreeHandlers(
|
||||
success: true,
|
||||
data: {
|
||||
success: false,
|
||||
message: hasConflicts ? 'Merge conflicts detected' : `Merge failed: ${stderr || stdout}`,
|
||||
message: hasConflicts
|
||||
? 'Merge conflicts detected'
|
||||
: `Merge failed: ${stripAnsiCodes(stderr || stdout)}`,
|
||||
conflictFiles: hasConflicts ? [] : undefined
|
||||
}
|
||||
});
|
||||
@@ -2528,7 +2531,7 @@ export function registerWorktreeHandlers(
|
||||
console.error('[IPC] stderr:', stderr);
|
||||
resolve({
|
||||
success: false,
|
||||
error: `Failed to parse preview result: ${stderr || stdout}`
|
||||
error: `Failed to parse preview result: ${stripAnsiCodes(stderr || stdout)}`
|
||||
});
|
||||
}
|
||||
} else {
|
||||
@@ -2537,7 +2540,7 @@ export function registerWorktreeHandlers(
|
||||
console.error('[IPC] stdout:', stdout);
|
||||
resolve({
|
||||
success: false,
|
||||
error: `Preview failed: ${stderr || stdout}`
|
||||
error: `Preview failed: ${stripAnsiCodes(stderr || stdout)}`
|
||||
});
|
||||
}
|
||||
});
|
||||
@@ -3129,7 +3132,7 @@ export function registerWorktreeHandlers(
|
||||
// Prefer stdout over stderr since stderr often contains debug messages
|
||||
resolve({
|
||||
success: false,
|
||||
error: stdout || stderr || 'Failed to create PR'
|
||||
error: stripAnsiCodes(stdout || stderr || 'Failed to create PR')
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,6 +44,20 @@ describe('stripAnsiCodes', () => {
|
||||
const expected = 'Text';
|
||||
expect(stripAnsiCodes(input)).toBe(expected);
|
||||
});
|
||||
|
||||
it('should handle CSI bracketed paste sequences', () => {
|
||||
// Bracketed paste start/end with non-alphabetic final byte (~)
|
||||
const input = '\x1b[200~pasted text\x1b[201~';
|
||||
const expected = 'pasted text';
|
||||
expect(stripAnsiCodes(input)).toBe(expected);
|
||||
});
|
||||
|
||||
it('should handle CSI with private mode parameters', () => {
|
||||
// Private mode sequences use ?<>=
|
||||
const input = '\x1b[?25lhide cursor\x1b[?25hshow cursor';
|
||||
const expected = 'hide cursorshow cursor';
|
||||
expect(stripAnsiCodes(input)).toBe(expected);
|
||||
});
|
||||
});
|
||||
|
||||
describe('OSC (Operating System Command) patterns', () => {
|
||||
|
||||
@@ -12,14 +12,15 @@
|
||||
|
||||
/**
|
||||
* ANSI CSI (Control Sequence Introducer) escape sequence pattern.
|
||||
* Matches:
|
||||
* - \x1b[ or \e[ or \033[ - Escape sequence start
|
||||
* - [0-9;]* - Parameter bytes (numbers and semicolons)
|
||||
* - [A-Za-z] - Final byte (command)
|
||||
* - m - Specifically for SGR (Select Graphic Rendition) color codes
|
||||
* Matches the full ANSI/VT100 CSI form: ESC [ parameter-bytes intermediate-bytes final-bytes
|
||||
* - Parameter bytes: 0x30-0x3F (digits 0-9, :;<=>?) -> [0-?]* in regex
|
||||
* - Intermediate bytes: 0x20-0x2F (space and !"#$%&'()*+,-./) -> [ -/]* in regex
|
||||
* - Final bytes: 0x40-0x7E (@ through ~) -> [@-~] in regex
|
||||
*
|
||||
* Examples: \x1b[31m (red), \x1b[?25l (hide cursor), \x1b[200~ (bracketed paste start)
|
||||
*/
|
||||
// biome-ignore lint/suspicious/noControlCharactersInRegex: ANSI escape codes require control characters
|
||||
const ANSI_CSI_PATTERN = /\x1b\[[0-9;]*[A-Za-z]/g;
|
||||
const ANSI_CSI_PATTERN = /\x1b\[[0-?]*[ -/]*[@-~]/g;
|
||||
|
||||
/**
|
||||
* OSC (Operating System Command) escape sequences.
|
||||
|
||||
Generated
+32
-14
@@ -247,6 +247,7 @@
|
||||
"integrity": "sha512-H3mcG6ZDLTlYfaSNi0iOKkigqMFvkTKlGUYlD8GW7nNOYRrevuA46iTypPyv+06V3fEmvvazfntkBU34L0azAw==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@babel/code-frame": "^7.28.6",
|
||||
"@babel/generator": "^7.28.6",
|
||||
@@ -811,6 +812,7 @@
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=18"
|
||||
},
|
||||
@@ -854,6 +856,7 @@
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=18"
|
||||
}
|
||||
@@ -893,6 +896,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@dnd-kit/core/-/core-6.3.1.tgz",
|
||||
"integrity": "sha512-xkGBRQQab4RLwgXxoqETICr6S5JlogafbhNsidmrkVv2YRs5MLwpjoF2qpiGjQt8S9AoxtIV603s0GIUpY5eYQ==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@dnd-kit/accessibility": "^3.1.1",
|
||||
"@dnd-kit/utilities": "^3.2.2",
|
||||
@@ -1311,7 +1315,6 @@
|
||||
"dev": true,
|
||||
"license": "BSD-2-Clause",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"cross-dirname": "^0.1.0",
|
||||
"debug": "^4.3.4",
|
||||
@@ -1333,7 +1336,6 @@
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"graceful-fs": "^4.2.0",
|
||||
"jsonfile": "^6.0.1",
|
||||
@@ -1350,7 +1352,6 @@
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"universalify": "^2.0.0"
|
||||
},
|
||||
@@ -1365,7 +1366,6 @@
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">= 10.0.0"
|
||||
}
|
||||
@@ -2245,6 +2245,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.0.tgz",
|
||||
"integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=8.0.0"
|
||||
}
|
||||
@@ -2266,6 +2267,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/context-async-hooks/-/context-async-hooks-2.4.0.tgz",
|
||||
"integrity": "sha512-jn0phJ+hU7ZuvaoZE/8/Euw3gvHJrn2yi+kXrymwObEPVPjtwCmkvXDRQCWli+fCTTF/aSOtXaLr7CLIvv3LQg==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": "^18.19.0 || >=20.6.0"
|
||||
},
|
||||
@@ -2278,6 +2280,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-2.4.0.tgz",
|
||||
"integrity": "sha512-KtcyFHssTn5ZgDu6SXmUznS80OFs/wN7y6MyFRRcKU6TOw8hNcGxKvt8hsdaLJfhzUszNSjURetq5Qpkad14Gw==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@opentelemetry/semantic-conventions": "^1.29.0"
|
||||
},
|
||||
@@ -2293,6 +2296,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/instrumentation/-/instrumentation-0.208.0.tgz",
|
||||
"integrity": "sha512-Eju0L4qWcQS+oXxi6pgh7zvE2byogAkcsVv0OjHF/97iOz1N/aKE6etSGowYkie+YA1uo6DNwdSxaaNnLvcRlA==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@opentelemetry/api-logs": "0.208.0",
|
||||
"import-in-the-middle": "^2.0.0",
|
||||
@@ -2695,6 +2699,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-2.4.0.tgz",
|
||||
"integrity": "sha512-RWvGLj2lMDZd7M/5tjkI/2VHMpXebLgPKvBUd9LRasEWR2xAynDwEYZuLvY9P2NGG73HF07jbbgWX2C9oavcQg==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@opentelemetry/core": "2.4.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.29.0"
|
||||
@@ -2711,6 +2716,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.4.0.tgz",
|
||||
"integrity": "sha512-WH0xXkz/OHORDLKqaxcUZS0X+t1s7gGlumr2ebiEgNZQl2b0upK2cdoD0tatf7l8iP74woGJ/Kmxe82jdvcWRw==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@opentelemetry/core": "2.4.0",
|
||||
"@opentelemetry/resources": "2.4.0",
|
||||
@@ -2728,6 +2734,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.39.0.tgz",
|
||||
"integrity": "sha512-R5R9tb2AXs2IRLNKLBJDynhkfmx7mX0vi8NkhZb3gUkPWHn6HXk5J8iQ/dql0U3ApfWym4kXXmBDRGO+oeOfjg==",
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=14"
|
||||
}
|
||||
@@ -4922,6 +4929,7 @@
|
||||
"integrity": "sha512-o4PXJQidqJl82ckFaXUeoAW+XysPLauYI43Abki5hABd853iMhitooc6znOnczgbTYmEP6U6/y1ZyKAIsvMKGg==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@babel/code-frame": "^7.10.4",
|
||||
"@babel/runtime": "^7.12.5",
|
||||
@@ -5220,6 +5228,7 @@
|
||||
"resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.8.tgz",
|
||||
"integrity": "sha512-3MbSL37jEchWZz2p2mjntRZtPt837ij10ApxKfgmXCTuHWagYg7iA5bqPw6C8BMPfwidlvfPI/fxOc42HLhcyg==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"csstype": "^3.2.2"
|
||||
}
|
||||
@@ -5230,6 +5239,7 @@
|
||||
"integrity": "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==",
|
||||
"devOptional": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"peerDependencies": {
|
||||
"@types/react": "^19.2.0"
|
||||
}
|
||||
@@ -5495,6 +5505,7 @@
|
||||
"resolved": "https://registry.npmjs.org/acorn/-/acorn-8.15.0.tgz",
|
||||
"integrity": "sha512-NZyJarBfL7nWwIq+FDL6Zp/yHEhePMNnnJ0y3qfieCrmNvYct8uvtiV41UvlSe6apAfk0fY1FbWx+NwfmpvtTg==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"bin": {
|
||||
"acorn": "bin/acorn"
|
||||
},
|
||||
@@ -5527,6 +5538,7 @@
|
||||
"integrity": "sha512-j3fVLgvTo527anyYyJOGTYJbG+vnnQYvE0m5mmkc1TK+nxAppkCLMIL0aZ4dblVCNoGShhm+kzE4ZUykBoMg4g==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"fast-deep-equal": "^3.1.1",
|
||||
"fast-json-stable-stringify": "^2.0.0",
|
||||
@@ -5968,6 +5980,7 @@
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"baseline-browser-mapping": "^2.9.0",
|
||||
"caniuse-lite": "^1.0.30001759",
|
||||
@@ -6666,8 +6679,7 @@
|
||||
"integrity": "sha512-+R08/oI0nl3vfPcqftZRpytksBXDzOUveBq/NBVx0sUp1axwzPQrKinNx5yd5sxPu8j1wIy8AfnVQ+5eFdha6Q==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true
|
||||
"optional": true
|
||||
},
|
||||
"node_modules/cross-env": {
|
||||
"version": "10.1.0",
|
||||
@@ -6984,6 +6996,7 @@
|
||||
"integrity": "sha512-ce4Ogns4VMeisIuCSK0C62umG0lFy012jd8LMZ6w/veHUeX4fqfDrGe+HTWALAEwK6JwKP+dhPvizhArSOsFbg==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"app-builder-lib": "26.4.0",
|
||||
"builder-util": "26.3.4",
|
||||
@@ -7140,6 +7153,7 @@
|
||||
"dev": true,
|
||||
"hasInstallScript": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@electron/get": "^2.0.0",
|
||||
"@types/node": "^22.7.7",
|
||||
@@ -7388,7 +7402,6 @@
|
||||
"dev": true,
|
||||
"hasInstallScript": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@electron/asar": "^3.2.1",
|
||||
"debug": "^4.1.1",
|
||||
@@ -7409,7 +7422,6 @@
|
||||
"integrity": "sha512-YJDaCJZEnBmcbw13fvdAM9AwNOJwOzrE4pqMqBq5nFiEqXUqHwlK4B+3pUw6JNvfSPtX05xFHtYy/1ni01eGCw==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"graceful-fs": "^4.1.2",
|
||||
"jsonfile": "^4.0.0",
|
||||
@@ -8419,6 +8431,7 @@
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@babel/runtime": "^7.28.4"
|
||||
},
|
||||
@@ -8745,6 +8758,7 @@
|
||||
"integrity": "sha512-mjzqwWRD9Y1J1KUi7W97Gja1bwOOM5Ug0EZ6UDK3xS7j7mndrkwozHtSblfomlzyB4NepioNt+B2sOSzczVgtQ==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"@acemir/cssom": "^0.9.28",
|
||||
"@asamuzakjp/dom-selector": "^6.7.6",
|
||||
@@ -11260,6 +11274,7 @@
|
||||
}
|
||||
],
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"nanoid": "^3.3.11",
|
||||
"picocolors": "^1.1.1",
|
||||
@@ -11335,7 +11350,6 @@
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"commander": "^9.4.0"
|
||||
},
|
||||
@@ -11353,7 +11367,6 @@
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": "^12.20.0 || >=14"
|
||||
}
|
||||
@@ -11480,6 +11493,7 @@
|
||||
"resolved": "https://registry.npmjs.org/react/-/react-19.2.3.tgz",
|
||||
"integrity": "sha512-Ku/hhYbVjOQnXDZFv2+RibmLFGwFdeeKHFcOTlrt7xplBnya5OGn/hIRDsqDiSUcfORsDC7MPxwork8jBwsIWA==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=0.10.0"
|
||||
}
|
||||
@@ -11489,6 +11503,7 @@
|
||||
"resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.2.3.tgz",
|
||||
"integrity": "sha512-yELu4WmLPw5Mr/lmeEpox5rw3RETacE++JgHqQzd2dg+YbJuat3jH4ingc+WPZhxaoFzdv9y33G+F7Nl5O0GBg==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"scheduler": "^0.27.0"
|
||||
},
|
||||
@@ -11891,7 +11906,6 @@
|
||||
"deprecated": "Rimraf versions prior to v4 are no longer supported",
|
||||
"dev": true,
|
||||
"license": "ISC",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"glob": "^7.1.3"
|
||||
},
|
||||
@@ -12470,7 +12484,8 @@
|
||||
"version": "4.1.18",
|
||||
"resolved": "https://registry.npmjs.org/tailwindcss/-/tailwindcss-4.1.18.tgz",
|
||||
"integrity": "sha512-4+Z+0yiYyEtUVCScyfHCxOYP06L5Ne+JiHhY2IjR2KWMIWhJOYZKLSGZaP5HkZ8+bY0cxfzwDE5uOmzFXyIwxw==",
|
||||
"license": "MIT"
|
||||
"license": "MIT",
|
||||
"peer": true
|
||||
},
|
||||
"node_modules/tapable": {
|
||||
"version": "2.3.0",
|
||||
@@ -12573,7 +12588,6 @@
|
||||
"integrity": "sha512-yYrrsWnrXMcdsnu/7YMYAofM1ktpL5By7vZhf15CrXijWWrEYZks5AXBudalfSWJLlnen/QUJUB5aoB0kqZUGA==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"mkdirp": "^0.5.1",
|
||||
"rimraf": "~2.6.2"
|
||||
@@ -12637,7 +12651,6 @@
|
||||
"integrity": "sha512-FP+p8RB8OWpF3YZBCrP5gtADmtXApB5AMLn+vdyA+PyxCjrCs00mjyUozssO33cwDeT3wNGdLxJ5M//YqtHAJw==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"minimist": "^1.2.6"
|
||||
},
|
||||
@@ -12729,6 +12742,7 @@
|
||||
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=12"
|
||||
},
|
||||
@@ -12887,6 +12901,7 @@
|
||||
"integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==",
|
||||
"devOptional": true,
|
||||
"license": "Apache-2.0",
|
||||
"peer": true,
|
||||
"bin": {
|
||||
"tsc": "bin/tsc",
|
||||
"tsserver": "bin/tsserver"
|
||||
@@ -13193,6 +13208,7 @@
|
||||
"integrity": "sha512-w+N7Hifpc3gRjZ63vYBXA56dvvRlNWRczTdmCBBa+CotUzAPf5b7YMdMR/8CQoeYE5LX3W4wj6RYTgonm1b9DA==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"dependencies": {
|
||||
"esbuild": "^0.27.0",
|
||||
"fdir": "^6.5.0",
|
||||
@@ -13785,6 +13801,7 @@
|
||||
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
|
||||
"dev": true,
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"engines": {
|
||||
"node": ">=12"
|
||||
},
|
||||
@@ -14226,6 +14243,7 @@
|
||||
"resolved": "https://registry.npmjs.org/zod/-/zod-4.3.5.tgz",
|
||||
"integrity": "sha512-k7Nwx6vuWx1IJ9Bjuf4Zt1PEllcwe7cls3VNzm4CQ1/hgtFUK2bRNG3rvnpPUhFjmqJKAKtjV576KnUkHocg/g==",
|
||||
"license": "MIT",
|
||||
"peer": true,
|
||||
"funding": {
|
||||
"url": "https://github.com/sponsors/colinhacks"
|
||||
}
|
||||
|
||||
@@ -0,0 +1,338 @@
|
||||
"""
|
||||
Task Logger Tests
|
||||
|
||||
Tests for the task_logger module including ANSI code stripping functionality.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
# Add backend to path for imports
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'apps', 'backend'))
|
||||
|
||||
from task_logger.ansi import strip_ansi_codes
|
||||
from task_logger.capture import StreamingLogCapture
|
||||
from task_logger.logger import TaskLogger
|
||||
from task_logger.models import LogEntryType, LogPhase
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Unit Tests for strip_ansi_codes() Function
|
||||
# ============================================================================
|
||||
|
||||
class TestStripAnsiCodes:
|
||||
"""Unit tests for the strip_ansi_codes() utility function."""
|
||||
|
||||
def test_empty_string(self):
|
||||
"""Empty string should return empty string."""
|
||||
assert strip_ansi_codes("") == ""
|
||||
|
||||
def test_none_input(self):
|
||||
"""None input should return empty string."""
|
||||
assert strip_ansi_codes(None) == ""
|
||||
|
||||
def test_no_ansi_codes(self):
|
||||
"""Plain text without ANSI codes should be unchanged."""
|
||||
assert strip_ansi_codes("plain text") == "plain text"
|
||||
assert strip_ansi_codes("Hello, World!") == "Hello, World!"
|
||||
assert strip_ansi_codes("12345") == "12345"
|
||||
|
||||
def test_simple_color_code(self):
|
||||
"""Simple CSI color codes should be removed."""
|
||||
assert strip_ansi_codes("\x1b[31mred\x1b[0m") == "red"
|
||||
assert strip_ansi_codes("\x1b[32mgreen\x1b[0m") == "green"
|
||||
assert strip_ansi_codes("\x1b[34mblue\x1b[0m") == "blue"
|
||||
|
||||
def test_vitest_like_output(self):
|
||||
"""Vitest-like timestamp and debug output should be cleaned."""
|
||||
input_text = "\x1b[90m[21:40:22.196]\x1b[0m \x1b[36m[DEBUG]\x1b[0m Test message"
|
||||
expected = "[21:40:22.196] [DEBUG] Test message"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_multiple_ansi_codes(self):
|
||||
"""Multiple consecutive ANSI codes should all be removed."""
|
||||
input_text = "\x1b[31m\x1b[1mbold red\x1b[0m"
|
||||
expected = "bold red"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_osc_bel_sequence(self):
|
||||
"""OSC sequences with BEL terminator should be removed."""
|
||||
assert strip_ansi_codes("\x1b]0;Window Title\x07") == ""
|
||||
assert strip_ansi_codes("Text\x1b]0;Title\x07More") == "TextMore"
|
||||
|
||||
def test_osc_st_sequence(self):
|
||||
"""OSC sequences with ST terminator should be removed."""
|
||||
assert strip_ansi_codes("\x1b]0;Window Title\x1b\\") == ""
|
||||
assert strip_ansi_codes("Text\x1b]0;Title\x1b\\More") == "TextMore"
|
||||
|
||||
def test_mixed_ansi_types(self):
|
||||
"""Mixed CSI and OSC sequences in same string should all be removed."""
|
||||
input_text = "\x1b[31mError:\x1b[0m \x1b]1;Title\x07Failed"
|
||||
expected = "Error: Failed"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_multiline_text(self):
|
||||
"""Multi-line text with ANSI codes should be cleaned."""
|
||||
input_text = "\x1b[31mLine 1\x1b[0m\nLine 2\x1b[32m\x1b[1m\x1b[0m\nLine 3"
|
||||
expected = "Line 1\nLine 2\nLine 3"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_private_mode_parameters(self):
|
||||
"""CSI sequences with private mode parameters should be removed."""
|
||||
# Cursor hide/show
|
||||
assert strip_ansi_codes("\x1b[?25lHide\x1b[?25hShow") == "HideShow"
|
||||
# Private mode with other chars
|
||||
assert strip_ansi_codes("\x1b[=1hApplication Mode\x1b[=0l") == "Application Mode"
|
||||
|
||||
def test_csi_with_parameters(self):
|
||||
"""CSI sequences with semicolon-separated parameters should be removed."""
|
||||
# Bold red (1;31)
|
||||
assert strip_ansi_codes("\x1b[1;31mText\x1b[0m") == "Text"
|
||||
# Multiple parameters
|
||||
assert strip_ansi_codes("\x1b[38;2;255;0;0mRGB Red\x1b[0m") == "RGB Red"
|
||||
|
||||
def test_csi_cursor_movement(self):
|
||||
"""CSI cursor movement sequences should be removed."""
|
||||
assert strip_ansi_codes("Text\x1b[2K") == "Text"
|
||||
assert strip_ansi_codes("\x1b[0G\x1b[2KClear line") == "Clear line"
|
||||
assert strip_ansi_codes("\x1b[A\x1b[B\x1b[C\x1b[D") == ""
|
||||
|
||||
def test_ansi_hyperlinks(self):
|
||||
"""ANSI hyperlink format (OSC 8) should be removed."""
|
||||
input_text = "\x1b]8;;https://example.com\x07Click here\x1b]8;;\x07"
|
||||
expected = "Click here"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_csi_bracketed_paste(self):
|
||||
"""CSI bracketed paste sequences should be removed (final byte ~)."""
|
||||
# Bracketed paste start/end
|
||||
assert strip_ansi_codes("\x1b[200~") == ""
|
||||
assert strip_ansi_codes("\x1b[201~") == ""
|
||||
# Bracketed paste with content
|
||||
assert strip_ansi_codes("\x1b[200~text\x1b[201~") == "text"
|
||||
|
||||
def test_unicode_with_ansi(self):
|
||||
"""Unicode text combined with ANSI codes should preserve Unicode."""
|
||||
input_text = "\x1b[31m你好\x1b[0m \x1b[32m世界\x1b[0m"
|
||||
expected = "你好 世界"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
# Emoji
|
||||
input_text = "\x1b[36m🎉\x1b[0m \x1b[33m🚀\x1b[0m"
|
||||
expected = "🎉 🚀"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
def test_very_long_input(self):
|
||||
"""Very long strings with many ANSI codes should be handled efficiently."""
|
||||
# Create a long string with alternating ANSI codes and text
|
||||
parts = []
|
||||
for i in range(100):
|
||||
parts.append(f"\x1b[{i % 10}mtext{i}\x1b[0m")
|
||||
input_text = "".join(parts)
|
||||
result = strip_ansi_codes(input_text)
|
||||
|
||||
# Verify all ANSI codes are removed
|
||||
assert "\x1b" not in result
|
||||
# Verify text content is preserved
|
||||
for i in range(100):
|
||||
assert f"text{i}" in result
|
||||
|
||||
def test_only_ansi_codes(self):
|
||||
"""String consisting entirely of ANSI codes should return empty."""
|
||||
assert strip_ansi_codes("\x1b[31m\x1b[1m\x1b[4m") == ""
|
||||
assert strip_ansi_codes("\x1b]0;Title\x07") == ""
|
||||
|
||||
def test_nested_ansi_sequences(self):
|
||||
"""Nested ANSI sequences should all be removed."""
|
||||
input_text = "\x1b[31m\x1b[1mbold red\x1b[0m \x1b[32mgreen\x1b[0m"
|
||||
expected = "bold red green"
|
||||
assert strip_ansi_codes(input_text) == expected
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Integration Tests for TaskLogger
|
||||
# ============================================================================
|
||||
|
||||
class TestTaskLoggerAnsiIntegration:
|
||||
"""Integration tests for TaskLogger ANSI code sanitization."""
|
||||
|
||||
def test_log_sanitizes_content(self, tmp_path):
|
||||
"""The log() method should sanitize content before storage."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
logger.log(
|
||||
"\x1b[31mError message\x1b[0m",
|
||||
LogEntryType.ERROR,
|
||||
print_to_console=False
|
||||
)
|
||||
|
||||
# Load the log file and verify content is sanitized
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
assert len(coding_entries) == 1
|
||||
assert coding_entries[0]["content"] == "Error message"
|
||||
assert "\x1b" not in coding_entries[0]["content"]
|
||||
|
||||
def test_log_with_detail_sanitizes_detail(self, tmp_path):
|
||||
"""log_with_detail() should sanitize detail parameter."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
logger.log_with_detail(
|
||||
content="Reading file",
|
||||
detail="\x1b[31mERROR:\x1b[0m File not found",
|
||||
print_to_console=False
|
||||
)
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
assert len(coding_entries) == 1
|
||||
assert coding_entries[0]["detail"] == "ERROR: File not found"
|
||||
assert "\x1b" not in coding_entries[0]["detail"]
|
||||
|
||||
def test_log_with_detail_sanitizes_content(self, tmp_path):
|
||||
"""log_with_detail() should sanitize content parameter."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
logger.log_with_detail(
|
||||
content="\x1b[33mWarning:\x1b[0m Check this",
|
||||
detail="Some detail text",
|
||||
print_to_console=False
|
||||
)
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
assert len(coding_entries) == 1
|
||||
assert coding_entries[0]["content"] == "Warning: Check this"
|
||||
assert "\x1b" not in coding_entries[0]["content"]
|
||||
|
||||
def test_tool_end_sanitizes_detail(self, tmp_path):
|
||||
"""tool_end() should sanitize detail parameter."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
logger.tool_start("Bash", "npm test")
|
||||
logger.tool_end(
|
||||
"Bash",
|
||||
success=True,
|
||||
result="Tests completed",
|
||||
detail="\x1b[36m$ npm test\x1b[0m\n\x1b[32mPASS\x1b[0m All tests passed"
|
||||
)
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
# Find the tool_end entry
|
||||
tool_end_entries = [e for e in coding_entries if e["type"] == "tool_end"]
|
||||
assert len(tool_end_entries) == 1
|
||||
assert tool_end_entries[0]["detail"] == "$ npm test\nPASS All tests passed"
|
||||
assert "\x1b" not in tool_end_entries[0]["detail"]
|
||||
|
||||
def test_tool_end_sanitizes_result_and_content(self, tmp_path):
|
||||
"""tool_end() should sanitize result and content parameters."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
logger.tool_start("Bash", "npm test")
|
||||
logger.tool_end(
|
||||
"Bash",
|
||||
success=True,
|
||||
result="\x1b[32mTests passed\x1b[0m",
|
||||
detail="Some output"
|
||||
)
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
tool_end_entries = [e for e in coding_entries if e["type"] == "tool_end"]
|
||||
assert len(tool_end_entries) == 1
|
||||
# Content should be "[Bash] Done: Tests passed" without ANSI codes
|
||||
assert tool_end_entries[0]["content"] == "[Bash] Done: Tests passed"
|
||||
assert "\x1b" not in tool_end_entries[0]["content"]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Integration Tests for StreamingLogCapture
|
||||
# ============================================================================
|
||||
|
||||
class TestStreamingLogCaptureAnsiIntegration:
|
||||
"""Integration tests for StreamingLogCapture ANSI code sanitization."""
|
||||
|
||||
def test_process_text_sanitizes(self, tmp_path):
|
||||
"""process_text() should sanitize text before logging."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
with StreamingLogCapture(logger, LogPhase.CODING) as capture:
|
||||
capture.process_text("\x1b[90m[DEBUG]\x1b[0m Processing...")
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
assert len(coding_entries) == 1
|
||||
assert coding_entries[0]["content"] == "[DEBUG] Processing..."
|
||||
assert "\x1b" not in coding_entries[0]["content"]
|
||||
|
||||
def test_process_text_multiple_calls(self, tmp_path):
|
||||
"""Multiple process_text calls should each sanitize."""
|
||||
logger = TaskLogger(tmp_path, emit_markers=False)
|
||||
|
||||
with StreamingLogCapture(logger, LogPhase.CODING) as capture:
|
||||
capture.process_text("\x1b[31mError\x1b[0m")
|
||||
capture.process_text("\x1b[32mSuccess\x1b[0m")
|
||||
|
||||
log_file = tmp_path / "task_logs.json"
|
||||
with open(log_file) as f:
|
||||
logs = json.load(f)
|
||||
|
||||
coding_entries = logs["phases"]["coding"]["entries"]
|
||||
assert len(coding_entries) == 2
|
||||
assert coding_entries[0]["content"] == "Error"
|
||||
assert coding_entries[1]["content"] == "Success"
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Public API Tests
|
||||
# ============================================================================
|
||||
|
||||
class TestTaskLoggerPublicAPI:
|
||||
"""Tests for the task_logger public API exports."""
|
||||
|
||||
def test_strip_ansi_codes_is_exported(self):
|
||||
"""strip_ansi_codes should be importable from task_logger package."""
|
||||
from task_logger import strip_ansi_codes as exported_strip
|
||||
|
||||
# Verify it's the same function
|
||||
assert exported_strip is strip_ansi_codes
|
||||
|
||||
# Verify it works
|
||||
assert exported_strip("\x1b[31mtest\x1b[0m") == "test"
|
||||
|
||||
def test_public_api_exports(self):
|
||||
"""All expected exports should be available."""
|
||||
from task_logger import (
|
||||
LogPhase,
|
||||
LogEntryType,
|
||||
LogEntry,
|
||||
TaskLogger,
|
||||
load_task_logs,
|
||||
get_active_phase,
|
||||
get_task_logger,
|
||||
clear_task_logger,
|
||||
update_task_logger_path,
|
||||
strip_ansi_codes,
|
||||
StreamingLogCapture,
|
||||
)
|
||||
# If imports succeed, the test passes
|
||||
Reference in New Issue
Block a user