File size: 1,446 Bytes
6172a47 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 | import json
from unittest.mock import MagicMock
import pytest
from config.nim import NimSettings
from providers.base import ProviderConfig
from providers.common import ContentBlockManager
from providers.nvidia_nim import NvidiaNimProvider
@pytest.mark.asyncio
async def test_task_tool_interception():
# Setup provider
config = ProviderConfig(api_key="test")
provider = NvidiaNimProvider(config, nim_settings=NimSettings())
# Mock request and sse builder with real ContentBlockManager
request = MagicMock()
request.model = "test-model"
sse = MagicMock()
sse.blocks = ContentBlockManager()
# Tool call data (Task tool)
tc = {
"index": 0,
"id": "tool_123",
"function": {
"name": "Task",
"arguments": json.dumps(
{
"description": "test task",
"prompt": "do something",
"run_in_background": True,
}
),
},
}
# Call the method (consume generator to trigger side effects)
list(provider._process_tool_call(tc, sse))
# Find the emit_tool_delta call and check args
calls = sse.emit_tool_delta.call_args_list
assert len(calls) > 0
args_passed = json.loads(calls[0][0][1])
assert args_passed["run_in_background"] is False
if __name__ == "__main__":
import asyncio
asyncio.run(test_task_tool_interception())
|