Katamari MCP Server

""" Comprehensive tests for Blue/Green deployment system. """ import pytest import asyncio from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock, patch # Note: These imports may fail due to module resolution issues # The tests demonstrate the intended functionality try: from katamari_mcp.acp.bluegreen import ( BlueGreenDeployer, DeploymentInstance, ComponentVersion, DeploymentStatus, InstanceType, StateSyncOperation ) from katamari_mcp.acp.heuristics import HeuristicProfile, DataExposure, Complexity from katamari_mcp.acp.sandbox import CapabilitySandbox IMPORTS_AVAILABLE = True except ImportError: IMPORTS_AVAILABLE = False class TestComponentVersion: """Test ComponentVersion dataclass.""" def test_component_version_creation(self): """Test creating a component version.""" version = ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123", capabilities=["test_func"], dependencies=["json"] ) assert version.version == "v1.0.0" assert version.git_commit == "abc123" assert len(version.capabilities) == 1 def test_component_version_serialization(self): """Test component version serialization.""" version = ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ) data = version.to_dict() assert data['version'] == "v1.0.0" assert data['git_commit'] == "abc123" assert 'timestamp' in data # Test deserialization restored = ComponentVersion.from_dict(data) assert restored.version == version.version assert restored.git_commit == version.git_commit class TestDeploymentInstance: """Test DeploymentInstance dataclass.""" def test_deployment_instance_creation(self): """Test creating a deployment instance.""" version = ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ) sandbox = CapabilitySandbox() heuristics = HeuristicProfile.default() instance = DeploymentInstance( instance_id="test-instance-1", component_name="test_component", version=version, instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=sandbox, heuristics=heuristics, created_at=datetime.now(timezone.utc) ) assert instance.instance_id == "test-instance-1" assert instance.component_name == "test_component" assert instance.instance_type == InstanceType.BLUE assert instance.status == DeploymentStatus.ACTIVE assert instance.traffic_percentage == 0.0 # Default value def test_deployment_instance_serialization(self): """Test deployment instance serialization.""" version = ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ) sandbox = CapabilitySandbox() heuristics = HeuristicProfile.default() instance = DeploymentInstance( instance_id="test-instance-1", component_name="test_component", version=version, instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=sandbox, heuristics=heuristics, created_at=datetime.now(timezone.utc), traffic_percentage=100.0 ) data = instance.to_dict() assert data['instance_id'] == "test-instance-1" assert data['component_name'] == "test_component" assert data['instance_type'] == "blue" assert data['status'] == "active" assert data['traffic_percentage'] == 100.0 class TestBlueGreenDeployer: """Test BlueGreenDeployer class.""" @pytest.fixture def deployer(self): """Create a BlueGreenDeployer instance for testing.""" with patch('katamari_mcp.acp.bluegreen.Config') as mock_config: mock_config.return_value.get.return_value = '/tmp/test_deployments' deployer = BlueGreenDeployer() return deployer @pytest.fixture def mock_git_tracker(self): """Mock GitTracker for testing.""" with patch('katamari_mcp.acp.bluegreen.GitTracker') as mock: tracker = mock.return_value tracker.get_current_commit_info.return_value = { 'commit': 'abc123', 'author': 'test@example.com', 'message': 'Test commit', 'branch': 'main' } tracker.get_commit_info.return_value = { 'commit': 'abc123', 'author': 'test@example.com', 'message': 'Test commit', 'branch': 'main' } tracker.get_file_content_at_commit.return_value = 'def test(): return "test"' return tracker @pytest.mark.asyncio async def test_initial_deployment(self, deployer, mock_git_tracker): """Test initial component deployment (blue instance).""" component_code = ''' def test_function(): return {"status": "ok", "message": "test"} ''' with patch.object(deployer, '_extract_capabilities', return_value=['test_function']): instance_id = await deployer.deploy_component("test_component", component_code) assert instance_id is not None assert "test_component-blue-" in instance_id # Check deployment was created assert instance_id in deployer.deployments instance = deployer.deployments[instance_id] assert instance.component_name == "test_component" assert instance.instance_type == InstanceType.BLUE assert instance.status == DeploymentStatus.ACTIVE assert instance.traffic_percentage == 100.0 # Check routing was set up assert "test_component" in deployer.traffic_router assert instance_id in deployer.traffic_router["test_component"] assert deployer.traffic_router["test_component"][instance_id] == 100.0 @pytest.mark.asyncio async def test_blue_green_deployment(self, deployer, mock_git_tracker): """Test blue/green deployment with parallel instances.""" # First deployment (blue) blue_code = ''' def test_function(): return {"version": "1.0", "status": "ok"} ''' with patch.object(deployer, '_extract_capabilities', return_value=['test_function']): blue_instance_id = await deployer.deploy_component("test_component", blue_code) # Second deployment (green) green_code = ''' def test_function(): return {"version": "2.0", "status": "ok"} ''' with patch.object(deployer, '_extract_capabilities', return_value=['test_function']): with patch.object(deployer, '_verify_green_deployment') as mock_verify: mock_verify.return_value = None green_instance_id = await deployer.deploy_component("test_component", green_code) assert green_instance_id is not None assert "test_component-green-" in green_instance_id # Check both instances exist assert blue_instance_id in deployer.deployments assert green_instance_id in deployer.deployments blue_instance = deployer.deployments[blue_instance_id] green_instance = deployer.deployments[green_instance_id] # Blue should be standby, green should be active assert blue_instance.status == DeploymentStatus.STANDBY assert green_instance.status == DeploymentStatus.VERIFYING # During verification # Traffic should still be on blue initially assert blue_instance.traffic_percentage == 100.0 assert green_instance.traffic_percentage == 0.0 @pytest.mark.asyncio async def test_state_synchronization(self, deployer, mock_git_tracker): """Test state synchronization between instances.""" # Create blue instance with state blue_instance = DeploymentInstance( instance_id="blue-instance", component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), state_data={ 'counter': 42, 'config': {'setting1': 'value1'}, 'external_data': 'should_not_sync' } ) # Create green instance green_instance = DeploymentInstance( instance_id="green-instance", component_name="test_component", version=ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit 2", file_hash="hash456" ), instance_type=InstanceType.GREEN, status=DeploymentStatus.DEPLOYING, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), state_data={} ) deployer.deployments["blue-instance"] = blue_instance deployer.deployments["green-instance"] = green_instance # Sync state await deployer._sync_state_to_green(blue_instance, green_instance) # Check state was synced (filtered by compatibility) assert 'counter' in green_instance.state_data assert 'config' in green_instance.state_data assert green_instance.state_data['counter'] == 42 # Check sync operation was recorded assert len(deployer.sync_operations) == 1 sync_op = deployer.sync_operations[0] assert sync_op.source_instance == "blue-instance" assert sync_op.target_instance == "green-instance" assert sync_op.status == "completed" @pytest.mark.asyncio async def test_gradual_traffic_switch(self, deployer): """Test gradual traffic switching.""" # Setup instances blue_instance_id = "blue-instance" green_instance_id = "green-instance" deployer.deployments[blue_instance_id] = DeploymentInstance( instance_id=blue_instance_id, component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=100.0 ) deployer.deployments[green_instance_id] = DeploymentInstance( instance_id=green_instance_id, component_name="test_component", version=ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit 2", file_hash="hash456" ), instance_type=InstanceType.GREEN, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=0.0 ) # Setup initial routing deployer.traffic_router["test_component"] = { blue_instance_id: 100.0, green_instance_id: 0.0 } # Test gradual switch with short duration for testing with patch('asyncio.sleep'): # Speed up the test await deployer.switch_traffic( "test_component", green_instance_id, gradual=True, duration_seconds=1 ) # Check final routing final_routing = deployer.traffic_router["test_component"] assert final_routing[green_instance_id] == 100.0 assert final_routing[blue_instance_id] == 0.0 # Check instance types were updated blue_instance = deployer.deployments[blue_instance_id] green_instance = deployer.deployments[green_instance_id] assert blue_instance.instance_type == InstanceType.OBSOLETE assert green_instance.instance_type == InstanceType.BLUE @pytest.mark.asyncio async def test_immediate_traffic_switch(self, deployer): """Test immediate traffic switching.""" # Setup instances blue_instance_id = "blue-instance" green_instance_id = "green-instance" deployer.deployments[blue_instance_id] = DeploymentInstance( instance_id=blue_instance_id, component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=100.0 ) deployer.deployments[green_instance_id] = DeploymentInstance( instance_id=green_instance_id, component_name="test_component", version=ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit 2", file_hash="hash456" ), instance_type=InstanceType.GREEN, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=0.0 ) # Setup initial routing deployer.traffic_router["test_component"] = { blue_instance_id: 100.0, green_instance_id: 0.0 } # Test immediate switch await deployer.switch_traffic( "test_component", green_instance_id, gradual=False ) # Check routing was switched immediately final_routing = deployer.traffic_router["test_component"] assert final_routing[green_instance_id] == 100.0 assert blue_instance_id not in final_routing # Check instance traffic percentages blue_instance = deployer.deployments[blue_instance_id] green_instance = deployer.deployments[green_instance_id] assert blue_instance.traffic_percentage == 0.0 assert green_instance.traffic_percentage == 100.0 @pytest.mark.asyncio async def test_rollback_deployment(self, deployer, mock_git_tracker): """Test deployment rollback.""" # Setup component history v1 = ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Version 1.0", file_hash="hash123" ) v2 = ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Version 2.0", file_hash="hash456" ) deployer.component_history["test_component"] = [v1, v2] # Mock deployment for rollback with patch.object(deployer, 'deploy_component') as mock_deploy: mock_deploy.return_value = "rollback-instance" with patch.object(deployer, 'switch_traffic') as mock_switch: rollback_instance_id = await deployer.rollback_deployment("test_component") # Check rollback was created mock_deploy.assert_called_once() args = mock_deploy.call_args[0] assert args[0] == "test_component" # component_name assert args[2] == "abc123" # git_commit (v1) # Check traffic was switched mock_switch.assert_called_once_with( "test_component", "rollback-instance", gradual=False ) assert rollback_instance_id == "rollback-instance" @pytest.mark.asyncio async def test_health_monitoring(self, deployer): """Test health monitoring functionality.""" # Create test instance instance = DeploymentInstance( instance_id="test-instance", component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc) ) deployer.deployments["test-instance"] = instance # Mock health check with patch.object(deployer, '_perform_health_check', return_value=0.8): await deployer._health_monitor_loop() # Run one iteration # Check health was updated assert instance.health_score == 0.8 assert instance.last_health_check is not None assert instance.metrics['last_health_score'] == 0.8 @pytest.mark.asyncio async def test_cleanup_deprecated_instances(self, deployer): """Test cleanup of deprecated instances.""" # Create old deprecated instance old_time = datetime.now(timezone.utc).timestamp() - (25 * 3600) # 25 hours ago old_instance = DeploymentInstance( instance_id="old-instance", component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.fromtimestamp(old_time, timezone.utc), author="test@example.com", message="Old commit", file_hash="hash123" ), instance_type=InstanceType.OBSOLETE, status=DeploymentStatus.DEPRECATED, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.fromtimestamp(old_time, timezone.utc) ) # Create recent instance recent_instance = DeploymentInstance( instance_id="recent-instance", component_name="test_component", version=ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Recent commit", file_hash="hash456" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc) ) deployer.deployments["old-instance"] = old_instance deployer.deployments["recent-instance"] = recent_instance # Cleanup with 24 hour threshold cleaned_count = await deployer.cleanup_deprecated_instances(max_age_hours=24) # Check old instance was cleaned up assert cleaned_count == 1 assert "old-instance" not in deployer.deployments assert "recent-instance" in deployer.deployments def test_get_deployment_status(self, deployer): """Test getting deployment status.""" # Create test instances blue_instance = DeploymentInstance( instance_id="blue-instance", component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=100.0 ) green_instance = DeploymentInstance( instance_id="green-instance", component_name="test_component", version=ComponentVersion( version="v2.0.0", git_commit="def456", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit 2", file_hash="hash456" ), instance_type=InstanceType.GREEN, status=DeploymentStatus.VERIFYING, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc), traffic_percentage=0.0 ) deployer.deployments["blue-instance"] = blue_instance deployer.deployments["green-instance"] = green_instance deployer.traffic_router["test_component"] = { "blue-instance": 100.0, "green-instance": 0.0 } # Get status status = await deployer.get_deployment_status("test_component") assert status['component_name'] == "test_component" assert status['total_instances'] == 2 assert status['active_instances'] == 1 # Only blue is active assert len(status['instances']) == 2 assert status['traffic_routing']['blue-instance'] == 100.0 assert status['traffic_routing']['green-instance'] == 0.0 def test_extract_capabilities(self, deployer): """Test capability extraction from code.""" code = ''' def process_data(data): return len(data) class DataProcessor: def __init__(self): self.count = 0 def process(self, items): self.count += len(items) return self.count def helper_function(): pass ''' capabilities = asyncio.run(deployer._extract_capabilities(code)) # Should extract function and class names assert 'process_data' in capabilities assert 'DataProcessor' in capabilities assert 'helper_function' in capabilities @pytest.mark.asyncio async def test_state_compatibility_filtering(self, deployer): """Test state compatibility filtering based on heuristics.""" # Create heuristics with no data exposure heuristics = HeuristicProfile.default() heuristics.tags.data_exposure = DataExposure.NONE # Test state data state = { 'internal_counter': 42, 'config_data': {'setting': 'value'}, 'external_api_key': 'secret123', 'public_stats': {'visits': 100} } # Filter state compatible_state = await deployer._filter_compatible_state( state, heuristics, heuristics ) # Should filter out external/public data assert 'internal_counter' in compatible_state assert 'config_data' in compatible_state assert 'external_api_key' not in compatible_state assert 'public_stats' not in compatible_state @pytest.mark.asyncio async def test_shutdown(self, deployer): """Test clean shutdown of deployer.""" # Start health monitoring deployer.health_monitor_task = asyncio.create_task( deployer._health_monitor_loop() ) # Add some deployments instance = DeploymentInstance( instance_id="test-instance", component_name="test_component", version=ComponentVersion( version="v1.0.0", git_commit="abc123", git_branch="main", timestamp=datetime.now(timezone.utc), author="test@example.com", message="Test commit", file_hash="hash123" ), instance_type=InstanceType.BLUE, status=DeploymentStatus.ACTIVE, sandbox=CapabilitySandbox(), heuristics=HeuristicProfile.default(), created_at=datetime.now(timezone.utc) ) deployer.deployments["test-instance"] = instance # Shutdown await deployer.shutdown() # Check monitoring was stopped assert deployer.monitoring_enabled is False assert deployer.health_monitor_task.cancelled() class TestStateSyncOperation: """Test StateSyncOperation dataclass.""" def test_state_sync_operation_creation(self): """Test creating a state sync operation.""" sync_op = StateSyncOperation( source_instance="blue-instance", target_instance="green-instance", sync_type="full", timestamp=datetime.now(timezone.utc) ) assert sync_op.source_instance == "blue-instance" assert sync_op.target_instance == "green-instance" assert sync_op.sync_type == "full" assert sync_op.status == "pending" assert sync_op.error is None assert len(sync_op.synced_keys) == 0 def test_state_sync_operation_completion(self): """Test completing a state sync operation.""" sync_op = StateSyncOperation( source_instance="blue-instance", target_instance="green-instance", sync_type="full", timestamp=datetime.now(timezone.utc) ) # Mark as completed sync_op.status = "completed" sync_op.synced_keys = ["counter", "config"] assert sync_op.status == "completed" assert len(sync_op.synced_keys) == 2 assert "counter" in sync_op.synced_keys assert "config" in sync_op.synced_keys def test_state_sync_operation_failure(self): """Test failing a state sync operation.""" sync_op = StateSyncOperation( source_instance="blue-instance", target_instance="green-instance", sync_type="full", timestamp=datetime.now(timezone.utc) ) # Mark as failed sync_op.status = "failed" sync_op.error = "Connection timeout" assert sync_op.status == "failed" assert sync_op.error == "Connection timeout" assert len(sync_op.synced_keys) == 0 if __name__ == "__main__": pytest.main([__file__])