ryan
d2f21a6d23
Pure functions build tool_start/tool_end payloads with run_id, friendly label, truncated input (config dropped), status, duration and artifact ref. Labels cover every agent tool, both old and #55 YNAB names, and the homebox_/bookstack_ MCP prefixes, falling back to "Using <tool>…". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""Tests for SSE tool-event shaping and the tool label registry."""
|
|
|
|
from langchain_core.tools import BaseTool
|
|
|
|
from blueprints.conversation import agents
|
|
from blueprints.conversation.tool_events import (
|
|
INPUT_VALUE_MAX_CHARS,
|
|
summarize_tool_input,
|
|
tool_end_payload,
|
|
tool_name_from_message,
|
|
tool_start_payload,
|
|
)
|
|
from blueprints.conversation.tool_labels import TOOL_LABELS, tool_label
|
|
|
|
|
|
def _event(name="web_search", run_id="run-1", input=None):
|
|
return {"name": name, "run_id": run_id, "data": {"input": input or {}}}
|
|
|
|
|
|
class TestToolLabel:
|
|
def test_known_tool(self):
|
|
assert tool_label("web_search") == "Searching the web…"
|
|
|
|
def test_fallback(self):
|
|
assert tool_label("mystery_tool") == "Using mystery_tool…"
|
|
|
|
def test_mcp_prefixes(self):
|
|
assert tool_label("homebox_search_items") == "Checking home inventory…"
|
|
assert tool_label("bookstack_get_page") == "Checking the wiki…"
|
|
|
|
def test_old_and_new_ynab_names(self):
|
|
for name in (
|
|
"ynab_category_spending",
|
|
"ynab_insights",
|
|
"ynab_account_balances",
|
|
"ynab_spending_report",
|
|
"ynab_category_budget",
|
|
"ynab_recent_changes",
|
|
"ynab_sql",
|
|
):
|
|
assert name in TOOL_LABELS
|
|
|
|
def test_every_agent_tool_has_a_label(self):
|
|
names = [v.name for v in vars(agents).values() if isinstance(v, BaseTool)]
|
|
assert len(names) > 20
|
|
assert [n for n in names if n not in TOOL_LABELS] == []
|
|
|
|
|
|
class TestSummarizeToolInput:
|
|
def test_drops_config_and_none(self):
|
|
raw = {"content": "hi", "config": {"configurable": {}}, "date": None}
|
|
assert summarize_tool_input(raw) == {"content": "hi"}
|
|
|
|
def test_truncates_long_strings(self):
|
|
out = summarize_tool_input({"query": "x" * 500})
|
|
assert len(out["query"]) == INPUT_VALUE_MAX_CHARS
|
|
assert out["query"].endswith("…")
|
|
|
|
def test_keeps_scalars(self):
|
|
assert summarize_tool_input({"days": 7, "flag": True, "f": 1.5}) == {
|
|
"days": 7,
|
|
"flag": True,
|
|
"f": 1.5,
|
|
}
|
|
|
|
def test_serialises_nested_values(self):
|
|
assert summarize_tool_input({"tags": ["a", "b"]}) == {"tags": '["a", "b"]'}
|
|
|
|
def test_non_dict_input(self):
|
|
assert summarize_tool_input("raw") == {"input": "raw"}
|
|
assert summarize_tool_input(None) == {}
|
|
|
|
|
|
class TestPayloads:
|
|
def test_tool_start(self):
|
|
payload = tool_start_payload(_event(input={"query": "cats"}))
|
|
assert payload == {
|
|
"type": "tool_start",
|
|
"run_id": "run-1",
|
|
"tool": "web_search",
|
|
"label": "Searching the web…",
|
|
"input": {"query": "cats"},
|
|
}
|
|
|
|
def test_tool_end_ok(self):
|
|
ref = {"id": "a1", "kind": "mealie.recipe", "title": "Soup"}
|
|
payload = tool_end_payload(_event(), 10.0, ref, now=10.25)
|
|
assert payload == {
|
|
"type": "tool_end",
|
|
"run_id": "run-1",
|
|
"tool": "web_search",
|
|
"label": "Searching the web…",
|
|
"status": "ok",
|
|
"ms": 250,
|
|
"artifact": ref,
|
|
}
|
|
|
|
def test_tool_end_error_without_start(self):
|
|
payload = tool_end_payload(_event(), None, None, now=5.0, status="error")
|
|
assert payload["status"] == "error"
|
|
assert payload["ms"] == 0
|
|
assert payload["artifact"] is None
|
|
|
|
|
|
class TestToolNameFromMessage:
|
|
def test_parses_header(self):
|
|
assert tool_name_from_message("[web_search result]\nbody") == "web_search"
|
|
|
|
def test_unknown_header(self):
|
|
assert tool_name_from_message("something else") == "tool"
|