starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(03-03): add failing tests for orchestrator

Browse Source 
- Test orchestrator initialization with dependencies
- Test command parsing for /gsd:* pattern extraction
- Test handle_message routing (session creation, command execution, error handling)
- Test handle_callback routing for approval workflow
- Add integration-style tests for full flows

18 tests total covering all orchestrator functionality.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
starmen1265
2026-03-11 16:14:13 +08:00
parent 9cd8ccb2e5
commit 086be891b9
1 changed files with 522 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gsd-openclaw/tests/test_orchestrator.py
+522
View File
@@ -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