Streaming: fast tools complete without triggering batch timeout.
()
| 641 | |
| 642 | @pytest.mark.asyncio |
| 643 | async def test_stream_no_timeout_when_fast(): |
| 644 | """Streaming: fast tools complete without triggering batch timeout.""" |
| 645 | manager = SlowMockToolManager(default_delay=0.01) |
| 646 | calls = [ |
| 647 | CTPToolCall(id="call_1", tool="tool_a", arguments={}), |
| 648 | CTPToolCall(id="call_2", tool="tool_b", arguments={}), |
| 649 | CTPToolCall(id="call_3", tool="tool_c", arguments={}), |
| 650 | ] |
| 651 | |
| 652 | results = [] |
| 653 | async for result in stream_execute_tools(manager, calls, batch_timeout=5.0): |
| 654 | results.append(result) |
| 655 | |
| 656 | assert len(results) == 3 |
| 657 | completed_tools = {r.tool for r in results} |
| 658 | assert completed_tools == {"tool_a", "tool_b", "tool_c"} |
| 659 | for r in results: |
| 660 | assert r.is_success |
| 661 | |
| 662 | |
| 663 | # ---------------------------------------------------------------------------- |
nothing calls this directly
no test coverage detected