from unittest.mock import MagicMock
import pytest
from boring.core.kernel import BoringKernel
from boring.flow.exceptions import FlowError
from boring.flow.graph import FlowGraph
from boring.flow.nodes.base import BaseNode, NodeResult, NodeResultStatus
class ExplodingNode(BaseNode):
async def process(self, context) -> NodeResult:
raise FlowError("CRITICAL FAILURE SIMULATION")
class FailingNode(BaseNode):
async def process(self, context) -> NodeResult:
return NodeResult(status=NodeResultStatus.FAILURE, message="Soft Fail")
@pytest.mark.asyncio
async def test_flow_error_halts_execution(tmp_path):
"""VERIFY: FlowError stops the graph immediately."""
kernel = BoringKernel(tmp_path)
ctx = kernel.create_context("Kill Switch Test")
graph = FlowGraph(ctx)
graph.add_node(ExplodingNode("Exploder"), is_start=True)
# Add a node that should NEVER be reached
mock_node = MagicMock(spec=BaseNode)
mock_node.name = "Unreachable"
graph.add_node(mock_node)
# We should catch the error or result implies error
# Current implementation captures FlowError and returns error message
result = await graph.run()
assert "CRITICAL FAILURE SIMULATION" in result.message
assert "Critical Flow Error" in result.message # Corrected string
# Prove the second node was NOT called (Kill Switch worked)
mock_node.process.assert_not_called()
def test_soft_fail_logs_but_continues_if_allowed(tmp_path):
"""VERIFY: Non-critical failures handled gracefully."""
# Current logic: If execute returns FAILURE, does it stop?
# FlowGraph logic: if result.status != SUCCESS -> stop?
pass # Wait, let's check FlowGraph logic.
