Function test_stream_execute_tools_max_concurrency

tests/tools/test_execution.py:462–495

Test stream_execute_tools respects max_concurrency.

()

@pytest.mark.asyncio
async def test_stream_execute_tools_max_concurrency():
    """Test stream_execute_tools respects max_concurrency."""
    execution_count = 0
    max_concurrent = 0

    original_execute = MockToolManager.execute_tool

    async def tracking_execute(
        self, tool_name, arguments, namespace=None, timeout=None
    ):
        nonlocal execution_count, max_concurrent
        execution_count += 1
        current = execution_count
        max_concurrent = max(max_concurrent, current)
        await asyncio.sleep(0.01)  # Simulate work
        execution_count -= 1
        return await original_execute(self, tool_name, arguments, namespace, timeout)

    manager = MockToolManager()
    manager.execute_tool = lambda *args, **kwargs: tracking_execute(
        manager, *args, **kwargs
    )

    calls = [
        CTPToolCall(id=f"call_{i}", tool=f"tool_{i}", arguments={}) for i in range(5)
    ]

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

    assert len(results) == 5
    # Max concurrent should not exceed 2 (with some slack for async timing)
    assert max_concurrent <= 3


# ----------------------------------------------------------------------------