Class DummyToolManager

tests/chat/test_tool_processor_extended.py:152–205

class DummyToolManager:
    def __init__(self, return_result=None, raise_exception=False):
        self.return_result = return_result or {
            "isError": False,
            "content": "Tool executed successfully",
        }
        self.raise_exception = raise_exception
        self.executed_tool = None
        self.executed_args = None

    async def stream_execute_tools(
        self, calls, timeout=None, on_tool_start=None, max_concurrency=4
    ):
        for call in calls:
            self.executed_tool = call.tool
            self.executed_args = call.arguments

            if on_tool_start:
                await on_tool_start(call)

            now = datetime.now(UTC)
            if self.raise_exception:
                yield CTPToolResult(
                    id=call.id,
                    tool=call.tool,
                    result=None,
                    error="Simulated exception",
                    start_time=now,
                    end_time=now,
                    machine=platform.node(),
                    pid=os.getpid(),
                )
            elif self.return_result.get("isError"):
                yield CTPToolResult(
                    id=call.id,
                    tool=call.tool,
                    result=None,
                    error=self.return_result.get("error", "Error"),
                    start_time=now,
                    end_time=now,
                    machine=platform.node(),
                    pid=os.getpid(),
                )
            else:
                yield CTPToolResult(
                    id=call.id,
                    tool=call.tool,
                    result=self.return_result.get("content"),
                    error=None,
                    start_time=now,
                    end_time=now,
                    machine=platform.node(),
                    pid=os.getpid(),
                )


class CancellingToolManager: