"""
Delete Agent Tool Tests
Comprehensive test suite for the delete_agent MCP tool implementation including
unit tests, integration tests, and property-based testing scenarios.
Test Categories:
- Input Validation: Agent name format, permissions, state checks
- Core Functionality: Graceful termination, resource cleanup, state deletion
- Error Handling: Timeout scenarios, partial failures, recovery
- Security: Permission validation, audit logging, secure deletion
- Performance: Concurrent deletions, resource cleanup timing
Author: ADDER_4 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch, MagicMock
from datetime import datetime
from pathlib import Path
from typing import Dict, Any
from hypothesis import given, strategies as st, assume
from hypothesis.stateful import RuleBasedStateMachine, rule, invariant
# Import implementation
from src.interfaces.mcp_tools import AgentOrchestrationTools
from src.interfaces.manager_protocols import (
AgentManagerProtocol, SessionManagerProtocol,
ITermManagerProtocol, ClaudeManagerProtocol
)
# Import types
from src.models.ids import AgentId, SessionId
from src.models.agent import AgentState, AgentInfo
from src.models.security import SecurityContext, SecurityLevel
from src.models.validation import ValidationError
from src.utils.errors import OperationError
# Import test fixtures
from tests.fixtures.domain import (
create_test_agent_id, create_test_session_id,
create_test_agent_state, create_test_security_context
)
class TestDeleteAgentValidation:
"""Test input validation for delete_agent tool."""
@pytest.fixture
def mock_managers(self):
"""Create mock managers for testing."""
agent_manager = Mock(spec=AgentManagerProtocol)
session_manager = Mock(spec=SessionManagerProtocol)
iterm_manager = Mock(spec=ITermManagerProtocol)
claude_manager = Mock(spec=ClaudeManagerProtocol)
return agent_manager, session_manager, iterm_manager, claude_manager
@pytest.fixture
async def tools(self, mock_managers):
"""Create AgentOrchestrationTools instance with mocks."""
agent_manager, session_manager, iterm_manager, claude_manager = mock_managers
tools = AgentOrchestrationTools(
agent_manager=agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
await tools.initialize()
return tools
@pytest.mark.asyncio
async def test_invalid_agent_name_format(self, tools):
"""Test validation rejects invalid agent name formats."""
invalid_names = [
"", # Empty string
"Agent", # Missing number
"Agent_", # Missing number
"agent_1", # Lowercase
"Agent_One", # Non-numeric
"Agent_1_2", # Extra parts
"Agent 1", # Space instead of underscore
"1_Agent", # Wrong order
]
for invalid_name in invalid_names:
result = await tools.delete_agent(
agent_name=invalid_name,
force=False
)
assert not result["success"]
assert "Invalid agent name format" in result["error"]
assert result["error_type"] == "ValidationError"
@pytest.mark.asyncio
async def test_agent_not_found(self, tools, mock_managers):
"""Test handling when agent doesn't exist."""
agent_manager, _, _, _ = mock_managers
# Mock agent not found
agent_manager.find_agent_by_name = AsyncMock(return_value=None)
result = await tools.delete_agent(
agent_name="Agent_99",
force=False
)
assert not result["success"]
assert "not found" in result["error"]
assert result["error_type"] == "ValidationError"
@pytest.mark.asyncio
async def test_critical_operations_block_deletion(self, tools, mock_managers):
"""Test that critical operations prevent deletion without force."""
agent_manager, _, _, _ = mock_managers
# Create agent with critical operations
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_123",
process_id=12345
)
agent_state = Mock(
has_critical_operations=True,
get_critical_operations=Mock(return_value=[
Mock(operation_type="database_migration"),
Mock(operation_type="file_encryption")
])
)
agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
agent_manager.get_agent_state = AsyncMock(return_value=agent_state)
result = await tools.delete_agent(
agent_name="Agent_1",
force=False
)
assert not result["success"]
assert "critical operations" in result["error"]
assert "force=True" in result["error"]
class TestDeleteAgentCore:
"""Test core delete_agent functionality."""
@pytest.fixture
def mock_agent_setup(self, mock_managers):
"""Set up mock agent for deletion testing."""
agent_manager, session_manager, iterm_manager, claude_manager = mock_managers
# Create test agent info
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_123",
process_id=12345,
allocated_memory_mb=512,
allocated_cpu_percent=25.0
)
# Create test agent state (no critical operations)
agent_state = Mock(
has_critical_operations=False,
get_critical_operations=Mock(return_value=[])
)
# Configure mocks
agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
agent_manager.get_agent_state = AsyncMock(return_value=agent_state)
agent_manager.delete_agent = AsyncMock(return_value=True)
iterm_manager.send_text = AsyncMock()
iterm_manager.close_tab = AsyncMock()
return agent_info, agent_state
@pytest.mark.asyncio
async def test_successful_graceful_deletion(self, tools, mock_managers, mock_agent_setup):
"""Test successful graceful agent deletion."""
agent_info, _ = mock_agent_setup
# Mock process termination check
with patch.object(tools, '_is_process_running') as mock_is_running:
mock_is_running.side_effect = [True, False] # Running, then terminated
result = await tools.delete_agent(
agent_name="Agent_1",
force=False,
timeout_seconds=30
)
assert result["success"]
assert result["agent_id"] == str(agent_info.agent_id)
assert result["cleanup_summary"]["graceful_termination"]
assert result["cleanup_summary"]["process_terminated"]
assert result["cleanup_summary"]["iterm_tab_closed"]
assert result["cleanup_summary"]["state_deleted"]
assert result["resources_freed"]["memory_mb"] == 512
assert result["resources_freed"]["cpu_percent"] == 25.0
@pytest.mark.asyncio
async def test_forced_termination_after_timeout(self, tools, mock_managers, mock_agent_setup):
"""Test forced termination when graceful fails."""
agent_info, _ = mock_agent_setup
# Mock process that doesn't terminate gracefully
with patch.object(tools, '_is_process_running') as mock_is_running:
# Process stays running during graceful attempt
mock_is_running.side_effect = [True] * 35 + [False]
with patch.object(tools, '_force_terminate_process') as mock_force:
mock_force.return_value = None
result = await tools.delete_agent(
agent_name="Agent_1",
force=False,
timeout_seconds=1 # Short timeout for test
)
assert result["success"]
assert not result["cleanup_summary"]["graceful_termination"]
assert result["cleanup_summary"]["process_terminated"]
@pytest.mark.asyncio
async def test_force_deletion_bypasses_checks(self, tools, mock_managers, mock_agent_setup):
"""Test force deletion bypasses safety checks."""
agent_info, agent_state = mock_agent_setup
# Add critical operations
agent_state.has_critical_operations = True
with patch.object(tools, '_is_process_running') as mock_is_running:
mock_is_running.return_value = False
with patch.object(tools, '_force_terminate_process') as mock_force:
mock_force.return_value = None
result = await tools.delete_agent(
agent_name="Agent_1",
force=True
)
assert result["success"]
assert result["force"]
# Should not check for critical operations when force=True
assert mock_managers[0].get_agent_state.call_count == 0
class TestDeleteAgentErrorHandling:
"""Test error handling in delete_agent tool."""
@pytest.mark.asyncio
async def test_partial_deletion_tracking(self, tools, mock_managers):
"""Test tracking of partial deletion on failure."""
agent_manager, _, iterm_manager, _ = mock_managers
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_123",
process_id=12345
)
agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
agent_manager.get_agent_state = AsyncMock(return_value=Mock(has_critical_operations=False))
# Make iTerm tab closure fail
iterm_manager.close_tab = AsyncMock(side_effect=Exception("Tab closure failed"))
with patch.object(tools, '_is_process_running', return_value=False):
result = await tools.delete_agent(
agent_name="Agent_1",
force=False
)
assert not result["success"]
assert "partial_cleanup" in result
assert result["partial_cleanup"]["manual_cleanup_required"]
@pytest.mark.asyncio
async def test_process_termination_failure(self, tools, mock_managers):
"""Test handling of process that won't terminate."""
agent_manager, _, _, _ = mock_managers
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_123",
process_id=12345
)
agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
agent_manager.get_agent_state = AsyncMock(return_value=Mock(has_critical_operations=False))
# Mock process that won't die
with patch.object(tools, '_is_process_running', return_value=True):
with patch.object(tools, '_force_terminate_process') as mock_force:
mock_force.return_value = None
result = await tools.delete_agent(
agent_name="Agent_1",
force=False,
timeout_seconds=1
)
assert not result["success"]
assert "Failed to terminate agent process" in result["error"]
class TestDeleteAgentProperties:
"""Property-based tests for delete_agent tool."""
@given(
agent_number=st.integers(min_value=1, max_value=999),
force=st.booleans(),
timeout=st.integers(min_value=1, max_value=300)
)
@pytest.mark.asyncio
async def test_delete_agent_properties(self, agent_number, force, timeout):
"""Test delete_agent with various property combinations."""
# Create mocks
agent_manager = Mock(spec=AgentManagerProtocol)
session_manager = Mock(spec=SessionManagerProtocol)
iterm_manager = Mock(spec=ITermManagerProtocol)
claude_manager = Mock(spec=ClaudeManagerProtocol)
tools = AgentOrchestrationTools(
agent_manager=agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
await tools.initialize()
# Configure successful deletion
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id=f"tab_{agent_number}",
process_id=10000 + agent_number,
allocated_memory_mb=256,
allocated_cpu_percent=10.0
)
agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
agent_manager.get_agent_state = AsyncMock(
return_value=Mock(has_critical_operations=False)
)
agent_manager.delete_agent = AsyncMock(return_value=True)
iterm_manager.send_text = AsyncMock()
iterm_manager.close_tab = AsyncMock()
with patch.object(tools, '_is_process_running', return_value=False):
result = await tools.delete_agent(
agent_name=f"Agent_{agent_number}",
force=force,
timeout_seconds=timeout
)
# Properties to verify
assert isinstance(result, dict)
assert "success" in result
assert "agent_name" in result
assert result["agent_name"] == f"Agent_{agent_number}"
if result["success"]:
assert "cleanup_summary" in result
assert "resources_freed" in result
assert "execution_duration_ms" in result
assert result["execution_duration_ms"] >= 0
else:
assert "error" in result
assert "error_type" in result
class AgentDeletionStateMachine(RuleBasedStateMachine):
"""State machine for testing agent deletion workflows."""
def __init__(self):
super().__init__()
self.agents = {} # agent_name -> agent_info
self.deleted_agents = set()
self.tools = None
self.agent_manager = None
async def setup(self):
"""Async setup for state machine."""
# Create mocks
self.agent_manager = Mock(spec=AgentManagerProtocol)
session_manager = Mock(spec=SessionManagerProtocol)
iterm_manager = Mock(spec=ITermManagerProtocol)
claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
agent_manager=self.agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
await self.tools.initialize()
# Configure mock behaviors
iterm_manager.send_text = AsyncMock()
iterm_manager.close_tab = AsyncMock()
@rule(agent_number=st.integers(min_value=1, max_value=10))
async def create_agent(self, agent_number):
"""Create an agent in the system."""
agent_name = f"Agent_{agent_number}"
if agent_name not in self.agents and agent_name not in self.deleted_agents:
agent_info = Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id=f"tab_{agent_number}",
process_id=10000 + agent_number,
allocated_memory_mb=256,
allocated_cpu_percent=10.0
)
self.agents[agent_name] = agent_info
@rule(
agent_number=st.integers(min_value=1, max_value=10),
force=st.booleans()
)
async def delete_agent(self, agent_number, force):
"""Delete an agent from the system."""
agent_name = f"Agent_{agent_number}"
# Configure mock for this specific call
if agent_name in self.agents:
agent_info = self.agents[agent_name]
self.agent_manager.find_agent_by_name = AsyncMock(return_value=agent_info)
self.agent_manager.get_agent_state = AsyncMock(
return_value=Mock(has_critical_operations=False)
)
self.agent_manager.delete_agent = AsyncMock(return_value=True)
else:
self.agent_manager.find_agent_by_name = AsyncMock(return_value=None)
with patch.object(self.tools, '_is_process_running', return_value=False):
result = await self.tools.delete_agent(
agent_name=agent_name,
force=force
)
if agent_name in self.agents and result["success"]:
del self.agents[agent_name]
self.deleted_agents.add(agent_name)
@invariant()
async def agents_are_unique(self):
"""Agents should not exist in both active and deleted sets."""
assert len(self.agents.keys() & self.deleted_agents) == 0
@invariant()
async def deleted_agents_stay_deleted(self):
"""Once deleted, agents should not reappear."""
for agent_name in self.deleted_agents:
assert agent_name not in self.agents
# Integration test
class TestDeleteAgentIntegration:
"""Integration tests for delete_agent with real component interaction."""
@pytest.mark.asyncio
@pytest.mark.integration
async def test_end_to_end_agent_deletion(self):
"""Test complete agent deletion workflow."""
# This would test with real manager implementations
# For now, using sophisticated mocks
# Create sophisticated mock managers that simulate real behavior
agent_manager = Mock(spec=AgentManagerProtocol)
session_manager = Mock(spec=SessionManagerProtocol)
iterm_manager = Mock(spec=ITermManagerProtocol)
claude_manager = Mock(spec=ClaudeManagerProtocol)
# Simulate real agent state
agent_states = {}
async def mock_find_agent(agent_name, **kwargs):
if agent_name == "Agent_1":
return Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_real_123",
process_id=54321,
allocated_memory_mb=1024,
allocated_cpu_percent=50.0
)
return None
async def mock_delete_agent(agent_name, **kwargs):
if agent_name == "Agent_1":
# Simulate state cleanup
return True
return False
agent_manager.find_agent_by_name = mock_find_agent
agent_manager.delete_agent = mock_delete_agent
agent_manager.get_agent_state = AsyncMock(
return_value=Mock(has_critical_operations=False)
)
iterm_manager.send_text = AsyncMock()
iterm_manager.close_tab = AsyncMock()
tools = AgentOrchestrationTools(
agent_manager=agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
await tools.initialize()
# Test deletion
with patch.object(tools, '_is_process_running') as mock_is_running:
mock_is_running.side_effect = [True, True, False] # Graceful termination
result = await tools.delete_agent(
agent_name="Agent_1",
force=False,
timeout_seconds=30
)
# Verify complete workflow
assert result["success"]
assert result["cleanup_summary"]["graceful_termination"]
assert result["cleanup_summary"]["process_terminated"]
assert result["cleanup_summary"]["iterm_tab_closed"]
assert result["cleanup_summary"]["state_deleted"]
assert result["resources_freed"]["memory_mb"] == 1024
# Verify agent is gone
assert await agent_manager.find_agent_by_name("Agent_1") is None
# Performance tests
class TestDeleteAgentPerformance:
"""Performance tests for delete_agent tool."""
@pytest.mark.asyncio
@pytest.mark.performance
async def test_deletion_performance(self):
"""Test agent deletion completes within performance targets."""
import time
# Setup mocks for performance testing
agent_manager = Mock(spec=AgentManagerProtocol)
session_manager = Mock(spec=SessionManagerProtocol)
iterm_manager = Mock(spec=ITermManagerProtocol)
claude_manager = Mock(spec=ClaudeManagerProtocol)
# Configure instant responses
agent_manager.find_agent_by_name = AsyncMock(
return_value=Mock(
agent_id=create_test_agent_id(),
session_id=create_test_session_id(),
iterm_tab_id="tab_perf",
process_id=99999,
allocated_memory_mb=256,
allocated_cpu_percent=10.0
)
)
agent_manager.get_agent_state = AsyncMock(
return_value=Mock(has_critical_operations=False)
)
agent_manager.delete_agent = AsyncMock(return_value=True)
iterm_manager.send_text = AsyncMock()
iterm_manager.close_tab = AsyncMock()
tools = AgentOrchestrationTools(
agent_manager=agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
await tools.initialize()
# Measure deletion time
with patch.object(tools, '_is_process_running', return_value=False):
start_time = time.time()
result = await tools.delete_agent(
agent_name="Agent_1",
force=True # Skip graceful termination wait
)
duration = time.time() - start_time
assert result["success"]
assert duration < 2.0 # Should complete within 2 seconds for force deletion
assert result["execution_duration_ms"] < 2000
if __name__ == "__main__":
pytest.main([__file__, "-v"])
