mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 15:37:55 +08:00
NCBM
2cf66edb60
Wrong-target edits (the test_run_aside pass clobber, the edit_replace
mangling) were silent: the success message reported counts only, so the
model had no signal it replaced the wrong thing until later.
edit_lineno now reports the first removed line ("first: 'pass'") and
edit_replace reports the matched pattern ('old' -> 'new'), giving the
model an immediate self-correction signal.
159 lines
6.3 KiB
Python
159 lines
6.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from plyngent.agent import reset_lineno_tracker
|
|
from plyngent.tools.file import edit_lineno, edit_replace, read_file, write_file
|
|
from tests.test_tools.helpers import call_sync
|
|
|
|
|
|
def _read_lineno(path: str) -> str:
|
|
"""Read file with line numbers so edit_lineno allows the edit."""
|
|
_ = reset_lineno_tracker()
|
|
return call_sync(read_file, path, with_lineno=True)
|
|
|
|
|
|
def test_edit_lineno_replace_middle(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "a.txt", "one\ntwo\nthree\nfour\n")
|
|
_ = _read_lineno("a.txt")
|
|
out = call_sync(edit_lineno, "a.txt", 2, 3, "TWO\nTHREE\n")
|
|
assert "replaced lines 2-3" in out
|
|
assert "first: 'two'" in out # old-content excerpt for self-correction
|
|
assert call_sync(read_file, "a.txt") == "L1-4\none\nTWO\nTHREE\nfour\n"
|
|
|
|
|
|
def test_edit_lineno_read_truncated_marks_only_displayed_lines(workspace: object) -> None:
|
|
"""A max_chars-truncated numbered read must not mark the unseen tail."""
|
|
del workspace
|
|
_ = call_sync(write_file, "t.txt", "".join(f"line {i}\n" for i in range(1, 101)))
|
|
# Displayed complete lines are editable...
|
|
_ = reset_lineno_tracker()
|
|
_ = call_sync(read_file, "t.txt", with_lineno=True, max_chars=80)
|
|
out = call_sync(edit_lineno, "t.txt", 5, 5, "FIVE\n")
|
|
assert "replaced lines 5-5" in out
|
|
# ...but the truncated tail (never shown) stays unread.
|
|
_ = reset_lineno_tracker()
|
|
_ = call_sync(read_file, "t.txt", with_lineno=True, max_chars=80)
|
|
out = call_sync(edit_lineno, "t.txt", 90, 90, "X\n")
|
|
assert "not read" in out and "90" in out
|
|
|
|
|
|
def test_edit_lineno_error_shows_read_window(workspace: object) -> None:
|
|
"""The unread error names the read window so the model can re-read precisely."""
|
|
del workspace
|
|
_ = call_sync(write_file, "j.txt", "a\nb\nc\nd\n")
|
|
_ = reset_lineno_tracker()
|
|
_ = call_sync(read_file, "j.txt", offset=1, limit=2, with_lineno=True) # lines 2-3
|
|
out = call_sync(edit_lineno, "j.txt", 1, 1, "X\n")
|
|
assert "not read" in out
|
|
assert "read lines 2-3" in out
|
|
assert "margin" in out
|
|
|
|
|
|
def test_edit_lineno_requires_all_lines_read(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "r.txt", "a\nb\nc\nd\ne\n")
|
|
_ = reset_lineno_tracker()
|
|
# Only lines 2-4 are read; editing line 1 or 5 must be rejected.
|
|
_ = call_sync(read_file, "r.txt", offset=1, limit=3, with_lineno=True)
|
|
out = call_sync(edit_lineno, "r.txt", 1, 2, "X\n")
|
|
assert "not read" in out and "1" in out
|
|
out = call_sync(edit_lineno, "r.txt", 4, 5, "Y\n")
|
|
assert "not read" in out and "5" in out
|
|
# Fully read range still edits.
|
|
out = call_sync(edit_lineno, "r.txt", 2, 3, "B\nC\n")
|
|
assert "replaced lines 2-3" in out
|
|
|
|
|
|
def test_edit_lineno_append_requires_last_line_read(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "ap.txt", "a\nb\nc\n")
|
|
_ = reset_lineno_tracker()
|
|
# Only lines 1-2 read; appending after line 3 requires line 3 read.
|
|
_ = call_sync(read_file, "ap.txt", offset=0, limit=2, with_lineno=True)
|
|
out = call_sync(edit_lineno, "ap.txt", 4, 4, "d\n")
|
|
assert "not read" in out and "3" in out
|
|
# Read the last line, then append succeeds.
|
|
_ = call_sync(read_file, "ap.txt", offset=2, limit=1, with_lineno=True)
|
|
out = call_sync(edit_lineno, "ap.txt", 4, 4, "d\n")
|
|
assert "appended" in out
|
|
assert call_sync(read_file, "ap.txt") == "L1-4\na\nb\nc\nd\n"
|
|
|
|
|
|
def test_edit_lineno_delete_range(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "b.txt", "a\nb\nc\n")
|
|
_ = _read_lineno("b.txt")
|
|
out = call_sync(edit_lineno, "b.txt", 2, 2, "")
|
|
assert "replaced" in out
|
|
assert call_sync(read_file, "b.txt") == "L1-2\na\nc\n"
|
|
|
|
|
|
def test_edit_lineno_append(workspace: object) -> None:
|
|
assert isinstance(workspace, Path)
|
|
_ = call_sync(write_file, "c.txt", "only\n")
|
|
_ = _read_lineno("c.txt")
|
|
out = call_sync(edit_lineno, "c.txt", 2, 2, "more\n")
|
|
assert "appended" in out
|
|
assert call_sync(read_file, "c.txt") == "L1-2\nonly\nmore\n"
|
|
|
|
|
|
def test_edit_lineno_invalid(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "d.txt", "x\n")
|
|
_ = _read_lineno("d.txt")
|
|
assert "start_line" in call_sync(edit_lineno, "d.txt", 0, 1, "y")
|
|
assert "end_line" in call_sync(edit_lineno, "d.txt", 2, 1, "y")
|
|
|
|
|
|
def test_edit_lineno_rejects_without_prior_lineno_read(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "e.txt", "hello\nworld\n")
|
|
_ = reset_lineno_tracker()
|
|
out = call_sync(edit_lineno, "e.txt", 1, 1, "hi\n")
|
|
assert "read" in out and "with_lineno=true" in out
|
|
|
|
|
|
def test_edit_lineno_requires_rerun_after_write(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "f.txt", "one\ntwo\nthree\n")
|
|
_ = _read_lineno("f.txt")
|
|
_ = call_sync(write_file, "f.txt", "changed\n")
|
|
out = call_sync(edit_lineno, "f.txt", 1, 1, "x\n")
|
|
assert "changed" in out and "re-read" in out
|
|
|
|
|
|
def test_edit_lineno_requires_rerun_after_edit_replace(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "g.txt", "one\ntwo\nthree\n")
|
|
_ = _read_lineno("g.txt")
|
|
_ = call_sync(edit_replace, "g.txt", "two", "TWO")
|
|
out = call_sync(edit_lineno, "g.txt", 2, 2, "x\n")
|
|
assert "changed" in out and "re-read" in out
|
|
|
|
|
|
def test_edit_lineno_invalidates_after_successful_edit(workspace: object) -> None:
|
|
del workspace
|
|
_ = call_sync(write_file, "h.txt", "a\nb\nc\n")
|
|
_ = _read_lineno("h.txt")
|
|
_ = call_sync(edit_lineno, "h.txt", 2, 2, "B\n")
|
|
# After the edit, line numbers are stale — must re-read before editing again.
|
|
out = call_sync(edit_lineno, "h.txt", 2, 2, "C\n")
|
|
assert "changed" in out and "re-read" in out
|
|
|
|
|
|
def test_edit_lineno_detects_external_change(workspace: object) -> None:
|
|
"""A mutation outside the file tools (run_argv, another process) is caught."""
|
|
import os
|
|
import time
|
|
|
|
assert isinstance(workspace, Path)
|
|
_ = call_sync(write_file, "i.txt", "one\ntwo\nthree\n")
|
|
_ = _read_lineno("i.txt")
|
|
# Simulate an external/process-tool mutation: rewrite content + new mtime.
|
|
_ = call_sync(write_file, "i.txt", "one\ntwo\nthree\nCHANGED\n")
|
|
os.utime(workspace / "i.txt", (time.time() + 1, time.time() + 1))
|
|
out = call_sync(edit_lineno, "i.txt", 1, 1, "x\n")
|
|
assert "changed since it was read" in out
|