mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 07:17:54 +08:00
184 lines
7.2 KiB
Python
184 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
from plyngent.agent import ToolRegistry
|
|
from plyngent.tools.danger import classify_danger
|
|
from plyngent.tools.file import edit_replace, listdir, read_file, write_file
|
|
from tests.test_tools.helpers import call_sync
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Mapping
|
|
|
|
get_truncated_module = importlib.import_module("plyngent.tools.file.get_truncated")
|
|
|
|
|
|
def _extract_token(out: str) -> str | None:
|
|
for line in out.splitlines():
|
|
if "truncate_token=" in line:
|
|
return line.split("truncate_token=", 1)[1].rstrip("]")
|
|
return None
|
|
|
|
|
|
def test_write_read_listdir_edit(workspace: object) -> None:
|
|
del workspace
|
|
assert "wrote" in call_sync(write_file, "notes/a.txt", "hello world")
|
|
assert call_sync(read_file, "notes/a.txt") == "L1-1\nhello world"
|
|
listing = call_sync(listdir, "notes")
|
|
assert "a.txt" in listing
|
|
assert "file" in listing
|
|
result = call_sync(edit_replace, "notes/a.txt", "world", "there")
|
|
assert "replaced" in result
|
|
assert call_sync(read_file, "notes/a.txt") == "L1-1\nhello there"
|
|
|
|
|
|
def test_write_file_refuses_existing_file_without_overwrite(workspace: Path) -> None:
|
|
assert "wrote" in call_sync(write_file, "keep.txt", "first")
|
|
refused = call_sync(write_file, "keep.txt", "second")
|
|
assert refused == "error: file exists: keep.txt (set overwrite=true to replace it)"
|
|
assert (workspace / "keep.txt").read_text(encoding="utf-8") == "first"
|
|
assert "wrote" in call_sync(write_file, "keep.txt", "second", overwrite=True)
|
|
assert (workspace / "keep.txt").read_text(encoding="utf-8") == "second"
|
|
|
|
|
|
async def test_same_file_writes_in_one_batch_keep_call_order(workspace: Path) -> None:
|
|
"""The second call writes onto the first one's text, not onto the stale text."""
|
|
target = workspace / "notes.txt"
|
|
_ = target.write_text("one", encoding="utf-8")
|
|
confirmed = asyncio.Event()
|
|
|
|
async def on_confirm(name: str, args: Mapping[str, object], reason: str) -> bool:
|
|
del name, args, reason
|
|
await confirmed.wait()
|
|
return True
|
|
|
|
registry = ToolRegistry([write_file, edit_replace], danger=classify_danger, on_confirm=on_confirm)
|
|
rewrite = asyncio.create_task(
|
|
registry.execute("write_file", '{"path": "notes.txt", "content": "two", "overwrite": true}')
|
|
)
|
|
await asyncio.sleep(0) # the rewrite is waiting on its overwrite confirm
|
|
edit = asyncio.create_task(
|
|
registry.execute(
|
|
"edit_replace",
|
|
'{"path": "notes.txt", "old_string": "two", "new_string": "three"}',
|
|
)
|
|
)
|
|
await asyncio.sleep(0) # the edit waits for the file instead of editing stale text
|
|
assert target.read_text(encoding="utf-8") == "one"
|
|
confirmed.set()
|
|
rewrite_out, edit_out = await asyncio.gather(rewrite, edit)
|
|
assert "wrote" in rewrite_out
|
|
assert "replaced" in edit_out
|
|
assert target.read_text(encoding="utf-8") == "three"
|
|
|
|
|
|
def test_read_file_max_chars_embeds_token(workspace: Path) -> None:
|
|
(workspace / "big.txt").write_text("a" * 1000, encoding="utf-8")
|
|
out = call_sync(read_file, "big.txt", max_chars=200)
|
|
header, _, rest = out.partition("\n")
|
|
assert header.startswith("L1-") # 1-based range of the raw slice read
|
|
content, _, marker = rest.partition("\n[Truncated")
|
|
assert content == "a" * len(content)
|
|
assert len(content) < 200
|
|
assert "truncate_token=" in marker
|
|
token = _extract_token(out)
|
|
assert token is not None
|
|
parsed = get_truncated_module.decode_truncate_token(token)
|
|
assert parsed is not None
|
|
assert parsed.kind == "file"
|
|
assert parsed.offset == len(content) # cursor lands exactly where content stops
|
|
|
|
|
|
def test_read_file_default_truncates_at_cap(workspace: Path) -> None:
|
|
"""Truncation is the default: a large read embeds a token without max_chars."""
|
|
(workspace / "huge.txt").write_text("\n".join(f"line-{i}" for i in range(10_000)), encoding="utf-8")
|
|
out = call_sync(read_file, "huge.txt")
|
|
header, _, rest = out.partition("\n")
|
|
assert header == "L1-10000" # the requested slice (whole file)
|
|
assert "truncate_token=" in rest
|
|
# The slice is capped near the default tool-result cap.
|
|
assert len(rest) <= 32_000
|
|
# No marker when the file is small (no truncation occurred).
|
|
small = call_sync(read_file, "huge.txt", max_chars=0)
|
|
assert "Truncated" not in small
|
|
assert small.startswith("L1-10000\n")
|
|
|
|
|
|
def test_read_file_max_chars_small_file_untouched(workspace: Path) -> None:
|
|
(workspace / "small.txt").write_text("tiny", encoding="utf-8")
|
|
out = call_sync(read_file, "small.txt", max_chars=200)
|
|
assert out == "L1-1\ntiny"
|
|
assert "truncate_token=" not in out
|
|
|
|
|
|
def test_edit_replace_first_only(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "aa aa")
|
|
result = call_sync(edit_replace, "t.txt", "aa", "bb")
|
|
assert call_sync(read_file, "t.txt") == "L1-1\nbb aa"
|
|
assert "1 of 2" in result or "1 of 2 matches" in result
|
|
assert "remain" in result
|
|
assert "'aa' → 'bb'" in result # old→new excerpt for self-correction
|
|
|
|
|
|
def test_edit_replace_max_replaces(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "aa aa aa")
|
|
result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=2)
|
|
assert call_sync(read_file, "t.txt") == "L1-1\nbb bb aa"
|
|
assert "2 of 3" in result
|
|
assert "1 remain" in result
|
|
|
|
|
|
def test_edit_replace_all_matches(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "aa aa")
|
|
result = call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=10)
|
|
assert call_sync(read_file, "t.txt") == "L1-1\nbb bb"
|
|
assert "all 2 matches" in result
|
|
|
|
|
|
def test_edit_replace_max_replaces_invalid(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "aa")
|
|
assert "max_replaces" in call_sync(edit_replace, "t.txt", "aa", "bb", max_replaces=0)
|
|
|
|
|
|
def test_edit_missing_old_string(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "x")
|
|
assert "not found" in call_sync(edit_replace, "t.txt", "missing", "y")
|
|
|
|
|
|
def test_read_offset_limit(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "lines.txt", "a\nb\nc\nd\n")
|
|
# offset is 0-based: offset=1 skips line 1, so the read starts at 1-based line 2.
|
|
assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "L2-3\nb\nc\n"
|
|
|
|
|
|
def test_read_file_missing_and_directory(workspace: Path) -> None:
|
|
(workspace / "sub").mkdir()
|
|
assert call_sync(read_file, "nope.txt") == "error: file not found: nope.txt"
|
|
assert call_sync(read_file, "sub") == "error: not a file: sub"
|
|
|
|
|
|
def test_read_with_lineno(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "num.txt", "a\nb\nc\n")
|
|
out = call_sync(read_file, "num.txt", with_lineno=True)
|
|
assert " 1|a\n" in out
|
|
assert " 2|b\n" in out
|
|
assert " 3|c\n" in out
|
|
# offset is 0-based; line numbers stay absolute 1-based file lines
|
|
mid = call_sync(read_file, "num.txt", offset=1, limit=1, with_lineno=True)
|
|
assert mid == " 2|b\n"
|
|
|
|
|
|
def test_listdir_missing(workspace: object) -> None:
|
|
del workspace
|
|
assert "error" in call_sync(listdir, "nope")
|