"""
Test Suite for send_message_to_agent MCP Tool
This module provides comprehensive testing for the send_message_to_agent tool
including unit tests, property-based tests, and security validation.
Author: ADDER_5 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime
from pathlib import Path
from hypothesis import given, strategies as st, settings
from hypothesis.stateful import rule, RuleBasedStateMachine, Bundle
from src.interfaces.mcp_tools import AgentOrchestrationTools
from src.models.ids import AgentId, SessionId
from src.models.agent import AgentInfo, AgentStatus, AgentSpecialization
from src.models.session import SessionState
from src.models.security import SecurityLevel, SecurityContext
from src.models.validation import ValidationError
from src.utils.errors import OperationError
from src.utils.message_sanitization import MessageSanitizer, SanitizationConfig
@pytest.fixture
def mock_managers():
"""Create mock managers for testing."""
agent_manager = AsyncMock()
session_manager = AsyncMock()
iterm_manager = AsyncMock()
claude_manager = AsyncMock()
return {
'agent_manager': agent_manager,
'session_manager': session_manager,
'iterm_manager': iterm_manager,
'claude_manager': claude_manager
}
@pytest.fixture
def mock_agent_info():
"""Create mock agent info for testing."""
return AgentInfo(
agent_id=AgentId("agent-123"),
session_id=SessionId("session-456"),
name="Agent_1",
status=AgentStatus.ACTIVE,
iterm_tab_id="tab-789",
specialization=AgentSpecialization.GENERAL
)
@pytest.fixture
def mock_session_state():
"""Create mock session state for testing."""
return SessionState(
session_id=SessionId("session-456"),
name="test-session",
root_path=Path("/test/path"),
security_level=SecurityLevel.HIGH,
agents={}
)
@pytest.fixture
def orchestration_tools(mock_managers):
"""Create AgentOrchestrationTools instance with mocked dependencies."""
tools = AgentOrchestrationTools(
agent_manager=mock_managers['agent_manager'],
session_manager=mock_managers['session_manager'],
iterm_manager=mock_managers['iterm_manager'],
claude_manager=mock_managers['claude_manager']
)
# Mock the audit logger
tools._audit_logger = AsyncMock()
return tools
class TestSendMessageToAgent:
"""Unit tests for send_message_to_agent tool."""
@pytest.mark.asyncio
async def test_send_message_success(
self,
orchestration_tools,
mock_managers,
mock_agent_info,
mock_session_state
):
"""Test successful message sending."""
# Setup mocks
mock_managers['agent_manager'].find_agent_by_name.return_value = mock_agent_info
mock_managers['session_manager'].get_session_state.return_value = mock_session_state
mock_managers['iterm_manager'].send_text.return_value = True
# Test message sending
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_1",
message="Test message",
prepend_adder=False
)
# Verify result
assert result["success"] is True
assert result["agent_name"] == "Agent_1"
assert result["message_delivered"] is True
assert "delivery_timestamp" in result
assert "execution_duration_ms" in result
# Verify mock calls
mock_managers['agent_manager'].find_agent_by_name.assert_called_once()
mock_managers['iterm_manager'].send_text.assert_called_once()
@pytest.mark.asyncio
async def test_send_message_with_adder_prompt(
self,
orchestration_tools,
mock_managers,
mock_agent_info,
mock_session_state
):
"""Test message sending with ADDER+ prompt prepending."""
# Setup mocks
mock_managers['agent_manager'].find_agent_by_name.return_value = mock_agent_info
mock_managers['session_manager'].get_session_state.return_value = mock_session_state
mock_managers['iterm_manager'].send_text.return_value = True
# Test message sending with ADDER+ prompt
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_1",
message="Test task",
prepend_adder=True
)
# Verify result
assert result["success"] is True
assert result["prepend_adder"] is True
# Verify the sent text includes ADDER+ formatting
call_args = mock_managers['iterm_manager'].send_text.call_args
sent_text = call_args[1]['text']
assert "Agent_1" in sent_text
assert "Task Assignment" in sent_text or "Task for Agent_1" in sent_text
@pytest.mark.asyncio
async def test_send_message_invalid_agent_name(self, orchestration_tools):
"""Test message sending with invalid agent name format."""
result = await orchestration_tools.send_message_to_agent(
agent_name="invalid_name",
message="Test message"
)
assert result["success"] is False
assert "error" in result
assert "Invalid agent name format" in result["error"]
@pytest.mark.asyncio
async def test_send_message_agent_not_found(
self,
orchestration_tools,
mock_managers
):
"""Test message sending when agent is not found."""
# Setup mock to return None (agent not found)
mock_managers['agent_manager'].find_agent_by_name.return_value = None
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_999",
message="Test message"
)
assert result["success"] is False
assert "Agent Agent_999 not found" in result["error"]
@pytest.mark.asyncio
async def test_send_message_inactive_agent(
self,
orchestration_tools,
mock_managers,
mock_agent_info
):
"""Test message sending to inactive agent."""
# Make agent inactive
mock_agent_info.status = AgentStatus.INACTIVE
mock_managers['agent_manager'].find_agent_by_name.return_value = mock_agent_info
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_1",
message="Test message"
)
assert result["success"] is False
assert "not active" in result["error"]
@pytest.mark.asyncio
async def test_send_message_delivery_failure(
self,
orchestration_tools,
mock_managers,
mock_agent_info,
mock_session_state
):
"""Test message sending when iTerm2 delivery fails."""
# Setup mocks
mock_managers['agent_manager'].find_agent_by_name.return_value = mock_agent_info
mock_managers['session_manager'].get_session_state.return_value = mock_session_state
mock_managers['iterm_manager'].send_text.return_value = False
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_1",
message="Test message"
)
assert result["success"] is False
assert "Message delivery to iTerm2 failed" in result["error"]
@pytest.mark.asyncio
async def test_message_sanitization(
self,
orchestration_tools,
mock_managers,
mock_agent_info,
mock_session_state
):
"""Test that dangerous message content is sanitized."""
# Setup mocks
mock_managers['agent_manager'].find_agent_by_name.return_value = mock_agent_info
mock_managers['session_manager'].get_session_state.return_value = mock_session_state
mock_managers['iterm_manager'].send_text.return_value = True
# Test with dangerous content
dangerous_message = "rm -rf / && echo 'malicious'; $(cat /etc/passwd)"
result = await orchestration_tools.send_message_to_agent(
agent_name="Agent_1",
message=dangerous_message,
prepend_adder=False
)
# Verify success (content was sanitized)
assert result["success"] is True
assert result["sanitized_message_length"] < len(dangerous_message)
# Verify the sent text was sanitized
call_args = mock_managers['iterm_manager'].send_text.call_args
sent_text = call_args[1]['text']
assert "rm -rf" not in sent_text
assert "$(cat" not in sent_text
class TestMessageSanitization:
"""Tests for message sanitization functionality."""
def test_sanitizer_removes_dangerous_chars(self):
"""Test that sanitizer removes dangerous shell characters."""
sanitizer = MessageSanitizer()
dangerous_input = "hello; rm -rf /; echo done"
sanitized = sanitizer.sanitize(dangerous_input)
assert ";" not in sanitized
assert "rm" not in sanitized or sanitized.count("rm") < dangerous_input.count("rm")
def test_sanitizer_handles_long_messages(self):
"""Test that sanitizer truncates long messages."""
sanitizer = MessageSanitizer(SanitizationConfig(max_length=100))
long_message = "A" * 200
sanitized = sanitizer.sanitize(long_message)
assert len(sanitized) <= 110 # Account for truncation suffix
assert "truncated" in sanitized
def test_sanitizer_preserves_safe_content(self):
"""Test that sanitizer preserves safe message content."""
sanitizer = MessageSanitizer()
safe_message = "Please implement the user authentication feature."
sanitized = sanitizer.sanitize(safe_message)
assert sanitized == safe_message
@given(st.text(min_size=1, max_size=1000))
@settings(max_examples=50)
def test_sanitization_is_idempotent(self, message):
"""Property: Sanitizing a message twice should give same result."""
sanitizer = MessageSanitizer()
first_sanitized = sanitizer.sanitize(message)
second_sanitized = sanitizer.sanitize(first_sanitized)
assert first_sanitized == second_sanitized
@given(st.text(min_size=1, max_size=500))
@settings(max_examples=50)
def test_sanitized_messages_are_safe(self, message):
"""Property: All sanitized messages should pass safety validation."""
sanitizer = MessageSanitizer()
sanitized = sanitizer.sanitize(message)
# Basic safety checks
assert len(sanitized) <= sanitizer.config.max_length
# No shell metacharacters in strict mode
if sanitizer.config.level.value == "strict":
dangerous_chars = {';', '|', '&', '$', '`', '(', ')', '<', '>'}
assert not any(char in sanitized for char in dangerous_chars)
class SendMessageStateMachine(RuleBasedStateMachine):
"""Stateful testing for send_message_to_agent operations."""
agents = Bundle('agents')
def __init__(self):
super().__init__()
self.active_agents = set()
self.message_count = 0
@rule(target=agents, agent_name=st.text(min_size=7, max_size=15).filter(
lambda x: x.startswith("Agent_") and x[6:].isdigit()
))
def create_agent(self, agent_name):
"""Create a new agent for testing."""
self.active_agents.add(agent_name)
return agent_name
@rule(agent=agents, message=st.text(min_size=1, max_size=1000))
def send_message(self, agent, message):
"""Send message to agent."""
if agent in self.active_agents:
self.message_count += 1
# Verify message can be sanitized
sanitizer = MessageSanitizer()
sanitized = sanitizer.sanitize(message)
assert len(sanitized) <= sanitizer.config.max_length
# Property-based test runner
class TestSendMessageProperties:
"""Property-based tests for send_message_to_agent."""
@given(
agent_name=st.text(min_size=7, max_size=15).filter(
lambda x: x.startswith("Agent_") and x[6:].isdigit()
),
message=st.text(min_size=1, max_size=10000),
prepend_adder=st.booleans()
)
@settings(max_examples=20)
def test_send_message_properties(self, agent_name, message, prepend_adder):
"""Property: Valid inputs should produce consistent results."""
# This would be run with actual mocked components
assert agent_name.startswith("Agent_")
assert len(message) > 0
assert isinstance(prepend_adder, bool)
# Verify message sanitization properties
sanitizer = MessageSanitizer()
sanitized = sanitizer.sanitize(message)
assert len(sanitized) <= sanitizer.config.max_length
# Integration test markers
pytestmark = [
pytest.mark.asyncio,
pytest.mark.interfaces,
pytest.mark.mcp_tools
]
