""" Unit tests for Task Assignment MCP Server Tests the core functionality of the task assignment MCP server including task decomposition, resource matching, assignment recommendations, and team collaboration. """ 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 task_assignment_mcp.server import mcp, TaskDecompositionResult, AssignmentRecommendation, TeamCollaborationPlan from shared.models import Task, HumanResource, AgentResource, TaskStatus, TaskPriority class TestTaskAssignmentMCP: """Test suite for Task Assignment MCP Server""" @pytest.fixture def mock_context(self): """Create a mock context for testing""" context = Mock() context.request_context = Mock() context.request_context.lifespan_context = Mock() # Mock app context with sample data app_ctx = Mock() app_ctx.tasks = {} app_ctx.resources = { "human_001": HumanResource( id="human_001", name="Alice Johnson", skills=["project_management", "requirements_analysis"], load=0.3, cost_per_hour=85.0, status="available" ), "human_002": HumanResource( id="human_002", name="Bob Smith", skills=["python", "backend_development", "api_design"], load=0.7, cost_per_hour=95.0, status="available" ), "agent_001": AgentResource( id="agent_001", name="Code Generation Agent", skills=["code_generation", "unit_testing"], load=0.2, cost_per_task=0.5, system_prompt="Generate clean code", status="available" ) } app_ctx.assignments = {} app_ctx.collaboration_plans = {} app_ctx.daily_reports = [] app_ctx.task_decomposition_agent = None app_ctx.dependency_manager = None context.request_context.lifespan_context = app_ctx return context def test_decompose_complex_task_without_ai(self, mock_context): """Test task decomposition when AI agents are not available""" from task_assignment_mcp.server import decompose_complex_task # Add a complex task complex_task = Task( id="complex_task", name="Build User Management System", description="Complete user management with auth, profiles, and admin", time_cost=20.0, project_id="test_project" ) mock_context.request_context.lifespan_context.tasks = {"complex_task": complex_task} result = decompose_complex_task( task_id="complex_task", ctx=mock_context, target_time_per_subtask=4.0 ) assert isinstance(result, TaskDecompositionResult) assert result.parent_task_id == "complex_task" assert len(result.subtasks) == 5 # 20 hours / 4 hours per subtask assert result.estimated_total_time == 20.0 assert "Simple time-based decomposition" in result.decomposition_reasoning @patch('shared.agents.task_decomposition_agent.TaskDecompositionAgent') def test_decompose_complex_task_with_ai(self, mock_decomp_agent, mock_context): """Test task decomposition with AI agent available""" from task_assignment_mcp.server import decompose_complex_task # Setup mock agent mock_agent_instance = Mock() mock_agent_instance.decompose_task.return_value = [ { "id": "sub_1", "name": "User Authentication", "description": "Implement login/logout functionality", "time_cost": 8.0 }, { "id": "sub_2", "name": "User Profiles", "description": "Create user profile management", "time_cost": 6.0 } ] mock_decomp_agent.return_value = mock_agent_instance # Add task and agent to context complex_task = Task( id="complex_task", name="Build User Management System", time_cost=20.0 ) mock_context.request_context.lifespan_context.tasks = {"complex_task": complex_task} mock_context.request_context.lifespan_context.task_decomposition_agent = mock_agent_instance result = decompose_complex_task( task_id="complex_task", ctx=mock_context ) assert isinstance(result, TaskDecompositionResult) assert len(result.subtasks) == 2 assert result.estimated_total_time == 14.0 assert "AI-powered intelligent decomposition" in result.decomposition_reasoning def test_recommend_task_assignment(self, mock_context): """Test task assignment recommendation""" from task_assignment_mcp.server import recommend_task_assignment # Add a task that matches Bob's skills task = Task( id="backend_task", name="Implement REST API", description="Build backend API with Python and database integration", time_cost=8.0 ) mock_context.request_context.lifespan_context.tasks = {"backend_task": task} result = recommend_task_assignment( task_id="backend_task", ctx=mock_context ) assert isinstance(result, AssignmentRecommendation) assert result.task_id == "backend_task" assert result.recommended_resource_id == "human_002" # Bob has matching skills assert result.resource_type == "human" assert result.confidence_score > 0 assert "python" in result.reasoning.lower() or "backend" in result.reasoning.lower() def test_recommend_task_assignment_prefer_agent(self, mock_context): """Test task assignment recommendation preferring agents""" from task_assignment_mcp.server import recommend_task_assignment # Add a task that could be done by an agent task = Task( id="code_task", name="Generate unit tests", description="Create comprehensive unit tests for the API", time_cost=4.0 ) mock_context.request_context.lifespan_context.tasks = {"code_task": task} result = recommend_task_assignment( task_id="code_task", ctx=mock_context, prefer_human=False ) assert isinstance(result, AssignmentRecommendation) # Should prefer the agent for code generation tasks assert result.recommended_resource_id == "agent_001" assert result.resource_type == "agent" def test_assign_task_to_resource(self, mock_context): """Test actual task assignment to a resource""" from task_assignment_mcp.server import assign_task_to_resource # Add task and ensure resource exists task = Task( id="test_task", name="Test Task", time_cost=4.0, status=TaskStatus.TODO ) mock_context.request_context.lifespan_context.tasks = {"test_task": task} result = assign_task_to_resource( task_id="test_task", resource_id="human_001", ctx=mock_context, notes="Assigned for testing" ) assert result["task_id"] == "test_task" assert result["resource_id"] == "human_001" assert result["resource_name"] == "Alice Johnson" assert result["status"] == "assigned" assert result["notes"] == "Assigned for testing" # Check that task was updated updated_task = mock_context.request_context.lifespan_context.tasks["test_task"] assert updated_task.assignee == "human_001" # Check that resource load was updated resource = mock_context.request_context.lifespan_context.resources["human_001"] assert resource.load > 0.3 # Should have increased from initial 0.3 def test_assign_task_to_nonexistent_resource(self, mock_context): """Test assignment to non-existent resource""" from task_assignment_mcp.server import assign_task_to_resource task = Task(id="test_task", name="Test Task") mock_context.request_context.lifespan_context.tasks = {"test_task": task} with pytest.raises(ValueError, match="Resource nonexistent not found"): assign_task_to_resource( task_id="test_task", resource_id="nonexistent", ctx=mock_context ) def test_create_collaboration_plan(self, mock_context): """Test creation of team collaboration plan""" from task_assignment_mcp.server import create_collaboration_plan # Add a complex task requiring collaboration complex_task = Task( id="complex_task", name="Full Stack Application", description="Build a complete web application with frontend, backend, and testing", time_cost=25.0 ) mock_context.request_context.lifespan_context.tasks = {"complex_task": complex_task} result = create_collaboration_plan( task_id="complex_task", ctx=mock_context, required_skills=["frontend", "backend", "testing"], max_team_size=3 ) assert isinstance(result, TeamCollaborationPlan) assert result.task_id == "complex_task" assert len(result.team_members) > 0 assert len(result.team_members) <= 3 assert result.collaboration_type in ["pair_programming", "parallel_development", "sequential_handoff"] assert len(result.coordination_strategy) > 0 assert len(result.milestone_schedule) > 0 def test_review_daily_report(self, mock_context): """Test daily report review and scoring""" from task_assignment_mcp.server import review_daily_report # Test a high-quality report high_quality_report = """ Today I completed the user authentication module and finished implementing the login/logout functionality. I thoroughly tested all endpoints and documented the API specifications. I collaborated with the frontend team to discuss the integration points and helped review their code. All tests are passing and the code has been optimized for performance. """ result = review_daily_report( team_member_id="human_002", report_content=high_quality_report, ctx=mock_context ) assert result.team_member_id == "human_002" assert result.productivity_score >= 7 # Should be high due to "completed" keywords assert result.quality_score >= 7 # Should be high due to "tested", "documented", "optimized" assert result.collaboration_score >= 6 # Should be good due to "collaborated", "helped" assert result.overall_score >= 7 assert len(result.achievements) > 0 assert len(result.feedback) > 0 def test_review_daily_report_poor_quality(self, mock_context): """Test daily report review with poor quality report""" from task_assignment_mcp.server import review_daily_report # Test a low-quality report poor_report = "Worked on stuff. Had some issues." result = review_daily_report( team_member_id="human_001", report_content=poor_report, ctx=mock_context ) assert result.productivity_score < 6 # Should be low assert result.quality_score < 6 # Should be low assert result.collaboration_score < 5 # Should be low assert result.overall_score < 6 assert len(result.improvement_areas) > 0 def test_get_resource_details(self, mock_context): """Test resource details retrieval""" from task_assignment_mcp.server import get_resource_details # Test human resource details human_details = get_resource_details("human_001", mock_context) assert "Alice Johnson" in human_details assert "Human Resource" in human_details assert "project_management" in human_details assert "$85.00" in human_details # Test agent resource details agent_details = get_resource_details("agent_001", mock_context) assert "Code Generation Agent" in agent_details assert "AI Agent" in agent_details assert "code_generation" in agent_details assert "$0.50" in agent_details def test_get_resource_details_not_found(self, mock_context): """Test resource details for non-existent resource""" from task_assignment_mcp.server import get_resource_details result = get_resource_details("nonexistent", mock_context) assert "not found" in result if __name__ == "__main__": pytest.main([__file__])