Update
This commit is contained in:
@@ -0,0 +1,266 @@
|
||||
"""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"
|
||||
Reference in New Issue
Block a user