bvandeusen
278927ec40
Adds read_article tool definition and execute_tool handler. Uses _fetch_full_article (trafilatura) from rss.py, caps tool output at 40K chars to keep context window manageable. Always registered (not gated on SearXNG). Tests cover success, failure, truncation, empty URL, and history builder replay. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
137 lines
4.9 KiB
Python
137 lines
4.9 KiB
Python
import json
|
|
import pytest
|
|
from unittest.mock import patch, AsyncMock
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# read_article tool tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_article_success():
|
|
"""read_article returns success with content when _fetch_full_article succeeds."""
|
|
from fabledassistant.services.tools import execute_tool
|
|
|
|
with patch(
|
|
"fabledassistant.services.rss._fetch_full_article",
|
|
new_callable=AsyncMock,
|
|
return_value="Article body text here.",
|
|
):
|
|
result = await execute_tool(user_id=1, tool_name="read_article", arguments={"url": "https://example.com/article"})
|
|
|
|
assert result["success"] is True
|
|
assert result["type"] == "article_content"
|
|
assert result["url"] == "https://example.com/article"
|
|
assert result["content"] == "Article body text here."
|
|
assert result["truncated"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_article_fetch_failure():
|
|
"""read_article returns success=False when _fetch_full_article returns None."""
|
|
from fabledassistant.services.tools import execute_tool
|
|
|
|
with patch(
|
|
"fabledassistant.services.rss._fetch_full_article",
|
|
new_callable=AsyncMock,
|
|
return_value=None,
|
|
):
|
|
result = await execute_tool(user_id=1, tool_name="read_article", arguments={"url": "https://example.com/broken"})
|
|
|
|
assert result["success"] is False
|
|
assert "Could not fetch" in result["error"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_article_truncates_at_40k():
|
|
"""read_article truncates content at 40,000 chars and sets truncated=True."""
|
|
from fabledassistant.services.tools import execute_tool
|
|
|
|
long_content = "x" * 50_000
|
|
|
|
with patch(
|
|
"fabledassistant.services.rss._fetch_full_article",
|
|
new_callable=AsyncMock,
|
|
return_value=long_content,
|
|
):
|
|
result = await execute_tool(user_id=1, tool_name="read_article", arguments={"url": "https://example.com/long"})
|
|
|
|
assert result["success"] is True
|
|
assert len(result["content"]) == 40_000
|
|
assert result["truncated"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_read_article_empty_url():
|
|
"""read_article returns success=False when URL is empty."""
|
|
from fabledassistant.services.tools import execute_tool
|
|
|
|
result = await execute_tool(user_id=1, tool_name="read_article", arguments={"url": ""})
|
|
assert result["success"] is False
|
|
assert "No URL provided" in result["error"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# History builder tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_history(messages: list[dict]) -> list[dict]:
|
|
"""Replicate the fixed history builder from routes/chat.py."""
|
|
history = []
|
|
for msg in messages:
|
|
if msg["role"] == "system":
|
|
continue
|
|
msg_dict = {"role": msg["role"], "content": msg.get("content") or ""}
|
|
tool_calls = msg.get("tool_calls")
|
|
if tool_calls:
|
|
msg_dict["tool_calls"] = [
|
|
{"function": {"name": tc["function"], "arguments": tc["arguments"]}}
|
|
for tc in tool_calls
|
|
]
|
|
history.append(msg_dict)
|
|
for tc in tool_calls:
|
|
history.append({"role": "tool", "content": json.dumps(tc.get("result", {}))})
|
|
else:
|
|
history.append(msg_dict)
|
|
return history
|
|
|
|
|
|
def test_history_builder_replays_tool_calls():
|
|
"""History builder with tool_calls produces assistant entry + tool result entry."""
|
|
messages = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [
|
|
{
|
|
"function": "read_article",
|
|
"arguments": {"url": "https://example.com"},
|
|
"result": {"success": True, "content": "Article text"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "user", "content": "Summarize it", "tool_calls": None},
|
|
]
|
|
history = _build_history(messages)
|
|
|
|
assert len(history) == 3
|
|
assert history[0]["role"] == "assistant"
|
|
assert history[0]["tool_calls"][0]["function"]["name"] == "read_article"
|
|
assert history[1]["role"] == "tool"
|
|
assert json.loads(history[1]["content"])["success"] is True
|
|
assert history[2]["role"] == "user"
|
|
assert history[2]["content"] == "Summarize it"
|
|
|
|
|
|
def test_history_builder_no_tool_calls_unchanged():
|
|
"""History builder with tool_calls=None produces same output as before."""
|
|
messages = [
|
|
{"role": "user", "content": "Hello", "tool_calls": None},
|
|
{"role": "assistant", "content": "Hi there!", "tool_calls": None},
|
|
]
|
|
history = _build_history(messages)
|
|
|
|
assert len(history) == 2
|
|
assert history[0] == {"role": "user", "content": "Hello"}
|
|
assert history[1] == {"role": "assistant", "content": "Hi there!"}
|