starmen1265
6f18550906
- Test OutputCapture._is_prompt for prompt pattern detection - Test OutputCapture.capture_until_complete for output capture - Test CommandExecutor._detect_errors for error keyword detection - Test CommandExecutor.execute for full command execution flow - Test CommandResult dataclass for result storage - Test CommandExecutor.send_command for pane interaction Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""Tests for command execution and output capture.
|
||
|
||
This module tests the CommandExecutor and OutputCapture classes which provide
|
||
the foundational I/O layer for all GSD command bridging.
|
||
"""
|
||
import pytest
|
||
import time
|
||
from unittest.mock import MagicMock, patch, PropertyMock
|
||
from dataclasses import fields
|
||
|
||
from gsd_openclaw.command_executor import (
|
||
CommandResult,
|
||
OutputCapture,
|
||
CommandExecutor,
|
||
)
|
||
|
||
|
||
class TestOutputCaptureIsPrompt:
|
||
"""Tests for OutputCapture._is_prompt method."""
|
||
|
||
def test_is_prompt_detects_dollar_prompt(self):
|
||
"""_is_prompt should return True for '$ ' prompt."""
|
||
capture = OutputCapture()
|
||
assert capture._is_prompt("$ ") is True
|
||
|
||
def test_is_prompt_detects_dollar_with_space_only(self):
|
||
"""_is_prompt should return True for '$' followed by only whitespace."""
|
||
capture = OutputCapture()
|
||
assert capture._is_prompt("$ ") is True
|
||
|
||
def test_is_prompt_detects_unicode_arrow_prompt(self):
|
||
"""_is_prompt should return True for Unicode arrow prompts like '❯ '."""
|
||
capture = OutputCapture()
|
||
assert capture._is_prompt("❯ ") is True
|
||
|
||
def test_is_prompt_detects_triangle_arrow_prompt(self):
|
||
"""_is_prompt should return True for triangle arrow prompts like '➜ '."""
|
||
capture = OutputCapture()
|
||
assert capture._is_prompt("➜ ") is True
|
||
|
||
def test_is_prompt_returns_false_for_non_prompt_lines(self):
|
||
"""_is_prompt should return False for non-prompt lines."""
|
||
capture = OutputCapture()
|
||
assert capture._is_prompt("some command output") is False
|
||
assert capture._is_prompt("$ not a prompt (has content after)") is False
|
||
assert capture._is_prompt("") is False
|
||
assert capture._is_prompt("error: something failed") is False
|
||
|
||
|
||
class TestOutputCaptureCaptureUntilComplete:
|
||
"""Tests for OutputCapture.capture_until_complete method."""
|
||
|
||
@pytest.fixture
|
||
def mock_pane(self):
|
||
"""Create a mock tmux pane."""
|
||
pane = MagicMock()
|
||
pane.pane_id = "%100"
|
||
return pane
|
||
|
||
def test_capture_until_complete_returns_output_and_true_when_prompt_detected(self, mock_pane):
|
||
"""capture_until_complete should return (output, True) when prompt is detected."""
|
||
capture = OutputCapture()
|
||
|
||
# Simulate output that ends with prompt
|
||
mock_pane.capture_pane.return_value = [
|
||
"Running command...",
|
||
"Output line 1",
|
||
"Output line 2",
|
||
"$ ", # Prompt appears
|
||
]
|
||
|
||
output, completed = capture.capture_until_complete(mock_pane, timeout=1.0)
|
||
|
||
assert completed is True
|
||
assert "Running command..." in output
|
||
assert "Output line 1" in output
|
||
|
||
def test_capture_until_complete_returns_true_when_output_stabilizes(self, mock_pane):
|
||
"""capture_until_complete should return (output, True) when output stabilizes."""
|
||
capture = OutputCapture()
|
||
|
||
# Simulate stable output (same for 3 consecutive polls)
|
||
stable_output = ["Line 1", "Line 2", "Line 3"]
|
||
mock_pane.capture_pane.return_value = stable_output
|
||
|
||
output, completed = capture.capture_until_complete(
|
||
mock_pane, timeout=1.0, poll_interval=0.01
|
||
)
|
||
|
||
assert completed is True
|
||
assert "Line 1" in output
|
||
assert "Line 2" in output
|
||
|
||
def test_capture_until_complete_returns_false_on_timeout(self, mock_pane):
|
||
"""capture_until_complete should return (output, False) on timeout."""
|
||
capture = OutputCapture()
|
||
|
||
# Simulate output that keeps changing (never stabilizes, no prompt)
|
||
call_count = [0]
|
||
|
||
def changing_output():
|
||
call_count[0] += 1
|
||
return [f"Line {call_count[0]}"]
|
||
|
||
mock_pane.capture_pane.side_effect = changing_output
|
||
|
||
output, completed = capture.capture_until_complete(
|
||
mock_pane, timeout=0.1, poll_interval=0.02
|
||
)
|
||
|
||
assert completed is False
|
||
|
||
|
||
class TestCommandExecutorDetectErrors:
|
||
"""Tests for CommandExecutor._detect_errors method."""
|
||
|
||
def test_detect_errors_returns_true_for_error_keyword(self):
|
||
"""_detect_errors should return (True, 'error') for 'Error: something'."""
|
||
executor = CommandExecutor()
|
||
has_error, error_type = executor._detect_errors("Error: something went wrong")
|
||
|
||
assert has_error is True
|
||
assert error_type == "error"
|
||
|
||
def test_detect_errors_returns_true_for_failed_keyword(self):
|
||
"""_detect_errors should return (True, 'failed') for 'Process failed'."""
|
||
executor = CommandExecutor()
|
||
has_error, error_type = executor._detect_errors("Process failed to start")
|
||
|
||
assert has_error is True
|
||
assert error_type == "failed"
|
||
|
||
def test_detect_errors_returns_true_for_exception_keyword(self):
|
||
"""_detect_errors should return (True, 'exception') for exception in output."""
|
||
executor = CommandExecutor()
|
||
has_error, error_type = executor._detect_errors("Traceback: ValueError occurred")
|
||
|
||
assert has_error is True
|
||
assert error_type == "exception"
|
||
|
||
def test_detect_errors_returns_true_for_traceback_keyword(self):
|
||
"""_detect_errors should return (True, 'traceback') for traceback in output."""
|
||
executor = CommandExecutor()
|
||
has_error, error_type = executor._detect_errors("Traceback (most recent call last):")
|
||
|
||
assert has_error is True
|
||
assert error_type == "traceback"
|
||
|
||
def test_detect_errors_returns_true_for_chinese_error_keywords(self):
|
||
"""_detect_errors should detect Chinese error keywords."""
|
||
executor = CommandExecutor()
|
||
|
||
# Test Chinese error
|
||
has_error, error_type = executor._detect_errors("发生错误：无法连接")
|
||
assert has_error is True
|
||
assert error_type == "错误"
|
||
|
||
# Test Chinese failure
|
||
has_error, error_type = executor._detect_errors("操作失败")
|
||
assert has_error is True
|
||
assert error_type == "失败"
|
||
|
||
def test_detect_errors_returns_false_for_normal_output(self):
|
||
"""_detect_errors should return (False, None) for normal output."""
|
||
executor = CommandExecutor()
|
||
|
||
normal_outputs = [
|
||
"Command completed successfully",
|
||
"All tests passed",
|
||
"Files created: 5",
|
||
"Done!",
|
||
]
|
||
|
||
for output in normal_outputs:
|
||
has_error, error_type = executor._detect_errors(output)
|
||
assert has_error is False
|
||
assert error_type is None
|
||
|
||
|
||
class TestCommandExecutorExecute:
|
||
"""Tests for CommandExecutor.execute method."""
|
||
|
||
@pytest.fixture
|
||
def mock_session(self):
|
||
"""Create a mock TmuxSession with active_pane."""
|
||
session = MagicMock()
|
||
session.session_id = "test-123"
|
||
session.project_name = "test-project"
|
||
session.tmux_session_name = "gsd-test-project"
|
||
|
||
# Mock the tmux_session.active_pane
|
||
mock_pane = MagicMock()
|
||
mock_pane.pane_id = "%100"
|
||
mock_pane.send_keys = MagicMock()
|
||
|
||
mock_tmux_session = MagicMock()
|
||
mock_tmux_session.active_pane = mock_pane
|
||
session.tmux_session = mock_tmux_session
|
||
|
||
return session
|
||
|
||
def test_execute_returns_command_result_with_all_fields(self, mock_session):
|
||
"""execute should return CommandResult with all required fields."""
|
||
executor = CommandExecutor()
|
||
|
||
# Mock capture to return immediately with prompt
|
||
mock_session.tmux_session.active_pane.capture_pane.return_value = [
|
||
"Command output",
|
||
"$ ",
|
||
]
|
||
|
||
result = executor.execute(mock_session, "echo hello", timeout=1.0)
|
||
|
||
# Verify CommandResult has all required fields
|
||
assert hasattr(result, 'output')
|
||
assert hasattr(result, 'completed')
|
||
assert hasattr(result, 'has_error')
|
||
assert hasattr(result, 'error_type')
|
||
|
||
def test_execute_calls_send_keys_with_enter(self, mock_session):
|
||
"""execute should call send_keys with the command."""
|
||
executor = CommandExecutor()
|
||
|
||
# Mock capture to return immediately
|
||
mock_session.tmux_session.active_pane.capture_pane.return_value = ["$ "]
|
||
|
||
executor.execute(mock_session, "echo hello", timeout=1.0)
|
||
|
||
# Verify send_keys was called
|
||
mock_session.tmux_session.active_pane.send_keys.assert_called_once()
|
||
|
||
def test_execute_sets_has_error_true_when_error_keywords_detected(self, mock_session):
|
||
"""execute should set has_error=True when error keywords detected."""
|
||
executor = CommandExecutor()
|
||
|
||
# Mock output with error
|
||
mock_session.tmux_session.active_pane.capture_pane.return_value = [
|
||
"Error: command failed",
|
||
"$ ",
|
||
]
|
||
|
||
result = executor.execute(mock_session, "bad-command", timeout=1.0)
|
||
|
||
assert result.has_error is True
|
||
assert result.error_type == "error"
|
||
|
||
def test_execute_respects_timeout_parameter(self, mock_session):
|
||
"""execute should respect the timeout parameter."""
|
||
executor = CommandExecutor()
|
||
|
||
# Mock output that never stabilizes
|
||
call_count = [0]
|
||
|
||
def changing_output():
|
||
call_count[0] += 1
|
||
return [f"Line {call_count[0]}"]
|
||
|
||
mock_session.tmux_session.active_pane.capture_pane.side_effect = changing_output
|
||
|
||
start_time = time.time()
|
||
result = executor.execute(mock_session, "slow-command", timeout=0.1, poll_interval=0.02)
|
||
elapsed = time.time() - start_time
|
||
|
||
assert result.completed is False
|
||
assert elapsed < 0.3 # Should timeout around 0.1s, not much longer
|
||
|
||
def test_execute_uses_default_timeout_of_60_seconds(self, mock_session):
|
||
"""execute should use default timeout of 60 seconds."""
|
||
executor = CommandExecutor()
|
||
|
||
# Verify DEFAULT_TIMEOUT is set
|
||
assert executor.DEFAULT_TIMEOUT == 60.0
|
||
|
||
|
||
class TestCommandResultDataclass:
|
||
"""Tests for CommandResult dataclass."""
|
||
|
||
def test_command_result_stores_all_fields(self):
|
||
"""CommandResult should store all required fields."""
|
||
result = CommandResult(
|
||
output="test output",
|
||
completed=True,
|
||
has_error=False,
|
||
error_type=None,
|
||
)
|
||
|
||
assert result.output == "test output"
|
||
assert result.completed is True
|
||
assert result.has_error is False
|
||
assert result.error_type is None
|
||
|
||
def test_command_result_with_error(self):
|
||
"""CommandResult should store error information correctly."""
|
||
result = CommandResult(
|
||
output="Error: something failed",
|
||
completed=True,
|
||
has_error=True,
|
||
error_type="error",
|
||
)
|
||
|
||
assert result.has_error is True
|
||
assert result.error_type == "error"
|
||
|
||
def test_command_result_incomplete_execution(self):
|
||
"""CommandResult should handle incomplete execution."""
|
||
result = CommandResult(
|
||
output="Partial output...",
|
||
completed=False,
|
||
has_error=False,
|
||
error_type=None,
|
||
)
|
||
|
||
assert result.completed is False
|
||
|
||
|
||
class TestCommandExecutorSendCommand:
|
||
"""Tests for CommandExecutor.send_command method."""
|
||
|
||
def test_send_command_calls_pane_send_keys(self):
|
||
"""send_command should call pane.send_keys with the command."""
|
||
executor = CommandExecutor()
|
||
|
||
mock_pane = MagicMock()
|
||
mock_pane.send_keys = MagicMock()
|
||
|
||
executor.send_command(mock_pane, "echo hello")
|
||
|
||
mock_pane.send_keys.assert_called_once()
|
||
|
||
def test_send_command_passes_enter_true(self):
|
||
"""send_command should send keys with enter=True to execute the command."""
|
||
executor = CommandExecutor()
|
||
|
||
mock_pane = MagicMock()
|
||
mock_pane.send_keys = MagicMock()
|
||
|
||
executor.send_command(mock_pane, "echo hello")
|
||
|
||
# Verify it was called - the exact implementation should include enter
|
||
call_args = mock_pane.send_keys.call_args
|
||
assert call_args is not None
|