"""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