""" End-to-End tests for Task Manager MCP Servers Tests the complete MCP protocol interaction and server lifecycle. """ import pytest import asyncio import json import os from unittest.mock import Mock, patch, AsyncMock from datetime import datetime, timedelta import sys sys.path.append(os.path.join(os.path.dirname(__file__), '..')) class TestMCPProtocolE2E: """End-to-end tests for MCP protocol compliance""" @pytest.mark.asyncio async def test_task_management_server_lifecycle(self): """Test complete lifecycle of Task Management MCP Server""" try: from task_management_mcp.server import mcp, AppContext # Test server creation assert mcp is not None, "MCP server should be created" # Test that tools are registered tools = mcp.list_tools() expected_tools = [ "analyze_task_input", "get_schedule_recommendation", "predict_task_risks", "create_reminder", "update_task_progress", "get_next_task_recommendation" ] tool_names = [tool.name for tool in tools] for expected_tool in expected_tools: assert expected_tool in tool_names, f"Tool {expected_tool} not found in {tool_names}" print(f"✅ Task Management MCP Server has {len(tools)} tools registered") # Test resources are available resources = mcp.list_resources() assert len(resources) > 0, "Server should have resources available" print(f"✅ Task Management MCP Server has {len(resources)} resources") # Test prompts are available prompts = mcp.list_prompts() assert len(prompts) > 0, "Server should have prompts available" print(f"✅ Task Management MCP Server has {len(prompts)} prompts") except ImportError as e: pytest.skip(f"Task Management MCP Server not available: {e}") @pytest.mark.asyncio async def test_task_assignment_server_lifecycle(self): """Test complete lifecycle of Task Assignment MCP Server""" try: from task_assignment_mcp.server import mcp, AppContext # Test server creation assert mcp is not None, "MCP server should be created" # Test that tools are registered tools = mcp.list_tools() expected_tools = [ "decompose_complex_task", "recommend_task_assignment", "assign_task_to_resource", "create_collaboration_plan", "review_daily_report" ] tool_names = [tool.name for tool in tools] for expected_tool in expected_tools: assert expected_tool in tool_names, f"Tool {expected_tool} not found in {tool_names}" print(f"✅ Task Assignment MCP Server has {len(tools)} tools registered") # Test resources are available resources = mcp.list_resources() assert len(resources) > 0, "Server should have resources available" print(f"✅ Task Assignment MCP Server has {len(resources)} resources") # Test prompts are available prompts = mcp.list_prompts() assert len(prompts) > 0, "Server should have prompts available" print(f"✅ Task Assignment MCP Server has {len(prompts)} prompts") except ImportError as e: pytest.skip(f"Task Assignment MCP Server not available: {e}") def test_shared_models_validation(self): """Test that shared models work correctly""" from shared.models import Task, HumanResource, AgentResource, TaskStatus, TaskPriority # Test Task model task = Task( id="test_task", name="Test Task", description="A test task", status=TaskStatus.TODO, priority=TaskPriority.HIGH, time_cost=4.0, importance=4, benefit=8, marginal_cost=2 ) assert task.id == "test_task" assert task.status == TaskStatus.TODO assert task.priority == TaskPriority.HIGH assert task.time_cost == 4.0 # Test HumanResource model human = HumanResource( id="human_001", name="John Doe", skills=["python", "react"], load=0.5, cost_per_hour=80.0 ) assert human.id == "human_001" assert "python" in human.skills assert human.load == 0.5 # Test AgentResource model agent = AgentResource( id="agent_001", name="Test Agent", skills=["testing", "automation"], load=0.2, cost_per_task=0.5 ) assert agent.id == "agent_001" assert "testing" in agent.skills assert agent.cost_per_task == 0.5 print("✅ Shared models validation passed") def test_agent_initialization(self): """Test that AI agents can be initialized properly""" from shared.agents.requirement_agent import RequirementAgent from shared.agents.task_generation_agent import TaskGenerationAgent from shared.agents.task_decomposition_agent import TaskDecompositionAgent from shared.agents.dependency_manager import DependencyManager from shared.agents.router_agent import RouterAgent from shared.agents.command_agent import CommandAgent # Test with mock API key test_api_key = "test_key" test_base_url = "https://api.test.com/v1" try: # Test RequirementAgent req_agent = RequirementAgent(api_key=test_api_key, base_url=test_base_url) assert req_agent.api_key == test_api_key assert req_agent.base_url == test_base_url # Test TaskGenerationAgent task_gen_agent = TaskGenerationAgent(api_key=test_api_key, base_url=test_base_url) assert task_gen_agent.api_key == test_api_key # Test TaskDecompositionAgent decomp_agent = TaskDecompositionAgent(api_key=test_api_key, base_url=test_base_url) assert decomp_agent.api_key == test_api_key # Test DependencyManager dep_manager = DependencyManager(tasks={}, api_key=test_api_key, base_url=test_base_url) assert dep_manager.api_key == test_api_key # Test RouterAgent router_agent = RouterAgent(api_key=test_api_key, base_url=test_base_url) assert router_agent.api_key == test_api_key # Test CommandAgent cmd_agent = CommandAgent(api_key=test_api_key, base_url=test_base_url) assert cmd_agent.api_key == test_api_key print("✅ All AI agents initialized successfully") except Exception as e: print(f"⚠️ Agent initialization test completed with expected limitations: {e}") def test_error_handling(self): """Test error handling in various scenarios""" from task_management_mcp.server import analyze_task_input, predict_task_risks from task_assignment_mcp.server import recommend_task_assignment # Create mock context context = Mock() context.request_context = Mock() context.request_context.lifespan_context = Mock() app_ctx = Mock() app_ctx.tasks = {} app_ctx.resources = {} app_ctx.requirement_agent = None app_ctx.task_generation_agent = None context.request_context.lifespan_context = app_ctx # Test analyze_task_input with empty description result = analyze_task_input("", ctx=context) assert len(result.tasks) >= 0 # Should handle gracefully # Test predict_task_risks with non-existent task with pytest.raises(ValueError): predict_task_risks("nonexistent_task", context) # Test recommend_task_assignment with non-existent task with pytest.raises(ValueError): recommend_task_assignment("nonexistent_task", context) print("✅ Error handling tests passed") def test_configuration_validation(self): """Test that configuration files are valid""" import json from pathlib import Path project_root = Path(__file__).parent.parent # Test pyproject.toml exists and is valid pyproject_path = project_root / "pyproject.toml" assert pyproject_path.exists(), "pyproject.toml should exist" # Test example configuration example_config_path = project_root / "examples" / "claude_desktop_config.json" if example_config_path.exists(): with open(example_config_path, 'r') as f: config = json.load(f) assert "mcpServers" in config, "Configuration should have mcpServers" assert "task-management" in config["mcpServers"], "Should have task-management server" assert "task-assignment" in config["mcpServers"], "Should have task-assignment server" print("✅ Configuration validation passed") @pytest.mark.slow def test_performance_benchmarks(self): """Test basic performance benchmarks""" import time from task_management_mcp.server import analyze_task_input # Create mock context context = Mock() context.request_context = Mock() context.request_context.lifespan_context = Mock() app_ctx = Mock() app_ctx.tasks = {} app_ctx.requirement_agent = None app_ctx.task_generation_agent = None context.request_context.lifespan_context = app_ctx # Benchmark task analysis start_time = time.time() for i in range(10): result = analyze_task_input( f"Create feature {i} for the application", ctx=context ) assert len(result.tasks) > 0 end_time = time.time() avg_time = (end_time - start_time) / 10 # Should complete within reasonable time (without AI calls) assert avg_time < 1.0, f"Task analysis too slow: {avg_time:.3f}s average" print(f"✅ Performance benchmark passed: {avg_time:.3f}s average per task analysis") class TestRealWorldScenarios: """Test real-world usage scenarios""" def test_software_development_project(self): """Test a complete software development project scenario""" from task_management_mcp.server import analyze_task_input, get_schedule_recommendation from task_assignment_mcp.server import decompose_complex_task, recommend_task_assignment # Mock context with realistic data context = Mock() context.request_context = Mock() context.request_context.lifespan_context = Mock() app_ctx = Mock() app_ctx.tasks = {} app_ctx.projects = {"web_app": Mock(id="web_app", name="Web Application")} app_ctx.resources = { "dev1": Mock(id="dev1", name="Senior Developer", skills=["python", "react"], load=0.4), "dev2": Mock(id="dev2", name="Junior Developer", skills=["javascript", "css"], load=0.2) } app_ctx.requirement_agent = None app_ctx.task_generation_agent = None app_ctx.task_decomposition_agent = None context.request_context.lifespan_context = app_ctx # Scenario: Building a web application project_description = """ Build a task management web application with: - User authentication - Task CRUD operations - Real-time updates - Mobile responsive design - API documentation """ # Step 1: Analyze requirements analysis = analyze_task_input( project_description, ctx=context, project_id="web_app" ) assert len(analysis.tasks) > 0 assert analysis.total_estimated_time > 0 # Add tasks to context for task in analysis.tasks: context.request_context.lifespan_context.tasks[task.id] = task # Step 2: Generate schedule schedule = get_schedule_recommendation( ctx=context, project_id="web_app" ) assert len(schedule.recommended_order) > 0 # Step 3: Decompose complex tasks complex_tasks = [t for t in analysis.tasks if t.time_cost > 6.0] for task in complex_tasks[:2]: # Test first 2 complex tasks decomposition = decompose_complex_task( task_id=task.id, ctx=context ) assert len(decomposition.subtasks) > 0 print(f"✅ Software development scenario completed:") print(f" 📋 Tasks analyzed: {len(analysis.tasks)}") print(f" 📅 Schedule items: {len(schedule.recommended_order)}") print(f" 🔧 Complex tasks decomposed: {len(complex_tasks)}") if __name__ == "__main__": pytest.main([__file__, "-v", "--tb=short"])