72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
import asyncio
|
|
import json
|
|
import time
|
|
|
|
from chat_api.tools import CurrentTimeArguments, ToolRegistry, ToolSpec
|
|
|
|
|
|
def test_builtin_tool_definitions() -> None:
|
|
registry = ToolRegistry()
|
|
|
|
definitions = registry.definitions()
|
|
|
|
assert [item["function"]["name"] for item in definitions] == [ # type: ignore[index]
|
|
"get_current_time"
|
|
]
|
|
|
|
|
|
def test_current_time_tool() -> None:
|
|
async def scenario() -> None:
|
|
registry = ToolRegistry()
|
|
current_time = await registry.execute(
|
|
"get_current_time", '{"timezone":"Asia/Shanghai"}'
|
|
)
|
|
|
|
assert current_time.is_error is False
|
|
assert json.loads(current_time.content)["timezone"] == "Asia/Shanghai"
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_tool_argument_and_execution_errors_are_safe() -> None:
|
|
async def scenario() -> None:
|
|
registry = ToolRegistry()
|
|
invalid_json = await registry.execute("get_current_time", "not-json")
|
|
unknown = await registry.execute("missing", "{}")
|
|
invalid_timezone = await registry.execute(
|
|
"get_current_time", '{"timezone":"Not/A_Real_Zone"}'
|
|
)
|
|
|
|
assert invalid_json.is_error is True
|
|
assert "invalid_arguments" in invalid_json.content
|
|
assert unknown.is_error is True
|
|
assert "unknown_tool" in unknown.content
|
|
assert invalid_timezone.is_error is True
|
|
assert "execution_failed" in invalid_timezone.content
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_tool_timeout_returns_error() -> None:
|
|
def slow_handler(arguments):
|
|
time.sleep(0.05)
|
|
return {"timezone": "UTC"}
|
|
|
|
async def scenario() -> None:
|
|
registry = ToolRegistry(timeout_seconds=0.01)
|
|
registry._specs["get_current_time"] = ToolSpec(
|
|
name="get_current_time",
|
|
description="slow",
|
|
arguments_model=CurrentTimeArguments,
|
|
handler=slow_handler,
|
|
)
|
|
|
|
result = await registry.execute(
|
|
"get_current_time", '{"timezone":"UTC"}'
|
|
)
|
|
|
|
assert result.is_error is True
|
|
assert "timeout" in result.content
|
|
|
|
asyncio.run(scenario())
|