""" Integration tests for pipeline mutation commands via WebSocket/UI panel. Tests the mutation API through the command interface. """ from unittest.mock import Mock from engine.app.pipeline_runner import _handle_pipeline_mutation from engine.pipeline import Pipeline from engine.pipeline.ui import UIConfig, UIPanel class TestPipelineMutationCommands: """Test pipeline mutation commands through the mutation API.""" def test_can_hot_swap_existing_stage(self): """Test can_hot_swap returns True for existing, non-critical stage.""" pipeline = Pipeline() # Add a test stage mock_stage = Mock() mock_stage.capabilities = {"test_capability"} pipeline.add_stage("test_stage", mock_stage) pipeline._capability_map = {"test_capability": ["test_stage"]} # Test that we can check hot-swap capability result = pipeline.can_hot_swap("test_stage") assert result is True def test_can_hot_swap_nonexistent_stage(self): """Test can_hot_swap returns False for non-existent stage.""" pipeline = Pipeline() result = pipeline.can_hot_swap("nonexistent_stage") assert result is False def test_can_hot_swap_minimum_capability(self): """Test can_hot_swap with minimum capability stage.""" pipeline = Pipeline() # Add a source stage (minimum capability) mock_stage = Mock() mock_stage.capabilities = {"source"} pipeline.add_stage("source", mock_stage) pipeline._capability_map = {"source": ["source"]} # Initialize pipeline to trigger capability validation pipeline._initialized = True # Source is the only provider of minimum capability result = pipeline.can_hot_swap("source") # Should be False because it's the sole provider of a minimum capability assert result is False def test_cleanup_stage(self): """Test cleanup_stage calls cleanup on specific stage.""" pipeline = Pipeline() # Add a stage with a mock cleanup method mock_stage = Mock() pipeline.add_stage("test_stage", mock_stage) # Cleanup the specific stage pipeline.cleanup_stage("test_stage") # Verify cleanup was called mock_stage.cleanup.assert_called_once() def test_cleanup_stage_nonexistent(self): """Test cleanup_stage on non-existent stage doesn't crash.""" pipeline = Pipeline() pipeline.cleanup_stage("nonexistent_stage") # Should not raise an exception def test_remove_stage_rebuilds_execution_order(self): """Test that remove_stage rebuilds execution order.""" pipeline = Pipeline() # Add two independent stages stage1 = Mock() stage1.capabilities = {"source"} stage1.dependencies = set() stage1.stage_dependencies = [] # Add empty list for stage dependencies stage2 = Mock() stage2.capabilities = {"render.output"} stage2.dependencies = set() # No dependencies stage2.stage_dependencies = [] # No stage dependencies pipeline.add_stage("stage1", stage1) pipeline.add_stage("stage2", stage2) # Build pipeline to establish execution order pipeline._initialized = True pipeline._capability_map = {"source": ["stage1"], "render.output": ["stage2"]} pipeline._execution_order = ["stage1", "stage2"] # Remove stage1 pipeline.remove_stage("stage1") # Verify execution order was rebuilt assert "stage1" not in pipeline._execution_order assert "stage2" in pipeline._execution_order def test_handle_pipeline_mutation_remove_stage(self): """Test _handle_pipeline_mutation with remove_stage command.""" pipeline = Pipeline() # Add a mock stage mock_stage = Mock() pipeline.add_stage("test_stage", mock_stage) # Create remove command command = {"action": "remove_stage", "stage": "test_stage"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled and stage was removed assert result is True assert "test_stage" not in pipeline._stages def test_handle_pipeline_mutation_swap_stages(self): """Test _handle_pipeline_mutation with swap_stages command.""" pipeline = Pipeline() # Add two mock stages stage1 = Mock() stage2 = Mock() pipeline.add_stage("stage1", stage1) pipeline.add_stage("stage2", stage2) # Create swap command command = {"action": "swap_stages", "stage1": "stage1", "stage2": "stage2"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled assert result is True def test_handle_pipeline_mutation_enable_stage(self): """Test _handle_pipeline_mutation with enable_stage command.""" pipeline = Pipeline() # Add a mock stage with set_enabled method mock_stage = Mock() mock_stage.set_enabled = Mock() pipeline.add_stage("test_stage", mock_stage) # Create enable command command = {"action": "enable_stage", "stage": "test_stage"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled assert result is True mock_stage.set_enabled.assert_called_once_with(True) def test_handle_pipeline_mutation_disable_stage(self): """Test _handle_pipeline_mutation with disable_stage command.""" pipeline = Pipeline() # Add a mock stage with set_enabled method mock_stage = Mock() mock_stage.set_enabled = Mock() pipeline.add_stage("test_stage", mock_stage) # Create disable command command = {"action": "disable_stage", "stage": "test_stage"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled assert result is True mock_stage.set_enabled.assert_called_once_with(False) def test_handle_pipeline_mutation_cleanup_stage(self): """Test _handle_pipeline_mutation with cleanup_stage command.""" pipeline = Pipeline() # Add a mock stage mock_stage = Mock() pipeline.add_stage("test_stage", mock_stage) # Create cleanup command command = {"action": "cleanup_stage", "stage": "test_stage"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled and cleanup was called assert result is True mock_stage.cleanup.assert_called_once() def test_handle_pipeline_mutation_can_hot_swap(self): """Test _handle_pipeline_mutation with can_hot_swap command.""" pipeline = Pipeline() # Add a mock stage mock_stage = Mock() mock_stage.capabilities = {"test"} pipeline.add_stage("test_stage", mock_stage) pipeline._capability_map = {"test": ["test_stage"]} # Create can_hot_swap command command = {"action": "can_hot_swap", "stage": "test_stage"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled assert result is True def test_handle_pipeline_mutation_move_stage(self): """Test _handle_pipeline_mutation with move_stage command.""" pipeline = Pipeline() # Add two mock stages stage1 = Mock() stage2 = Mock() pipeline.add_stage("stage1", stage1) pipeline.add_stage("stage2", stage2) # Initialize execution order pipeline._execution_order = ["stage1", "stage2"] # Create move command to move stage1 after stage2 command = {"action": "move_stage", "stage": "stage1", "after": "stage2"} # Handle the mutation result = _handle_pipeline_mutation(pipeline, command) # Verify it was handled (result might be True or False depending on validation) # The key is that the command was processed assert result in (True, False) def test_ui_panel_execute_command_mutation_actions(self): """Test UI panel execute_command with mutation actions.""" ui_panel = UIPanel(UIConfig()) # Test that mutation actions return False (not handled by UI panel) # These should be handled by the WebSocket command handler instead mutation_actions = [ {"action": "remove_stage", "stage": "test"}, {"action": "swap_stages", "stage1": "a", "stage2": "b"}, {"action": "enable_stage", "stage": "test"}, {"action": "disable_stage", "stage": "test"}, {"action": "cleanup_stage", "stage": "test"}, {"action": "can_hot_swap", "stage": "test"}, ] for command in mutation_actions: result = ui_panel.execute_command(command) assert result is False, ( f"Mutation action {command['action']} should not be handled by UI panel" )