""" Integration tests for Task Manager MCP Servers Tests the interaction between both MCP servers and their shared components. """ import pytest import asyncio import os from datetime import datetime, timedelta from unittest.mock import Mock, patch import sys sys.path.append(os.path.join(os.path.dirname(__file__), '..')) from shared.models import Task, HumanResource, AgentResource, TaskStatus, TaskPriority from shared.agents.requirement_agent import RequirementAgent from shared.agents.task_generation_agent import TaskGenerationAgent from shared.agents.task_decomposition_agent import TaskDecompositionAgent class TestMCPIntegration: """Integration tests for both MCP servers working together""" @pytest.fixture def sample_project_data(self): """Create sample project data for testing""" return { "project_id": "integration_test_project", "project_name": "E-commerce Platform", "requirements": """ # E-commerce Platform Requirements ## Vision Build a modern e-commerce platform for small businesses ## Core Features - User authentication and profiles - Product catalog management - Shopping cart and checkout - Payment processing - Order management - Admin dashboard ## Technical Requirements - React frontend - Python backend (FastAPI) - PostgreSQL database - Redis for caching """, "resources": { "frontend_dev": HumanResource( id="frontend_dev", name="Sarah Chen", skills=["react", "javascript", "css", "ui_design"], load=0.2, cost_per_hour=80.0, status="available" ), "backend_dev": HumanResource( id="backend_dev", name="Mike Rodriguez", skills=["python", "fastapi", "postgresql", "api_design"], load=0.4, cost_per_hour=90.0, status="available" ), "fullstack_dev": HumanResource( id="fullstack_dev", name="Alex Kim", skills=["react", "python", "postgresql", "devops"], load=0.6, cost_per_hour=95.0, status="available" ), "qa_agent": AgentResource( id="qa_agent", name="QA Testing Agent", skills=["testing", "automation", "quality_assurance"], load=0.1, cost_per_task=0.3, system_prompt="Focus on comprehensive testing", status="available" ) } } def create_mock_context(self, project_data): """Create mock context with project data""" context = Mock() context.request_context = Mock() context.request_context.lifespan_context = Mock() app_ctx = Mock() app_ctx.tasks = {} app_ctx.projects = {project_data["project_id"]: Mock(id=project_data["project_id"], name=project_data["project_name"])} app_ctx.resources = project_data["resources"] app_ctx.reminders = {} app_ctx.assignments = {} app_ctx.collaboration_plans = {} app_ctx.daily_reports = [] # Mock AI agents (will use fallback implementations) app_ctx.requirement_agent = None app_ctx.task_generation_agent = None app_ctx.task_decomposition_agent = None app_ctx.dependency_manager = None context.request_context.lifespan_context = app_ctx return context def test_end_to_end_project_workflow(self, sample_project_data): """Test complete workflow from requirements to task assignment""" from task_management_mcp.server import analyze_task_input, get_schedule_recommendation, predict_task_risks from task_assignment_mcp.server import decompose_complex_task, recommend_task_assignment, assign_task_to_resource ctx = self.create_mock_context(sample_project_data) # Step 1: Analyze high-level requirements into tasks print("\n=== Step 1: Task Analysis ===") analysis_result = analyze_task_input( task_description="Build complete e-commerce platform with user auth, product catalog, shopping cart, and admin dashboard", ctx=ctx, project_id=sample_project_data["project_id"], due_date=(datetime.now() + timedelta(days=60)).isoformat() ) assert len(analysis_result.tasks) > 0 assert analysis_result.total_estimated_time > 0 # Add tasks to context for task in analysis_result.tasks: ctx.request_context.lifespan_context.tasks[task.id] = task print(f"✅ Generated {len(analysis_result.tasks)} high-level tasks") # Step 2: Decompose complex tasks print("\n=== Step 2: Task Decomposition ===") complex_tasks = [t for t in analysis_result.tasks if t.time_cost > 8.0] all_subtasks = [] for complex_task in complex_tasks: decomposition = decompose_complex_task( task_id=complex_task.id, ctx=ctx, target_time_per_subtask=4.0 ) all_subtasks.extend(decomposition.subtasks) # Add subtasks to context for subtask in decomposition.subtasks: ctx.request_context.lifespan_context.tasks[subtask.id] = subtask print(f"✅ Decomposed {len(complex_tasks)} complex tasks into {len(all_subtasks)} subtasks") # Step 3: Generate schedule recommendations print("\n=== Step 3: Schedule Planning ===") schedule = get_schedule_recommendation( ctx=ctx, project_id=sample_project_data["project_id"], available_hours_per_day=8.0 ) assert len(schedule.recommended_order) > 0 assert schedule.estimated_completion_date > datetime.now() print(f"✅ Generated schedule with {len(schedule.recommended_order)} tasks") print(f"📅 Estimated completion: {schedule.estimated_completion_date.strftime('%Y-%m-%d')}") # Step 4: Risk assessment for critical tasks print("\n=== Step 4: Risk Assessment ===") high_priority_tasks = [ t for t in ctx.request_context.lifespan_context.tasks.values() if t.priority <= TaskPriority.HIGH ] risk_assessments = [] for task in high_priority_tasks[:3]: # Assess top 3 high-priority tasks risk_assessment = predict_task_risks(task.id, ctx) risk_assessments.append(risk_assessment) print(f"✅ Assessed risks for {len(risk_assessments)} critical tasks") # Step 5: Task assignment recommendations print("\n=== Step 5: Task Assignment ===") assignments = [] available_tasks = [ t for t in ctx.request_context.lifespan_context.tasks.values() if not t.assignee and t.time_cost <= 8.0 # Focus on manageable tasks ] for task in available_tasks[:5]: # Assign first 5 tasks assignment_rec = recommend_task_assignment( task_id=task.id, ctx=ctx ) if assignment_rec.recommended_resource_id: assignment_result = assign_task_to_resource( task_id=task.id, resource_id=assignment_rec.recommended_resource_id, ctx=ctx, notes=f"Auto-assigned based on {assignment_rec.confidence_score:.2f} confidence" ) assignments.append(assignment_result) print(f"✅ Completed {len(assignments)} task assignments") # Step 6: Verify workflow results print("\n=== Step 6: Workflow Verification ===") total_tasks = len(ctx.request_context.lifespan_context.tasks) assigned_tasks = len([t for t in ctx.request_context.lifespan_context.tasks.values() if t.assignee]) assert total_tasks > 0, "No tasks were created" assert assigned_tasks > 0, "No tasks were assigned" assert len(risk_assessments) > 0, "No risk assessments were performed" print(f"✅ Workflow completed successfully:") print(f" 📋 Total tasks: {total_tasks}") print(f" 👥 Assigned tasks: {assigned_tasks}") print(f" ⚠️ Risk assessments: {len(risk_assessments)}") print(f" 📊 Schedule items: {len(schedule.recommended_order)}") # Return summary for further testing return { "total_tasks": total_tasks, "assigned_tasks": assigned_tasks, "risk_assessments": len(risk_assessments), "schedule_items": len(schedule.recommended_order), "assignments": assignments } def test_resource_workload_balancing(self, sample_project_data): """Test that task assignments balance workload across resources""" from task_assignment_mcp.server import recommend_task_assignment, assign_task_to_resource ctx = self.create_mock_context(sample_project_data) # Create multiple tasks with different skill requirements tasks = [ Task(id="frontend_task_1", name="Build login component", time_cost=4.0, description="React component for user login"), Task(id="frontend_task_2", name="Build product listing", time_cost=6.0, description="React component for product catalog"), Task(id="backend_task_1", name="User authentication API", time_cost=5.0, description="FastAPI endpoints for user auth"), Task(id="backend_task_2", name="Product management API", time_cost=7.0, description="FastAPI endpoints for product CRUD"), Task(id="fullstack_task_1", name="Payment integration", time_cost=8.0, description="End-to-end payment processing") ] # Add tasks to context for task in tasks: ctx.request_context.lifespan_context.tasks[task.id] = task # Assign all tasks assignments = [] for task in tasks: assignment_rec = recommend_task_assignment(task.id, ctx) if assignment_rec.recommended_resource_id: assignment_result = assign_task_to_resource( task_id=task.id, resource_id=assignment_rec.recommended_resource_id, ctx=ctx ) assignments.append(assignment_result) # Verify workload distribution resource_workloads = {} for resource_id, resource in sample_project_data["resources"].items(): resource_workloads[resource_id] = resource.load print(f"Resource workload after assignments:") for resource_id, workload in resource_workloads.items(): print(f" {resource_id}: {workload:.2f}") # Check that no resource is overloaded (> 0.9) overloaded_resources = [rid for rid, load in resource_workloads.items() if load > 0.9] assert len(overloaded_resources) == 0, f"Resources overloaded: {overloaded_resources}" # Check that assignments were made assert len(assignments) > 0, "No assignments were made" print(f"✅ Workload balancing test passed with {len(assignments)} assignments") def test_skill_matching_accuracy(self, sample_project_data): """Test that tasks are assigned to resources with appropriate skills""" from task_assignment_mcp.server import recommend_task_assignment ctx = self.create_mock_context(sample_project_data) # Create tasks with specific skill requirements test_cases = [ { "task": Task(id="react_task", name="React component", description="Build a React component with modern hooks"), "expected_skills": ["react", "javascript"] }, { "task": Task(id="python_task", name="Python API", description="Build FastAPI endpoints with PostgreSQL"), "expected_skills": ["python", "fastapi", "postgresql"] }, { "task": Task(id="testing_task", name="Automated testing", description="Create comprehensive test suite"), "expected_skills": ["testing", "automation"] } ] skill_match_results = [] for test_case in test_cases: task = test_case["task"] expected_skills = test_case["expected_skills"] ctx.request_context.lifespan_context.tasks[task.id] = task assignment_rec = recommend_task_assignment(task.id, ctx) if assignment_rec.recommended_resource_id: recommended_resource = sample_project_data["resources"][assignment_rec.recommended_resource_id] # Check skill overlap skill_overlap = len(set(expected_skills) & set(recommended_resource.skills)) skill_match_score = skill_overlap / len(expected_skills) skill_match_results.append({ "task_id": task.id, "recommended_resource": assignment_rec.recommended_resource_id, "skill_match_score": skill_match_score, "confidence": assignment_rec.confidence_score }) print(f"Task: {task.name}") print(f" Recommended: {recommended_resource.name}") print(f" Skill match: {skill_match_score:.2f}") print(f" Confidence: {assignment_rec.confidence_score:.2f}") # Verify that most assignments have good skill matches good_matches = [r for r in skill_match_results if r["skill_match_score"] >= 0.5] match_rate = len(good_matches) / len(skill_match_results) if skill_match_results else 0 assert match_rate >= 0.7, f"Skill matching accuracy too low: {match_rate:.2f}" print(f"✅ Skill matching test passed with {match_rate:.2%} accuracy") if __name__ == "__main__": pytest.main([__file__, "-v"])