mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
NCBM
7d6e367143
run_argv, run_argv_batch, fetch, fetch_local, and the vcs tools now return their output whole, so the request-prep cap trims it on the wire and names its document. A fetched page or a long log reads on from the tool message, with no re-fetch and no re-run. The run_argv_batch run budget stays as an execution guard: it stops a runaway batch from spawning further huge-output steps. Co-authored-by: CommandCodeBot <noreply@commandcode.ai>
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""The remote ``fetch`` tool (DeepSeek's Fetch API, Markdown pages)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from plyngent.agent.tools import ToolTag
|
|
from plyngent.lmproto.deepseek.webtools import FetchResponse
|
|
from plyngent.tools.net.fetch_remote import FETCH_TOOL_NAME, build_fetch_tool
|
|
|
|
|
|
class _Client:
|
|
"""Stands in for DeepseekWebToolsClient.fetch."""
|
|
|
|
def __init__(self, *, page: FetchResponse | None = None, error: str | None = None) -> None:
|
|
self.page = page or FetchResponse(title="T", content="body")
|
|
self.error = error
|
|
self.urls: list[str] = []
|
|
|
|
async def fetch(self, url: str) -> FetchResponse:
|
|
self.urls.append(url)
|
|
if self.error is not None:
|
|
raise RuntimeError(self.error)
|
|
return self.page
|
|
|
|
|
|
def _tool(client: _Client):
|
|
return build_fetch_tool(client)
|
|
|
|
|
|
def test_definition_shape() -> None:
|
|
definition = _tool(_Client())
|
|
assert definition.name == FETCH_TOOL_NAME == "fetch"
|
|
assert definition.tags == ToolTag.LOCAL | ToolTag.READ_ONLY
|
|
properties = definition.parameters["properties"]
|
|
assert set(properties) == {"url", "offset"}
|
|
assert definition.parameters["required"] == ["url"]
|
|
assert "fetch_local" in definition.description
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handler_renders_title_url_and_content() -> None:
|
|
client = _Client(page=FetchResponse(title=" Example article ", content="# Heading\n\nBody."))
|
|
result = await _tool(client).handler(url="https://example.com/article")
|
|
|
|
assert client.urls == ["https://example.com/article"]
|
|
assert result.splitlines() == [
|
|
"title=Example article",
|
|
"url=https://example.com/article",
|
|
"chars=16",
|
|
"--- content ---",
|
|
"# Heading",
|
|
"",
|
|
"Body.",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_offset_skips_leading_content() -> None:
|
|
client = _Client(page=FetchResponse(title="", content="0123456789"))
|
|
result = await _tool(client).handler(url="https://example.com/", offset=5)
|
|
|
|
assert "title=(untitled)" in result
|
|
assert "--- content ---\n56789" in result
|
|
assert "truncate_token=" not in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_handler_reports_a_client_error() -> None:
|
|
result = await _tool(_Client(error="HTTP 404: no usable content")).handler(url="https://example.com/x")
|
|
assert result == "error: fetch failed: HTTP 404: no usable content"
|