diff --git a/gsd-openclaw/tests/test_orchestrator.py b/gsd-openclaw/tests/test_orchestrator.py new file mode 100644 index 0000000..e07da5a --- /dev/null +++ b/gsd-openclaw/tests/test_orchestrator.py @@ -0,0 +1,522 @@ +"""Tests for orchestrator command routing and integration.""" +import pytest +from unittest.mock import MagicMock, patch + +from gsd_openclaw.command_executor import CommandResult +from gsd_openclaw.tmux_session import TmuxSession + + +class TestOrchestratorInit: + """Tests for Orchestrator initialization.""" + + def test_orchestrator_initializes_with_required_components(self): + """Orchestrator should initialize with all required components.""" + from gsd_openclaw.orchestrator import Orchestrator + + mock_tmux_manager = MagicMock() + mock_formatter = MagicMock() + + orchestrator = Orchestrator( + tmux_manager=mock_tmux_manager, + formatter=mock_formatter + ) + + assert orchestrator.tmux_manager is mock_tmux_manager + assert orchestrator.formatter is mock_formatter + + def test_orchestrator_creates_default_components_if_not_provided(self): + """Orchestrator should create default components if not provided.""" + from gsd_openclaw.orchestrator import Orchestrator + + orchestrator = Orchestrator() + + assert orchestrator.tmux_manager is not None + assert orchestrator.formatter is not None + + +class TestCommandParsing: + """Tests for command parsing.""" + + def test_command_parsing_extracts_gsd_pattern(self): + """Command parsing should extract /gsd:* pattern.""" + from gsd_openclaw.orchestrator import Orchestrator + + orchestrator = Orchestrator() + + # Test valid commands + result = orchestrator.parse_command("/gsd:status") + assert result == ("status", "") + + result = orchestrator.parse_command("/gsd:plan feature-x") + assert result == ("plan", "feature-x") + + result = orchestrator.parse_command("/gsd:use my-project") + assert result == ("use", "my-project") + + def test_command_parsing_returns_none_for_non_gsd(self): + """Command parsing should return None for non-gsd commands.""" + from gsd_openclaw.orchestrator import Orchestrator + + orchestrator = Orchestrator() + + result = orchestrator.parse_command("/help") + assert result is None + + result = orchestrator.parse_command("regular message") + assert result is None + + +class TestHandleMessage: + """Tests for handle_message routing.""" + + @pytest.fixture + def mock_orchestrator(self): + """Create orchestrator with mocked dependencies.""" + from gsd_openclaw.orchestrator import Orchestrator + + mock_tmux_manager = MagicMock() + mock_formatter = MagicMock() + mock_executor = MagicMock() + mock_approval_handler = MagicMock() + + orchestrator = Orchestrator( + tmux_manager=mock_tmux_manager, + formatter=mock_formatter + ) + orchestrator.executor = mock_executor + orchestrator.approval_handler = mock_approval_handler + + return orchestrator + + def test_handle_message_routes_to_session_creation_when_no_session(self, mock_orchestrator): + """handle_message should create session when no session exists.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_tmux_session.session_id = "test-123" + mock_orchestrator.tmux_manager.get_session.return_value = None + mock_orchestrator.tmux_manager.create_session.return_value = mock_tmux_session + + mock_orchestrator.executor.execute.return_value = CommandResult( + output="Command executed", + completed=True, + has_error=False, + error_type=None + ) + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify session was created + mock_orchestrator.tmux_manager.create_session.assert_called_once() + + def test_handle_message_executes_command_and_returns_success_card(self, mock_orchestrator): + """handle_message should execute command and return success card.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_tmux_session.session_id = "test-123" + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + mock_orchestrator.executor.execute.return_value = CommandResult( + output="Status: OK", + completed=True, + has_error=False, + error_type=None + ) + + mock_orchestrator.formatter.format_success.return_value = {"card": "success"} + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify command was executed + mock_orchestrator.executor.execute.assert_called_once() + # Verify success card was returned + mock_orchestrator.formatter.format_success.assert_called_once() + + def test_handle_message_returns_error_card_when_command_fails(self, mock_orchestrator): + """handle_message should return error card when command fails.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + mock_orchestrator.executor.execute.return_value = CommandResult( + output="Error: Command failed", + completed=True, + has_error=True, + error_type="error" + ) + + mock_orchestrator.formatter.format_error.return_value = {"card": "error"} + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify error card was returned + mock_orchestrator.formatter.format_error.assert_called_once() + + def test_handle_message_returns_approval_card_when_approval_detected(self, mock_orchestrator): + """handle_message should return approval card when approval is detected.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + mock_orchestrator.executor.execute.return_value = CommandResult( + output="AskUserQuestion: Should I continue?", + completed=False, + has_error=False, + error_type=None + ) + + mock_orchestrator.approval_handler.detect_approval_request.return_value = { + "type": "AskUserQuestion", + "matched_text": "AskUserQuestion: Should I continue?" + } + mock_orchestrator.approval_handler.create_approval_card.return_value = {"card": "approval"} + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:plan feature", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify approval detection was called + mock_orchestrator.approval_handler.detect_approval_request.assert_called_once() + # Verify approval card was created + mock_orchestrator.approval_handler.create_approval_card.assert_called_once() + + def test_handle_message_truncates_output_to_2000_chars(self, mock_orchestrator): + """handle_message should truncate output to 2000 chars for card limits.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + # Create long output + long_output = "x" * 3000 + mock_orchestrator.executor.execute.return_value = CommandResult( + output=long_output, + completed=True, + has_error=False, + error_type=None + ) + + mock_orchestrator.formatter.format_success.return_value = {"card": "success"} + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify content was truncated + call_args = mock_orchestrator.formatter.format_success.call_args + content = call_args[1]["content"] + assert len(content) <= 2000 + + def test_handle_message_returns_error_on_exception(self, mock_orchestrator): + """handle_message should return error card when exception occurs.""" + mock_orchestrator.tmux_manager.get_session.side_effect = Exception("Session error") + + mock_orchestrator.formatter.format_error.return_value = {"card": "error"} + + result = mock_orchestrator.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify error card was returned + mock_orchestrator.formatter.format_error.assert_called_once() + + +class TestHandleCallback: + """Tests for handle_callback routing.""" + + @pytest.fixture + def mock_orchestrator(self): + """Create orchestrator with mocked dependencies.""" + from gsd_openclaw.orchestrator import Orchestrator + + mock_tmux_manager = MagicMock() + mock_formatter = MagicMock() + mock_approval_handler = MagicMock() + mock_executor = MagicMock() + + orchestrator = Orchestrator( + tmux_manager=mock_tmux_manager, + formatter=mock_formatter + ) + orchestrator.approval_handler = mock_approval_handler + orchestrator.executor = mock_executor + + return orchestrator + + def test_handle_callback_routes_approval_response_correctly(self, mock_orchestrator): + """handle_callback should route approval response correctly.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + mock_orchestrator.approval_handler.handle_callback.return_value = "approved" + mock_orchestrator.executor.execute.return_value = CommandResult( + output="Continuing...", + completed=True, + has_error=False, + error_type=None + ) + + mock_orchestrator.formatter.format_success.return_value = {"card": "success"} + + result = mock_orchestrator.handle_callback({ + "action": "approve", + "approval_id": "test-approval-id", + "project": "test-project" + }) + + # Verify callback was handled + mock_orchestrator.approval_handler.handle_callback.assert_called_once() + # Verify confirmation was sent + mock_orchestrator.formatter.format_success.assert_called_once() + + def test_handle_callback_sends_rejection_to_tmux(self, mock_orchestrator): + """handle_callback should send rejection to tmux session.""" + mock_tmux_session = MagicMock(spec=TmuxSession) + mock_orchestrator.tmux_manager.get_session.return_value = mock_tmux_session + + mock_orchestrator.approval_handler.handle_callback.return_value = "rejected" + mock_orchestrator.executor.execute.return_value = CommandResult( + output="Cancelled", + completed=True, + has_error=False, + error_type=None + ) + + mock_orchestrator.formatter.format_info.return_value = {"card": "info"} + + result = mock_orchestrator.handle_callback({ + "action": "reject", + "approval_id": "test-approval-id", + "project": "test-project" + }) + + # Verify callback was handled + mock_orchestrator.approval_handler.handle_callback.assert_called_once() + + def test_handle_callback_returns_error_for_unknown_approval(self, mock_orchestrator): + """handle_callback should return error for unknown approval.""" + mock_orchestrator.approval_handler.handle_callback.return_value = "error: approval not found" + + mock_orchestrator.formatter.format_error.return_value = {"card": "error"} + + result = mock_orchestrator.handle_callback({ + "action": "approve", + "approval_id": "unknown-id", + "project": "test-project" + }) + + # Verify error card was returned + mock_orchestrator.formatter.format_error.assert_called_once() + + +class TestIntegrationFlows: + """Integration-style tests for complete flows.""" + + @pytest.fixture + def orchestrator_with_real_components(self): + """Create orchestrator with real components (mock only tmux).""" + from gsd_openclaw.orchestrator import Orchestrator + from gsd_openclaw.response_formatter import FeishuResponseFormatter + from gsd_openclaw.approval_handler import ApprovalHandler + from gsd_openclaw.command_executor import CommandExecutor + + mock_tmux_manager = MagicMock() + + formatter = FeishuResponseFormatter() + approval_handler = ApprovalHandler(formatter) + + orchestrator = Orchestrator( + tmux_manager=mock_tmux_manager, + formatter=formatter + ) + orchestrator.approval_handler = approval_handler + orchestrator.executor = MagicMock() + + return orchestrator + + def test_full_flow_success(self, orchestrator_with_real_components): + """Test full flow: handle_message -> session creation -> command execution -> success card.""" + orch = orchestrator_with_real_components + + mock_tmux_session = MagicMock(spec=TmuxSession) + orch.tmux_manager.get_session.return_value = None + orch.tmux_manager.create_session.return_value = mock_tmux_session + + orch.executor.execute.return_value = CommandResult( + output="Status: All systems operational", + completed=True, + has_error=False, + error_type=None + ) + + result = orch.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify session was created + orch.tmux_manager.create_session.assert_called_once_with("test-project") + # Verify result is a card + assert "schema" in result + assert result["header"]["template"] == "green" # success theme + + def test_error_flow(self, orchestrator_with_real_components): + """Test error flow: handle_message -> command fails -> error card.""" + orch = orchestrator_with_real_components + + mock_tmux_session = MagicMock(spec=TmuxSession) + orch.tmux_manager.get_session.return_value = mock_tmux_session + + orch.executor.execute.return_value = CommandResult( + output="Error: Command execution failed", + completed=True, + has_error=True, + error_type="error" + ) + + result = orch.handle_message( + user_id="user1", + message="/gsd:status", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify result is an error card + assert "schema" in result + assert result["header"]["template"] == "red" # error theme + + def test_approval_flow(self, orchestrator_with_real_components): + """Test approval flow: handle_message -> approval detected -> approval card.""" + orch = orchestrator_with_real_components + + mock_tmux_session = MagicMock(spec=TmuxSession) + orch.tmux_manager.get_session.return_value = mock_tmux_session + + orch.executor.execute.return_value = CommandResult( + output="AskUserQuestion: Do you want to continue?", + completed=False, + has_error=False, + error_type=None + ) + + result = orch.handle_message( + user_id="user1", + message="/gsd:plan feature", + channel="test-channel", + context={"project": "test-project"} + ) + + # Verify result is an approval card with buttons + assert "schema" in result + # Check for action buttons + elements = result.get("elements", []) + has_actions = any(e.get("tag") == "action" for e in elements) + assert has_actions, "Approval card should have action buttons" + + def test_callback_flow_approval(self, orchestrator_with_real_components): + """Test callback flow: handle_callback -> approval processed -> confirmation.""" + orch = orchestrator_with_real_components + + mock_tmux_session = MagicMock(spec=TmuxSession) + orch.tmux_manager.get_session.return_value = mock_tmux_session + + # First create an approval via handle_message + orch.executor.execute.return_value = CommandResult( + output="AskUserQuestion: Continue?", + completed=False, + has_error=False, + error_type=None + ) + + orch.handle_message( + user_id="user1", + message="/gsd:plan feature", + channel="test-channel", + context={"project": "test-project"} + ) + + # Get the approval_id from the approval handler + approval_ids = list(orch.approval_handler.pending_approvals.keys()) + assert len(approval_ids) > 0, "Approval should be created" + approval_id = approval_ids[0] + + # Now simulate callback + orch.executor.execute.return_value = CommandResult( + output="Continuing with plan...", + completed=True, + has_error=False, + error_type=None + ) + + result = orch.handle_callback({ + "action": "approve", + "approval_id": approval_id, + "project": "test-project" + }) + + # Verify result is a confirmation card + assert "schema" in result + assert result["header"]["template"] == "green" # success theme + + def test_callback_flow_rejection(self, orchestrator_with_real_components): + """Test callback flow: handle_callback -> rejection processed -> confirmation.""" + orch = orchestrator_with_real_components + + mock_tmux_session = MagicMock(spec=TmuxSession) + orch.tmux_manager.get_session.return_value = mock_tmux_session + + # First create an approval via handle_message + orch.executor.execute.return_value = CommandResult( + output="AskUserQuestion: Continue?", + completed=False, + has_error=False, + error_type=None + ) + + orch.handle_message( + user_id="user1", + message="/gsd:plan feature", + channel="test-channel", + context={"project": "test-project"} + ) + + # Get the approval_id from the approval handler + approval_ids = list(orch.approval_handler.pending_approvals.keys()) + assert len(approval_ids) > 0, "Approval should be created" + approval_id = approval_ids[0] + + # Now simulate rejection callback + orch.executor.execute.return_value = CommandResult( + output="Operation cancelled", + completed=True, + has_error=False, + error_type=None + ) + + result = orch.handle_callback({ + "action": "reject", + "approval_id": approval_id, + "project": "test-project" + }) + + # Verify result is a confirmation card + assert "schema" in result