Function test_stream_execute_tools_with_error

tests/tools/test_execution.py:321–342

Test streaming handles errors gracefully.

()

@pytest.mark.asyncio
async def test_stream_execute_tools_with_error():
    """Test streaming handles errors gracefully."""
    manager = MockToolManager(
        results={
            "failing_tool": ToolCallResult(
                tool_name="failing_tool",
                success=False,
                error="tool failed",
            )
        }
    )

    calls = [
        CTPToolCall(id="call_1", tool="failing_tool", arguments={}),
        CTPToolCall(id="call_2", tool="success_tool", arguments={}),
    ]

    results = []
    async for result in stream_execute_tools(manager, calls):
        results.append(result)

    assert len(results) == 2


@pytest.mark.asyncio