"""
Test Suite: delete_session MCP Tool - Agent Orchestration Platform
Comprehensive testing for session deletion with cascade cleanup,
work preservation, and security validation.
Author: Adder_1 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import pytest
import asyncio
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime
from typing import Dict, Any, List
# Import system under test
from src.interfaces.mcp_tools import AgentOrchestrationTools
from src.models.session import SessionState, SessionId
from src.models.agent import AgentState, AgentId, AgentStatus
from src.models.security import SecurityLevel, SecurityContext
from src.utils.errors import ValidationError, OperationError
from src.validators.session_deletion import SessionDeletionValidationResult
from src.utils.cascade_deletion import DeletionResult
from src.utils.work_preservation import PreservationResult
class TestDeleteSessionTool:
"""Comprehensive test suite for delete_session MCP tool."""
@pytest.fixture
async def mcp_tools(self):
"""Create MCP tools instance with mocked dependencies."""
# Mock managers
agent_manager = AsyncMock()
session_manager = AsyncMock()
iterm_manager = AsyncMock()
claude_manager = AsyncMock()
# Create tools instance
tools = AgentOrchestrationTools(
agent_manager=agent_manager,
session_manager=session_manager,
iterm_manager=iterm_manager,
claude_manager=claude_manager
)
# Mock audit logger
tools._audit_logger = AsyncMock()
return tools
@pytest.fixture
def sample_session_state(self):
"""Create sample session state for testing."""
return SessionState(
session_id=SessionId("test_session_123"),
name="Test Session",
root_path=Path("/tmp/test_session"),
created_at=datetime.now(),
last_activity=datetime.now(),
security_level=SecurityLevel.HIGH,
agents=[AgentId("agent_1"), AgentId("agent_2")],
max_agents=8,
is_active=True
)
@pytest.fixture
def sample_agent_states(self):
"""Create sample agent states for testing."""
return [
AgentState(
agent_id=AgentId("agent_1"),
session_id=SessionId("test_session_123"),
name="Agent_1",
status=AgentStatus.ACTIVE,
process_id=1234,
iterm_tab_id="tab_1",
has_critical_operations=False
),
AgentState(
agent_id=AgentId("agent_2"),
session_id=SessionId("test_session_123"),
name="Agent_2",
status=AgentStatus.ACTIVE,
process_id=1235,
iterm_tab_id="tab_2",
has_critical_operations=False
)
]
@pytest.mark.asyncio
async def test_successful_session_deletion_with_work_preservation(
self, mcp_tools, sample_session_state, sample_agent_states
):
"""Test successful session deletion with work preservation."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
for i, agent_state in enumerate(sample_agent_states):
mcp_tools.agent_manager.get_agent_state.return_value = agent_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True,
warning_message=None,
error_message=None
)
# Mock work preservation success
with patch('src.utils.work_preservation.WorkPreservationHandler') as mock_work_handler:
work_instance = AsyncMock()
mock_work_handler.return_value = work_instance
work_instance.preserve_session_work.return_value = PreservationResult(
success=True,
preserved_count=5,
preservation_path=Path("/tmp/preserved_work"),
preserved_files=["file1.py", "file2.js"],
git_commits=["abc123 Initial commit"]
)
# Mock cascade deletion success
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.prepare_cascade_deletion.return_value = ["agent_1", "agent_2"]
cascade_instance.execute_cascade_deletion.return_value = [
DeletionResult(agent_id="agent_1", success=True, error_message=None),
DeletionResult(agent_id="agent_2", success=True, error_message=None)
]
cascade_instance.cleanup_orphaned_resources.return_value = {
"orphaned_processes": 0,
"orphaned_tabs": 0,
"cleanup_success": True
}
# Execute deletion
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=True,
force=False
)
# Verify success
assert result["success"] is True
assert result["session_id"] == "test_session_123"
assert result["agents_deleted"] == 2
assert result["work_preserved"] is True
assert result["preserved_items"] == 5
assert "execution_duration_ms" in result
# Verify session manager was called
mcp_tools.session_manager.delete_session.assert_called_once()
@pytest.mark.asyncio
async def test_session_not_found_error(self, mcp_tools):
"""Test error when session does not exist."""
# Mock session not found
mcp_tools.session_manager.get_session_state.return_value = None
result = await mcp_tools.delete_session(
session_id="nonexistent_session",
cleanup_agents=True,
preserve_work=True
)
assert result["success"] is False
assert result["error_type"] == "VALIDATION_ERROR"
assert "not found" in result["error"].lower()
@pytest.mark.asyncio
async def test_validation_failure(self, mcp_tools, sample_session_state):
"""Test deletion blocked by validation failure."""
# Setup session state
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
# Mock validation failure
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=False,
session_id="test_session_123",
can_delete=False,
warning_message=None,
error_message="Session has critical operations in progress"
)
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=True,
force=False
)
assert result["success"] is False
assert result["error_type"] == "OPERATION_ERROR"
assert "validation failed" in result["error"].lower()
@pytest.mark.asyncio
async def test_work_preservation_failure_without_force(self, mcp_tools, sample_session_state):
"""Test deletion failure when work preservation fails and force=False."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True,
warning_message=None,
error_message=None
)
# Mock work preservation failure
with patch('src.utils.work_preservation.WorkPreservationHandler') as mock_work_handler:
work_instance = AsyncMock()
mock_work_handler.return_value = work_instance
work_instance.preserve_session_work.side_effect = Exception("Git repository corrupted")
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=True,
force=False
)
assert result["success"] is False
assert result["error_type"] == "OPERATION_ERROR"
assert "preservation failed" in result["error"].lower()
@pytest.mark.asyncio
async def test_work_preservation_failure_with_force(self, mcp_tools, sample_session_state, sample_agent_states):
"""Test deletion continues when work preservation fails but force=True."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
for agent_state in sample_agent_states:
mcp_tools.agent_manager.get_agent_state.return_value = agent_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock work preservation failure
with patch('src.utils.work_preservation.WorkPreservationHandler') as mock_work_handler:
work_instance = AsyncMock()
mock_work_handler.return_value = work_instance
work_instance.preserve_session_work.side_effect = Exception("Git repository corrupted")
# Mock cascade deletion success
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.prepare_cascade_deletion.return_value = ["agent_1", "agent_2"]
cascade_instance.execute_cascade_deletion.return_value = [
DeletionResult(agent_id="agent_1", success=True),
DeletionResult(agent_id="agent_2", success=True)
]
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=True,
force=True
)
# Should succeed despite work preservation failure
assert result["success"] is True
assert result["work_preserved"] is False
assert result["agents_deleted"] == 2
@pytest.mark.asyncio
async def test_partial_agent_deletion_failure(self, mcp_tools, sample_session_state, sample_agent_states):
"""Test handling of partial agent deletion failures."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
for agent_state in sample_agent_states:
mcp_tools.agent_manager.get_agent_state.return_value = agent_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock work preservation success
with patch('src.utils.work_preservation.WorkPreservationHandler') as mock_work_handler:
work_instance = AsyncMock()
mock_work_handler.return_value = work_instance
work_instance.preserve_session_work.return_value = PreservationResult(
success=True,
preserved_count=3
)
# Mock partial agent deletion failure
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.prepare_cascade_deletion.return_value = ["agent_1", "agent_2"]
cascade_instance.execute_cascade_deletion.return_value = [
DeletionResult(agent_id="agent_1", success=True),
DeletionResult(agent_id="agent_2", success=False, error_message="Process termination timeout")
]
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=True,
force=False
)
assert result["success"] is False
assert result["error_type"] == "OPERATION_ERROR"
assert "failed to delete" in result["error"].lower()
@pytest.mark.asyncio
async def test_session_deletion_without_agent_cleanup(self, mcp_tools, sample_session_state):
"""Test session deletion with cleanup_agents=False."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock cascade deletion (should not be called)
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=False,
preserve_work=False,
force=False
)
assert result["success"] is True
assert result["agents_deleted"] == 0
assert result["work_preserved"] is False
# Verify agent deletion was not attempted
cascade_instance.execute_cascade_deletion.assert_not_called()
@pytest.mark.asyncio
async def test_session_deletion_without_work_preservation(self, mcp_tools, sample_session_state, sample_agent_states):
"""Test session deletion with preserve_work=False."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
for agent_state in sample_agent_states:
mcp_tools.agent_manager.get_agent_state.return_value = agent_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock cascade deletion success
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.prepare_cascade_deletion.return_value = ["agent_1", "agent_2"]
cascade_instance.execute_cascade_deletion.return_value = [
DeletionResult(agent_id="agent_1", success=True),
DeletionResult(agent_id="agent_2", success=True)
]
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
# Mock work preservation handler (should not be called)
with patch('src.utils.work_preservation.WorkPreservationHandler') as mock_work_handler:
work_instance = AsyncMock()
mock_work_handler.return_value = work_instance
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=False,
force=False
)
assert result["success"] is True
assert result["agents_deleted"] == 2
assert result["work_preserved"] is False
assert result["preserved_items"] == 0
# Verify work preservation was not attempted
work_instance.preserve_session_work.assert_not_called()
@pytest.mark.asyncio
async def test_empty_session_deletion(self, mcp_tools):
"""Test deletion of session with no agents."""
# Create empty session
empty_session = SessionState(
session_id=SessionId("empty_session"),
name="Empty Session",
root_path=Path("/tmp/empty_session"),
created_at=datetime.now(),
last_activity=datetime.now(),
security_level=SecurityLevel.HIGH,
agents=[], # No agents
max_agents=8,
is_active=True
)
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = empty_session
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="empty_session",
can_delete=True
)
# Mock cascade deletion (should handle empty agent list)
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
result = await mcp_tools.delete_session(
session_id="empty_session",
cleanup_agents=True,
preserve_work=False
)
assert result["success"] is True
assert result["agents_deleted"] == 0
assert result["work_preserved"] is False
# Verify session was still deleted
mcp_tools.session_manager.delete_session.assert_called_once()
@pytest.mark.asyncio
async def test_audit_logging(self, mcp_tools, sample_session_state, sample_agent_states):
"""Test comprehensive audit logging during deletion."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
for agent_state in sample_agent_states:
mcp_tools.agent_manager.get_agent_state.return_value = agent_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock cascade deletion success
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.prepare_cascade_deletion.return_value = ["agent_1", "agent_2"]
cascade_instance.execute_cascade_deletion.return_value = [
DeletionResult(agent_id="agent_1", success=True),
DeletionResult(agent_id="agent_2", success=True)
]
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=True,
preserve_work=False
)
assert result["success"] is True
# Verify audit logging was called
mcp_tools._audit_logger.log_event.assert_called()
# Verify audit log details
call_args = mcp_tools._audit_logger.log_event.call_args
assert call_args[1]["operation"] == "delete_session"
assert call_args[1]["details"]["session_id"] == "test_session_123"
assert call_args[1]["details"]["agents_deleted"] == 2
@pytest.mark.asyncio
async def test_performance_tracking(self, mcp_tools, sample_session_state):
"""Test performance statistics tracking during deletion."""
# Setup mocks
mcp_tools.session_manager.get_session_state.return_value = sample_session_state
# Mock validation success
with patch('src.validators.session_deletion.SessionDeletionValidator') as mock_validator:
validator_instance = AsyncMock()
mock_validator.return_value = validator_instance
validator_instance.validate_session_deletion.return_value = SessionDeletionValidationResult(
valid=True,
session_id="test_session_123",
can_delete=True
)
# Mock cascade deletion
with patch('src.utils.cascade_deletion.CascadeDeletionOrchestrator') as mock_cascade:
cascade_instance = AsyncMock()
mock_cascade.return_value = cascade_instance
cascade_instance.cleanup_orphaned_resources.return_value = {"cleanup_success": True}
# Mock performance tracking
mcp_tools._update_tool_performance = MagicMock()
result = await mcp_tools.delete_session(
session_id="test_session_123",
cleanup_agents=False,
preserve_work=False
)
assert result["success"] is True
assert "execution_duration_ms" in result
# Verify performance tracking was called
mcp_tools._update_tool_performance.assert_called_with(
"delete_session",
result["execution_duration_ms"],
success=True
)
class TestDeleteSessionIntegration:
"""Integration tests for delete_session with real components."""
@pytest.mark.asyncio
async def test_with_real_work_preservation_handler(self):
"""Test delete_session with actual WorkPreservationHandler."""
# Create temporary directory for test
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Create test files
test_file = temp_path / "test.py"
test_file.write_text("print('Hello, World!')")
# Create git repository
import subprocess
subprocess.run(["git", "init"], cwd=temp_path, capture_output=True)
subprocess.run(["git", "add", "."], cwd=temp_path, capture_output=True)
subprocess.run(["git", "commit", "-m", "Initial commit"], cwd=temp_path, capture_output=True)
# Make uncommitted changes
test_file.write_text("print('Modified!')")
# Test work preservation
from src.utils.work_preservation import WorkPreservationHandler
handler = WorkPreservationHandler()
# Create mock session state
session_state = SessionState(
session_id=SessionId("integration_test"),
name="Integration Test",
root_path=temp_path,
created_at=datetime.now(),
last_activity=datetime.now(),
security_level=SecurityLevel.HIGH,
agents=[],
max_agents=8,
is_active=True
)
result = await handler.preserve_session_work(
session_state=session_state,
include_git=True,
include_files=True,
include_metadata=True
)
assert result.success
assert result.preserved_count > 0
assert result.preservation_path.exists()
@pytest.mark.asyncio
async def test_concurrent_deletion_safety(self):
"""Test that concurrent deletion attempts are handled safely."""
# This would test race conditions and concurrency safety
# Implementation would require more complex setup
pass
