330 lines
12 KiB
Python
330 lines
12 KiB
Python
"""filectl parsing + execution — the pseudo-commands the server can relay to
|
|
read/write/edit local files instead of a raw shell heredoc.
|
|
|
|
filectl {"op": ...} is a single-line JSON envelope (not a multi-line
|
|
marker block) because it rides the "command" tool marker, which the main
|
|
repo's tool-call extractor only captures up to the next newline — see the
|
|
module docstring in bolt_pet/file_ops.py."""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from bolt_pet import file_ops
|
|
|
|
|
|
def _cmd(payload: dict) -> str:
|
|
return "filectl " + json.dumps(payload)
|
|
|
|
|
|
# ── parsing ──────────────────────────────────────────────────────────────────
|
|
|
|
def test_non_file_commands_are_left_alone():
|
|
assert file_ops.parse("ls -la") is None
|
|
assert file_ops.parse("systemctl restart nginx") is None
|
|
assert file_ops.parse("") is None
|
|
# "filed" must not be mistaken for the "file" prefix
|
|
assert file_ops.parse("filed --list") is None
|
|
|
|
|
|
def test_list_parses_defaults():
|
|
assert file_ops.parse(_cmd({"op": "list", "path": "/tmp"})) == {
|
|
"action": "list", "path": "/tmp", "pattern": "*", "recursive": False,
|
|
}
|
|
|
|
|
|
def test_list_parses_pattern_and_recursive():
|
|
assert file_ops.parse(_cmd({"op": "list", "path": "/tmp", "pattern": "*.py", "recursive": True})) == {
|
|
"action": "list", "path": "/tmp", "pattern": "*.py", "recursive": True,
|
|
}
|
|
|
|
|
|
def test_list_requires_a_path():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "list"}))
|
|
|
|
|
|
def test_list_rejects_empty_pattern():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "list", "path": "/tmp", "pattern": ""}))
|
|
|
|
|
|
def test_read_parses_path_and_optional_line_range():
|
|
assert file_ops.parse(_cmd({"op": "read", "path": "/tmp/a.txt"})) == {
|
|
"action": "read", "path": "/tmp/a.txt", "start": None, "end": None,
|
|
}
|
|
assert file_ops.parse(_cmd({"op": "read", "path": "/tmp/a.txt", "start": 10, "end": 40})) == {
|
|
"action": "read", "path": "/tmp/a.txt", "start": 10, "end": 40,
|
|
}
|
|
|
|
|
|
def test_read_requires_a_path():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "read"}))
|
|
|
|
|
|
def test_read_rejects_non_numeric_line_args():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse('filectl {"op": "read", "path": "/tmp/a.txt", "start": "start"}')
|
|
|
|
|
|
def test_write_parses_path_and_content():
|
|
assert file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt", "content": "hello\nworld"})) == {
|
|
"action": "write", "path": "/tmp/a.txt", "content": "hello\nworld",
|
|
}
|
|
|
|
|
|
def test_write_allows_empty_content():
|
|
assert file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt", "content": ""})) == {
|
|
"action": "write", "path": "/tmp/a.txt", "content": "",
|
|
}
|
|
|
|
|
|
def test_write_without_content_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "write", "path": "/tmp/a.txt"}))
|
|
|
|
|
|
def test_edit_parses_old_and_new():
|
|
assert file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt", "old": "foo\nbar", "new": "baz"})) == {
|
|
"action": "edit", "path": "/tmp/a.txt", "old": "foo\nbar", "new": "baz",
|
|
}
|
|
|
|
|
|
def test_edit_rejects_identical_old_and_new():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt", "old": "same", "new": "same"}))
|
|
|
|
|
|
def test_edit_missing_fields_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "edit", "path": "/tmp/a.txt"}))
|
|
|
|
|
|
def test_content_with_embedded_quotes_and_shell_metacharacters_survives():
|
|
payload = 'echo "hi $USER" `whoami` && rm -rf /'
|
|
command = _cmd({"op": "write", "path": "/tmp/a.txt", "content": payload})
|
|
assert "\n" not in command # stays on one line, as the command marker requires
|
|
assert file_ops.parse(command) == {"action": "write", "path": "/tmp/a.txt", "content": payload}
|
|
|
|
|
|
def test_multiline_content_stays_on_one_physical_line():
|
|
content = "line one\nline two\nline three with \"quotes\" and \\backslashes\\"
|
|
command = _cmd({"op": "write", "path": "/tmp/a.txt", "content": content})
|
|
assert "\n" not in command
|
|
assert file_ops.parse(command)["content"] == content
|
|
|
|
|
|
def test_help():
|
|
assert file_ops.parse("filectl help") == {"action": "help"}
|
|
assert file_ops.parse("filectl") == {"action": "help"}
|
|
assert file_ops.parse(_cmd({"op": "help"})) == {"action": "help"}
|
|
|
|
|
|
def test_invalid_json_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse("filectl {not valid json")
|
|
|
|
|
|
def test_non_object_json_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse("filectl [1, 2, 3]")
|
|
|
|
|
|
def test_unknown_op_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.parse(_cmd({"op": "frobnicate", "path": "/tmp/a.txt"}))
|
|
|
|
|
|
# ── execution ────────────────────────────────────────────────────────────────
|
|
|
|
def test_list_shows_files_and_subdirectories(tmp_path):
|
|
(tmp_path / "a.txt").write_text("hi")
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "b.txt").write_text("nested")
|
|
|
|
action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)}))
|
|
output = file_ops.execute(action)
|
|
|
|
assert "a.txt\t2B" in output
|
|
assert "sub/" in output
|
|
assert "b.txt" not in output # non-recursive: nested file not shown
|
|
|
|
|
|
def test_list_recursive_finds_nested_files(tmp_path):
|
|
(tmp_path / "sub").mkdir()
|
|
(tmp_path / "sub" / "b.txt").write_text("nested")
|
|
|
|
action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path), "recursive": True}))
|
|
output = file_ops.execute(action)
|
|
|
|
assert "sub/b.txt" in output or "sub\\b.txt" in output # os-dependent separator
|
|
|
|
|
|
def test_list_pattern_filters_entries(tmp_path):
|
|
(tmp_path / "a.py").write_text("x")
|
|
(tmp_path / "b.txt").write_text("y")
|
|
|
|
action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path), "pattern": "*.py"}))
|
|
output = file_ops.execute(action)
|
|
|
|
assert "a.py" in output
|
|
assert "b.txt" not in output
|
|
|
|
|
|
def test_list_empty_directory_says_so(tmp_path):
|
|
action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)}))
|
|
output = file_ops.execute(action)
|
|
assert "no entries" in output
|
|
|
|
|
|
def test_list_missing_directory_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.execute({"action": "list", "path": "/no/such/dir", "pattern": "*", "recursive": False})
|
|
|
|
|
|
def test_list_rejects_a_file_path(tmp_path):
|
|
target = tmp_path / "a.txt"
|
|
target.write_text("hi")
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.execute({"action": "list", "path": str(target), "pattern": "*", "recursive": False})
|
|
|
|
|
|
def test_list_truncates_past_the_entry_cap(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(file_ops, "_MAX_LIST_ENTRIES", 3)
|
|
for i in range(5):
|
|
(tmp_path / f"f{i}.txt").write_text("x")
|
|
|
|
action = file_ops.parse(_cmd({"op": "list", "path": str(tmp_path)}))
|
|
output = file_ops.execute(action)
|
|
|
|
assert "truncated" in output
|
|
assert output.count(".txt") == 3
|
|
|
|
|
|
def test_write_then_read_round_trips(tmp_path):
|
|
target = tmp_path / "notes.txt"
|
|
write_action = file_ops.parse(_cmd({"op": "write", "path": str(target), "content": "line one\nline two"}))
|
|
result = file_ops.execute(write_action)
|
|
assert target.read_text() == "line one\nline two"
|
|
assert str(target) in result
|
|
|
|
read_action = file_ops.parse(_cmd({"op": "read", "path": str(target)}))
|
|
output = file_ops.execute(read_action)
|
|
assert "line one" in output
|
|
assert "line two" in output
|
|
|
|
|
|
def test_read_missing_file_raises():
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.execute({"action": "read", "path": "/no/such/file.txt", "start": None, "end": None})
|
|
|
|
|
|
def test_read_respects_line_range(tmp_path):
|
|
target = tmp_path / "a.txt"
|
|
target.write_text("\n".join(f"line{i}" for i in range(1, 11)))
|
|
action = file_ops.parse(_cmd({"op": "read", "path": str(target), "start": 3, "end": 5}))
|
|
output = file_ops.execute(action)
|
|
assert "line3" in output and "line5" in output
|
|
assert "line1" not in output and "line6" not in output
|
|
|
|
|
|
def test_edit_replaces_a_unique_match(tmp_path):
|
|
target = tmp_path / "a.py"
|
|
target.write_text("def foo():\n return 1\n")
|
|
action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "return 1", "new": "return 2"}))
|
|
file_ops.execute(action)
|
|
assert target.read_text() == "def foo():\n return 2\n"
|
|
|
|
|
|
def test_edit_fails_when_text_not_found(tmp_path):
|
|
target = tmp_path / "a.py"
|
|
target.write_text("def foo():\n return 1\n")
|
|
action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "nope", "new": "x"}))
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.execute(action)
|
|
assert target.read_text() == "def foo():\n return 1\n" # untouched
|
|
|
|
|
|
def test_edit_fails_when_text_is_ambiguous(tmp_path):
|
|
target = tmp_path / "a.py"
|
|
target.write_text("x = 1\nx = 1\n")
|
|
action = file_ops.parse(_cmd({"op": "edit", "path": str(target), "old": "x = 1", "new": "x = 2"}))
|
|
with pytest.raises(file_ops.FileOpError):
|
|
file_ops.execute(action)
|
|
assert target.read_text() == "x = 1\nx = 1\n" # untouched
|
|
|
|
|
|
def test_write_creates_parent_directories(tmp_path):
|
|
target = tmp_path / "nested" / "dir" / "a.txt"
|
|
action = file_ops.parse(_cmd({"op": "write", "path": str(target), "content": "hi"}))
|
|
file_ops.execute(action)
|
|
assert target.read_text() == "hi"
|
|
|
|
|
|
# ── lenient JSON (see relay_json) ───────────────────────────────────────────
|
|
# Machine-written JSON fails in a small, repeatable set of ways. Observed live
|
|
# 2026-07-30: a stray quote after `false` cost a whole desk turn — rejected,
|
|
# re-sent identically, rejected again, then abandoned with a promise to the
|
|
# user that nothing fulfilled.
|
|
|
|
def test_the_stray_quote_that_cost_a_live_turn_now_parses():
|
|
action = file_ops.parse(
|
|
'filectl {"op":"list","path":"/home/maji/Documents","pattern":"*","recursive":false"}'
|
|
)
|
|
assert action["action"] == "list"
|
|
assert action["path"] == "/home/maji/Documents"
|
|
assert action["recursive"] is False
|
|
assert action["_repairs"] == ["removed a stray quote after a bare value"]
|
|
|
|
|
|
@pytest.mark.parametrize("payload,expected", [
|
|
('{"op": "read", "path": "/tmp/a.txt",}', "removed a trailing comma"),
|
|
("{'op': 'read', 'path': '/tmp/a.txt'}", "converted single-quoted strings to double-quoted"),
|
|
('{“op”: “read”, “path”: “/tmp/a.txt”}', "replaced smart quotes with straight ones"),
|
|
('```json {"op": "read", "path": "/tmp/a.txt"} ```', "stripped a markdown code fence"),
|
|
])
|
|
def test_common_model_json_mistakes_are_repaired(payload, expected):
|
|
action = file_ops.parse("filectl " + payload)
|
|
assert action["action"] == "read"
|
|
assert action["path"] == "/tmp/a.txt"
|
|
assert expected in action["_repairs"]
|
|
|
|
|
|
def test_python_literals_are_converted():
|
|
action = file_ops.parse('filectl {"op": "list", "path": "/tmp", "recursive": True}')
|
|
assert action["recursive"] is True
|
|
|
|
|
|
def test_a_repair_is_reported_in_the_output_never_hidden(tmp_path):
|
|
"""Silently fixing it would work today and guarantee the same broken call
|
|
tomorrow — the model has to be told while it still has the turn."""
|
|
(tmp_path / "a.txt").write_text("hello", encoding="utf-8")
|
|
action = file_ops.parse(
|
|
'filectl {"op":"list","path":"%s","recursive":false"}' % tmp_path
|
|
)
|
|
output = file_ops.execute(action)
|
|
assert "a.txt" in output
|
|
assert "your JSON was malformed" in output
|
|
assert "stray quote" in output
|
|
|
|
|
|
def test_valid_json_gets_no_repair_note(tmp_path):
|
|
(tmp_path / "a.txt").write_text("hello", encoding="utf-8")
|
|
action = file_ops.parse('filectl {"op": "list", "path": "%s"}' % tmp_path)
|
|
assert "_repairs" not in action
|
|
assert "malformed" not in file_ops.execute(action)
|
|
|
|
|
|
def test_genuinely_unparseable_json_points_at_the_character():
|
|
""""Expecting ',' delimiter: char 74" is not something a model can act on."""
|
|
with pytest.raises(file_ops.FileOpError) as excinfo:
|
|
file_ops.parse('filectl {"op": "read", "path": "/tmp/a.txt" "extra": 1}')
|
|
message = str(excinfo.value)
|
|
assert "^" in message # caret under the offending character
|
|
assert '"extra"' in message # ...and the fragment around it
|