(
self, tool_name, arguments, namespace=None, timeout=None
)
| 188 | original_execute = MockToolManager.execute_tool |
| 189 | |
| 190 | async def tracking_execute( |
| 191 | self, tool_name, arguments, namespace=None, timeout=None |
| 192 | ): |
| 193 | nonlocal execution_count, max_concurrent |
| 194 | execution_count += 1 |
| 195 | current = execution_count |
| 196 | max_concurrent = max(max_concurrent, current) |
| 197 | await asyncio.sleep(0.01) # Simulate work |
| 198 | execution_count -= 1 |
| 199 | return await original_execute(self, tool_name, arguments, namespace, timeout) |
| 200 | |
| 201 | manager = MockToolManager() |
| 202 | manager.execute_tool = lambda *args, **kwargs: tracking_execute( |
no outgoing calls
no test coverage detected