"""
Comprehensive test suite for create_session MCP tool.
Tests all aspects of session creation including validation, security,
Git integration, and error handling with property-based testing.
Test Categories:
- Input validation and sanitization
- Security boundary establishment
- Git integration detection
- Filesystem operations
- Error handling and recovery
- Performance characteristics
Author: Adder_3 | Created: 2025-06-26
"""
import asyncio
import tempfile
import shutil
import pytest
from unittest.mock import Mock, AsyncMock, patch
from pathlib import Path
from typing import Dict, Any
from hypothesis import given, strategies as st
from hypothesis import settings, HealthCheck
# Import system under test
from src.interfaces.mcp_tools import AgentOrchestrationTools
from src.models.session import SecurityLevel, SessionStatus
from src.models.ids import SessionId
from src.core.session_manager import SessionManager
from src.interfaces.manager_protocols import (
SessionManagerProtocol, AgentManagerProtocol,
ITermManagerProtocol, ClaudeManagerProtocol
)
class TestCreateSessionValidation:
"""Test input validation and sanitization."""
def setup_method(self):
"""Set up test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
# Create temporary directory for testing
self.test_dir = tempfile.mkdtemp()
def teardown_method(self):
"""Clean up test fixtures."""
if Path(self.test_dir).exists():
shutil.rmtree(self.test_dir)
@pytest.mark.asyncio
async def test_valid_session_creation(self):
"""Test successful session creation with valid inputs."""
# Setup
session_id = SessionId.generate()
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = True
mock_result.session_id = session_id
mock_result.git_integrated = True
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = "test_fingerprint"
mock_result.warnings = []
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
# Execute
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="test_session",
security_level="HIGH"
)
# Verify
assert result["success"] is True
assert result["session_name"] == "test_session"
assert result["root_path"] == str(Path(self.test_dir).resolve())
assert result["security_level"] == "HIGH"
assert result["git_integrated"] is True
assert result["task_monitoring_enabled"] is True
assert "execution_duration_ms" in result
@pytest.mark.asyncio
async def test_invalid_root_path(self):
"""Test validation of invalid root paths."""
test_cases = [
("/nonexistent/path", "PATH_NOT_FOUND"),
("relative/path", "VALIDATION_ERROR"),
("", "VALIDATION_ERROR"),
]
for invalid_path, expected_error_type in test_cases:
result = await self.tools.create_session(
root_path=invalid_path,
session_name="test_session"
)
assert result["success"] is False
assert result["error_type"] == expected_error_type
@pytest.mark.asyncio
async def test_invalid_session_name(self):
"""Test validation of invalid session names."""
test_cases = [
("", "VALIDATION_ERROR"),
(" ", "VALIDATION_ERROR"),
("x" * 101, "VALIDATION_ERROR"), # Too long
]
for invalid_name, expected_error_type in test_cases:
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name=invalid_name
)
assert result["success"] is False
assert result["error_type"] == expected_error_type
@pytest.mark.asyncio
async def test_invalid_security_level(self):
"""Test validation of invalid security levels."""
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="test_session",
security_level="INVALID"
)
assert result["success"] is False
assert result["error_type"] == "VALIDATION_ERROR"
assert "Invalid security level" in result["error"]
@pytest.mark.asyncio
async def test_duplicate_session_name(self):
"""Test prevention of duplicate session names."""
# Setup existing session
existing_session = Mock()
existing_session.name = "existing_session"
self.mock_session_manager.sessions = {"id1": existing_session}
self.mock_session_manager.max_sessions = 10
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="existing_session"
)
assert result["success"] is False
assert result["error_type"] == "DUPLICATE_SESSION"
assert "already exists" in result["error"]
@pytest.mark.asyncio
async def test_capacity_limit_reached(self):
"""Test handling of maximum session capacity."""
# Setup at capacity
self.mock_session_manager.sessions = {f"id{i}": Mock() for i in range(5)}
self.mock_session_manager.max_sessions = 5
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="test_session"
)
assert result["success"] is False
assert result["error_type"] == "CAPACITY_ERROR"
assert "Maximum session limit" in result["error"]
class TestCreateSessionSecurity:
"""Test security aspects of session creation."""
def setup_method(self):
"""Set up security test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
self.test_dir = tempfile.mkdtemp()
def teardown_method(self):
"""Clean up security test fixtures."""
if Path(self.test_dir).exists():
shutil.rmtree(self.test_dir)
@pytest.mark.asyncio
async def test_path_traversal_prevention(self):
"""Test prevention of path traversal attacks."""
malicious_paths = [
"../../../etc/passwd",
"/etc/passwd",
self.test_dir + "/../../../etc/passwd",
"../../../../usr/bin/python",
]
for malicious_path in malicious_paths:
result = await self.tools.create_session(
root_path=malicious_path,
session_name="test_session"
)
# Should fail validation, not create session with malicious path
assert result["success"] is False
assert result["error_type"] in ["PATH_NOT_FOUND", "VALIDATION_ERROR"]
@pytest.mark.asyncio
async def test_session_name_sanitization(self):
"""Test sanitization of session names."""
# Test with special characters that should be sanitized
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="test<script>alert('xss')</script>session"
)
# Should not contain dangerous characters in response
if result["success"]:
assert "<script>" not in result["session_name"]
assert "alert" not in result["session_name"]
@pytest.mark.asyncio
async def test_security_level_enforcement(self):
"""Test that security levels are properly enforced."""
security_levels = ["LOW", "MEDIUM", "HIGH", "MAXIMUM"]
for level in security_levels:
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = True
mock_result.session_id = SessionId.generate()
mock_result.git_integrated = True
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = f"test_fingerprint_{level}"
mock_result.warnings = []
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name=f"test_session_{level}",
security_level=level
)
if result["success"]:
assert result["security_level"] == level
class TestCreateSessionGitIntegration:
"""Test Git integration aspects."""
def setup_method(self):
"""Set up Git integration test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
self.test_dir = tempfile.mkdtemp()
def teardown_method(self):
"""Clean up Git test fixtures."""
if Path(self.test_dir).exists():
shutil.rmtree(self.test_dir)
@pytest.mark.asyncio
async def test_git_repository_detection(self):
"""Test detection of Git repositories."""
# Setup mock session manager
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = True
mock_result.session_id = SessionId.generate()
mock_result.git_integrated = True # Git detected
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = "test_fingerprint"
mock_result.warnings = []
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="git_test_session"
)
if result["success"]:
assert "git_integrated" in result
# SessionManager should have been called with auto_git_integration=True
self.mock_session_manager.create_session.assert_called_once()
call_args = self.mock_session_manager.create_session.call_args[1]
assert call_args["auto_git_integration"] is True
@pytest.mark.asyncio
async def test_non_git_directory_handling(self):
"""Test handling of non-Git directories."""
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = True
mock_result.session_id = SessionId.generate()
mock_result.git_integrated = False # No Git detected
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = "test_fingerprint"
mock_result.warnings = ["No Git repository detected"]
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="non_git_session"
)
if result["success"]:
assert result["git_integrated"] is False
assert len(result["warnings"]) > 0
class TestCreateSessionErrorHandling:
"""Test error handling and recovery."""
def setup_method(self):
"""Set up error handling test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
self.test_dir = tempfile.mkdtemp()
def teardown_method(self):
"""Clean up error handling test fixtures."""
if Path(self.test_dir).exists():
shutil.rmtree(self.test_dir)
@pytest.mark.asyncio
async def test_session_manager_failure(self):
"""Test handling of SessionManager failures."""
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
# Mock SessionManager to raise exception
self.mock_session_manager.create_session = AsyncMock(
side_effect=Exception("Session manager failed")
)
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="fail_test"
)
assert result["success"] is False
assert result["error_type"] == "INTERNAL_ERROR"
assert "Failed to create session" in result["error"]
@pytest.mark.asyncio
async def test_session_manager_returns_failure(self):
"""Test handling when SessionManager returns failure result."""
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = False
mock_result.message = "Session creation failed due to disk space"
mock_result.warnings = ["Low disk space"]
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="disk_fail_test"
)
assert result["success"] is False
assert result["error_type"] == "SESSION_CREATION_ERROR"
assert result["error"] == "Session creation failed due to disk space"
assert result["warnings"] == ["Low disk space"]
@pytest.mark.asyncio
async def test_unexpected_exception_handling(self):
"""Test handling of unexpected exceptions."""
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
# Mock to cause unexpected exception during validation
with patch('pathlib.Path.resolve', side_effect=RuntimeError("Unexpected error")):
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="unexpected_fail"
)
assert result["success"] is False
assert result["error_type"] == "UNEXPECTED_ERROR"
assert "Unexpected error" in result["error"]
@pytest.mark.performance
class TestCreateSessionPerformance:
"""Test performance characteristics of session creation."""
def setup_method(self):
"""Set up performance test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
self.test_dir = tempfile.mkdtemp()
def teardown_method(self):
"""Clean up performance test fixtures."""
if Path(self.test_dir).exists():
shutil.rmtree(self.test_dir)
@pytest.mark.asyncio
async def test_session_creation_performance(self):
"""Test that session creation meets performance targets."""
# Setup fast mock response
session_id = SessionId.generate()
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
mock_result = Mock()
mock_result.success = True
mock_result.session_id = session_id
mock_result.git_integrated = True
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = "perf_test_fingerprint"
mock_result.warnings = []
# Add small delay to simulate real work
async def mock_create_session(*args, **kwargs):
await asyncio.sleep(0.1) # 100ms simulated work
return mock_result
self.mock_session_manager.create_session = mock_create_session
# Measure execution time
import time
start_time = time.time()
result = await self.tools.create_session(
root_path=str(self.test_dir),
session_name="perf_test_session"
)
end_time = time.time()
execution_time = (end_time - start_time) * 1000 # Convert to ms
# Verify success and performance
assert result["success"] is True
assert execution_time < 5000 # Should complete in under 5 seconds
assert "execution_duration_ms" in result
assert result["execution_duration_ms"] > 0
# Property-based testing
class TestCreateSessionProperties:
"""Property-based tests for session creation."""
def setup_method(self):
"""Set up property test fixtures."""
self.mock_session_manager = Mock(spec=SessionManagerProtocol)
self.mock_agent_manager = Mock(spec=AgentManagerProtocol)
self.mock_iterm_manager = Mock(spec=ITermManagerProtocol)
self.mock_claude_manager = Mock(spec=ClaudeManagerProtocol)
self.tools = AgentOrchestrationTools(
session_manager=self.mock_session_manager,
agent_manager=self.mock_agent_manager,
iterm_manager=self.mock_iterm_manager,
claude_manager=self.mock_claude_manager
)
@given(
session_name=st.text(min_size=1, max_size=100),
security_level=st.sampled_from(["LOW", "MEDIUM", "HIGH"])
)
@settings(
max_examples=50,
suppress_health_check=[HealthCheck.function_scoped_fixture]
)
@pytest.mark.asyncio
async def test_session_name_sanitization_properties(self, session_name, security_level):
"""Test properties of session name sanitization."""
with tempfile.TemporaryDirectory() as test_dir:
# Setup
self.mock_session_manager.sessions = {}
self.mock_session_manager.max_sessions = 10
# Mock successful result for valid inputs
if len(session_name.strip()) > 0:
mock_result = Mock()
mock_result.success = True
mock_result.session_id = SessionId.generate()
mock_result.git_integrated = False
mock_result.task_monitoring_enabled = True
mock_result.security_fingerprint = "prop_test_fingerprint"
mock_result.warnings = []
self.mock_session_manager.create_session = AsyncMock(return_value=mock_result)
result = await self.tools.create_session(
root_path=test_dir,
session_name=session_name,
security_level=security_level
)
# Properties that should always hold
assert isinstance(result, dict)
assert "success" in result
assert isinstance(result["success"], bool)
if result["success"]:
# Successful results should have required fields
assert "session_name" in result
assert "security_level" in result
assert "execution_duration_ms" in result
assert result["security_level"] == security_level
# Session name should be sanitized
assert len(result["session_name"]) <= 100
assert result["session_name"].strip() != ""
else:
# Failed results should have error information
assert "error" in result
assert "error_type" in result
assert isinstance(result["error"], str)
assert len(result["error"]) > 0
