Function test_execute_tools_parallel_max_concurrency

tests/tools/test_execution.py:183–213

Test max_concurrency limits parallel execution.

()

@pytest.mark.asyncio
async def test_execute_tools_parallel_max_concurrency():
    """Test max_concurrency limits parallel execution."""
    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(10)
    ]

    await execute_tools_parallel(manager, calls, max_concurrency=2)

    # Max concurrent should not exceed 2 (though timing may vary)
    assert max_concurrent <= 3  # Allow some slack due to async timing


@pytest.mark.asyncio