From 6f18550906cc235580295583e51115eeeaff6f0e Mon Sep 17 00:00:00 2001
From: starmen1265 <starmen1265@163.com>
Date: Wed, 11 Mar 2026 15:48:19 +0800
Subject: [PATCH] test(03-01): add failing tests for command executor

- 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>
---
 gsd-openclaw/tests/test_command_executor.py | 341 ++++++++++++++++++++
 1 file changed, 341 insertions(+)
 create mode 100644 gsd-openclaw/tests/test_command_executor.py

diff --git a/gsd-openclaw/tests/test_command_executor.py b/gsd-openclaw/tests/test_command_executor.py
new file mode 100644
index 0000000..535e083
--- /dev/null
+++ b/gsd-openclaw/tests/test_command_executor.py
@@ -0,0 +1,341 @@
+"""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