"""Unit tests for the split filesystem tools (read/write/AI)."""
import pytest
from navi.tools.filesystem import (
FileAiTool,
FilesystemReadTool,
FilesystemWriteTool,
_check_path,
)
class TestCheckPath:
@pytest.fixture(autouse=True)
def _allow_all(self, monkeypatch):
import navi.tools._filesystem.base as _base
import navi.tools._filesystem.helpers as _helpers
from navi.config import Settings
new = Settings(fs_allowed_paths="*")
for mod in (_helpers, _base):
monkeypatch.setattr(mod, "settings", new)
def test_resolves_relative(self):
p = _check_path(".")
assert p is not None
assert p.is_dir()
def test_expands_tilde(self):
p = _check_path("~")
assert p is not None
def test_rejects_empty(self):
assert _check_path("") is None
def test_restricted_paths(self, monkeypatch, tmp_path):
import navi.tools._filesystem.helpers as _helpers
from navi.config import Settings
allowed = tmp_path / "allowed"
allowed.mkdir()
blocked = tmp_path / "blocked"
blocked.mkdir()
new = Settings(fs_allowed_paths=str(allowed))
monkeypatch.setattr(_helpers, "settings", new)
assert _check_path(str(allowed / "file.txt")) is not None
assert _check_path(str(blocked / "file.txt")) is None
class TestCheckPathSandbox:
"""Non-admin paths live in the sandbox or in the current session directory."""
@pytest.fixture(autouse=True)
def _areas(self, monkeypatch, tmp_path):
import navi.session_files as _sf_mod
from navi.config import Settings
from navi.tools._internal.base import current_session_id
monkeypatch.setattr(
_sf_mod,
"settings",
Settings(session_files_dir=str(tmp_path / "sessions"), fs_allowed_paths="*"),
)
monkeypatch.chdir(tmp_path)
self.root = tmp_path
token = current_session_id.set("sess 1")
yield
current_session_id.reset(token)
def test_absolute_inside_sandbox_allowed(self):
target = self.root / "user_data" / "7" / "notes.txt"
checked = _check_path(str(target), user_id="7", role="user")
assert checked == target.resolve()
def test_absolute_inside_session_dir_allowed(self):
session = self.root / "sessions" / "sess 1"
session.mkdir(parents=True)
target = session / "upload.csv"
target.write_text("a,b\n")
checked = _check_path(str(target), user_id="7", role="user")
assert checked == target.resolve()
def test_absolute_outside_both_denied(self):
outside = self.root / "elsewhere.txt"
outside.write_text("x")
assert _check_path(str(outside), user_id="7", role="user") is None
def test_other_session_dir_denied(self):
other = self.root / "sessions" / "sess 2"
other.mkdir(parents=True)
target = other / "leaked.txt"
target.write_text("x")
assert _check_path(str(target), user_id="7", role="user") is None
def test_relative_still_resolves_into_sandbox(self):
checked = _check_path("notes.txt", user_id="7", role="user")
assert checked == (self.root / "user_data" / "7" / "notes.txt").resolve()
def test_relative_dotdot_escape_denied(self):
(self.root / "elsewhere.txt").write_text("x")
assert _check_path("../elsewhere.txt", user_id="7", role="user") is None
def test_relative_symlink_escape_denied(self):
outside = self.root / "elsewhere.txt"
outside.write_text("x")
sandbox = self.root / "user_data" / "7"
sandbox.mkdir(parents=True, exist_ok=True)
(sandbox / "link.txt").symlink_to(outside)
assert _check_path("link.txt", user_id="7", role="user") is None
@pytest.fixture
def allow_all(monkeypatch):
import navi.tools._filesystem.base as _base
import navi.tools._filesystem.helpers as _helpers
from navi.config import Settings
new = Settings(fs_allowed_paths="*")
for mod in (_helpers, _base):
monkeypatch.setattr(mod, "settings", new)
@pytest.fixture
def reader():
return FilesystemReadTool()
@pytest.fixture
def writer():
return FilesystemWriteTool()
class TestFilesystemRead:
async def test_read_file(self, reader, tmp_path, allow_all):
f = tmp_path / "hello.txt"
f.write_text("world")
result = await reader.execute({"action": "read", "path": str(f)})
assert result.success
assert "world" in result.output
async def test_read_missing(self, reader, tmp_path, allow_all):
f = tmp_path / "missing.txt"
result = await reader.execute({"action": "read", "path": str(f)})
assert not result.success
assert "not found" in result.output.lower()
async def test_read_directory_rejected(self, reader, tmp_path, allow_all):
result = await reader.execute({"action": "read", "path": str(tmp_path)})
assert not result.success
assert result.error == "is_directory"
assert "list" in result.output
async def test_read_pages_large_file(self, reader, tmp_path, allow_all):
f = tmp_path / "big.txt"
f.write_text("\n".join(f"line {i}" for i in range(1, 801)))
result = await reader.execute({"action": "read", "path": str(f)})
assert result.success
assert "lines 1–500 of 800" in result.output
assert "[more: call read with offset=501]" in result.output
async def test_read_numbered_false_has_no_footer(self, reader, tmp_path, allow_all):
f = tmp_path / "big.txt"
f.write_text("\n".join(f"line {i}" for i in range(1, 801)))
result = await reader.execute(
{"action": "read", "path": str(f), "numbered": False}
)
assert result.success
assert "line 1\n" in result.output
assert "[more:" not in result.output
async def test_read_binary_rejected(self, reader, tmp_path, allow_all):
f = tmp_path / "bin.dat"
f.write_bytes(b"abc\x00def")
result = await reader.execute({"action": "read", "path": str(f)})
assert not result.success
assert result.error == "binary_file"
async def test_list_dir(self, reader, tmp_path, allow_all):
(tmp_path / "a.txt").write_text("a")
(tmp_path / "b.txt").write_text("b")
result = await reader.execute({"action": "list", "path": str(tmp_path)})
assert result.success
assert "a.txt" in result.output
assert "b.txt" in result.output
async def test_exists_true_and_false(self, reader, tmp_path, allow_all):
f = tmp_path / "exists.txt"
f.write_text("")
assert (await reader.execute({"action": "exists", "path": str(f)})).output == "true"
assert (
await reader.execute({"action": "exists", "path": str(tmp_path / "no.txt")})
).output == "false"
async def test_info_file(self, reader, tmp_path, allow_all):
f = tmp_path / "info.txt"
f.write_text("content")
result = await reader.execute({"action": "info", "path": str(f)})
assert result.success
assert "info.txt" in result.output
async def test_find(self, reader, tmp_path, allow_all):
(tmp_path / "src").mkdir()
(tmp_path / "src" / "a.py").write_text("")
(tmp_path / "src" / "b.txt").write_text("")
result = await reader.execute(
{"action": "find", "path": str(tmp_path), "pattern": "*.py"}
)
assert result.success
assert "src/a.py" in result.output
assert "b.txt" not in result.output
async def test_find_up(self, reader, tmp_path, allow_all):
(tmp_path / "pyproject.toml").write_text("")
nested = tmp_path / "a" / "b"
nested.mkdir(parents=True)
result = await reader.execute({"action": "find_up", "path": str(nested), "name": "pyproject.toml"})
assert result.success
assert str(tmp_path / "pyproject.toml") in result.output
async def test_grep_file(self, reader, tmp_path, allow_all):
f = tmp_path / "grep.txt"
f.write_text("apple\nbanana\napricot\n")
result = await reader.execute({"action": "grep", "path": str(f), "pattern": "ap"})
assert result.success
assert "apple" in result.output
assert "apricot" in result.output
assert "banana" not in result.output
async def test_grep_regex(self, reader, tmp_path, allow_all):
f = tmp_path / "grep.txt"
f.write_text("cat 123\ndog 456\ncat 789\n")
result = await reader.execute(
{"action": "grep", "path": str(f), "pattern": r"cat\s+\d+", "regex": True}
)
assert result.success
assert "cat 123" in result.output
assert "cat 789" in result.output
assert "dog 456" not in result.output
async def test_grep_no_match(self, reader, tmp_path, allow_all):
f = tmp_path / "grep.txt"
f.write_text("hello\nworld\n")
result = await reader.execute({"action": "grep", "path": str(f), "pattern": "xyz"})
assert result.success
assert "No matches" in result.output
assert "regex=true" in result.output
async def test_grep_missing_pattern(self, reader, tmp_path, allow_all):
f = tmp_path / "grep.txt"
f.write_text("hello\n")
result = await reader.execute({"action": "grep", "path": str(f)})
assert not result.success
assert "pattern" in result.output.lower()
async def test_grep_skips_ignored_dirs_by_default(self, reader, tmp_path, allow_all):
(tmp_path / "node_modules").mkdir()
(tmp_path / "node_modules" / "x.js").write_text("needle")
(tmp_path / "src").mkdir()
(tmp_path / "src" / "x.py").write_text("needle")
result = await reader.execute({"action": "grep", "path": str(tmp_path), "pattern": "needle"})
assert "src/x.py" in result.output
assert "node_modules" not in result.output
result = await reader.execute(
{"action": "grep", "path": str(tmp_path), "pattern": "needle", "include_ignored": True}
)
assert "node_modules/x.js" in result.output
async def test_grep_context(self, reader, tmp_path, allow_all):
f = tmp_path / "grep.txt"
f.write_text("before\nneedle\nafter\n")
result = await reader.execute(
{"action": "grep", "path": str(f), "pattern": "needle", "context": 1}
)
assert "grep.txt:2: needle" in result.output
assert "grep.txt-1- before" in result.output
assert "grep.txt-3- after" in result.output
async def test_diff_identical(self, reader, tmp_path, allow_all):
a = tmp_path / "a.txt"
a.write_text("same\ncontent\n")
b = tmp_path / "b.txt"
b.write_text("same\ncontent\n")
result = await reader.execute({"action": "diff", "path": str(a), "other": str(b)})
assert result.success
assert "identical" in result.output.lower()
async def test_diff_different(self, reader, tmp_path, allow_all):
a = tmp_path / "a.txt"
a.write_text("line1\nline2\n")
b = tmp_path / "b.txt"
b.write_text("line1\nmodified\n")
result = await reader.execute({"action": "diff", "path": str(a), "other": str(b)})
assert result.success
assert "---" in result.output
assert "+++" in result.output
assert "modified" in result.output
async def test_diff_includes_real_line_numbers(self, reader, tmp_path, allow_all):
a = tmp_path / "a.txt"
a.write_text("line1\nline2\n")
b = tmp_path / "b.txt"
b.write_text("line1\nmodified\n")
result = await reader.execute({"action": "diff", "path": str(a), "other": str(b)})
assert " 1│ line1" in result.output
assert "- 2│ line2" in result.output
assert "+ 2│ modified" in result.output
assert "@@ -1,2 +1,2 @@" in result.output
async def test_diff_missing_other(self, reader, tmp_path, allow_all):
a = tmp_path / "a.txt"
a.write_text("data")
result = await reader.execute({"action": "diff", "path": str(a)})
assert not result.success
assert "other" in result.output.lower()
async def test_diff_directory_rejected(self, reader, tmp_path, allow_all):
a = tmp_path / "a.txt"
a.write_text("data")
d = tmp_path / "dir"
d.mkdir()
result = await reader.execute({"action": "diff", "path": str(a), "other": str(d)})
assert not result.success
assert "directories" in result.output.lower()
async def test_read_action_redirects_write(self, reader, tmp_path, allow_all):
result = await reader.execute(
{"action": "write", "path": str(tmp_path / "x.txt"), "content": "x"}
)
assert not result.success
assert result.error == "wrong_tool"
assert "filesystem_write" in result.output
class TestFilesystemWrite:
async def test_write_file(self, writer, tmp_path, allow_all):
f = tmp_path / "write.txt"
result = await writer.execute({"action": "write", "path": str(f), "content": "data"})
assert result.success
assert f.read_text() == "data"
async def test_write_empty_overwrite_needs_force(self, writer, tmp_path, allow_all):
f = tmp_path / "write.txt"
f.write_text("data")
result = await writer.execute({"action": "write", "path": str(f), "content": ""})
assert not result.success
assert result.error == "empty_overwrite"
assert f.read_text() == "data"
async def test_write_shrink_warns(self, writer, tmp_path, allow_all):
f = tmp_path / "write.txt"
f.write_text("\n".join("x" for _ in range(10)))
result = await writer.execute({"action": "write", "path": str(f), "content": "x\n"})
assert result.success
assert "much shorter" in result.output
async def test_append_file(self, writer, tmp_path, allow_all):
f = tmp_path / "append.txt"
f.write_text("first")
result = await writer.execute(
{"action": "append", "path": str(f), "content": "-second"}
)
assert result.success
assert f.read_text() == "first-second"
async def test_edit_replaces_exact_unique_text(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha\nbeta\ngamma\n")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "beta\n", "new": "BETA\n"}
)
assert result.success
assert f.read_text() == "alpha\nBETA\ngamma\n"
assert "replaced" in result.output
assert "2: BETA" in result.output
assert "- 2│ beta" in result.metadata["diff"]
assert "+ 2│ BETA" in result.metadata["diff"]
async def test_edit_requires_old_text(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha")
result = await writer.execute({"action": "edit", "path": str(f), "new": "beta"})
assert not result.success
assert result.error == "missing_old"
assert f.read_text() == "alpha"
async def test_edit_rejects_empty_old_text(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "", "new": "beta"}
)
assert not result.success
assert result.error == "empty_old"
assert f.read_text() == "alpha"
async def test_edit_rejects_missing_match(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha\nbeta\n")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "betta", "new": "gamma"}
)
assert not result.success
assert result.error == "old_not_found"
assert "Closest lines" in result.output
assert f.read_text() == "alpha\nbeta\n"
async def test_edit_detects_numbered_copy(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha\n")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "1: alpha", "new": "beta"}
)
assert not result.success
assert result.error == "old_not_found"
assert "line-number column" in result.output
async def test_edit_rejects_non_unique_match(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha beta beta")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "beta", "new": "BETA"}
)
assert not result.success
assert result.error == "old_not_unique"
assert "lines: 1" in result.output
assert f.read_text() == "alpha beta beta"
async def test_edit_replace_all(self, writer, tmp_path, allow_all):
f = tmp_path / "edit.txt"
f.write_text("alpha beta beta")
result = await writer.execute(
{"action": "edit", "path": str(f), "old": "beta", "new": "BETA", "replace_all": True}
)
assert result.success
assert f.read_text() == "alpha BETA BETA"
async def test_edit_lines(self, writer, tmp_path, allow_all):
f = tmp_path / "lines.txt"
f.write_text("a\nb\nc\n")
result = await writer.execute(
{
"action": "edit_lines",
"path": str(f),
"operations": [{"op": "replace", "start": 2, "end": 2, "content": "B"}],
}
)
assert result.success
assert f.read_text() == "a\nB\nc\n"
assert "Applied 1 operation" in result.output
async def test_edit_lines_overlap_rejected(self, writer, tmp_path, allow_all):
f = tmp_path / "lines.txt"
f.write_text("a\nb\nc\n")
result = await writer.execute(
{
"action": "edit_lines",
"path": str(f),
"operations": [
{"op": "replace", "start": 1, "end": 2, "content": "x"},
{"op": "replace", "start": 2, "end": 3, "content": "y"},
],
}
)
assert not result.success
assert result.error == "invalid_ops"
assert "overlaps" in result.output
async def test_insert(self, writer, tmp_path, allow_all):
f = tmp_path / "insert.txt"
f.write_text("a\nc\n")
result = await writer.execute(
{"action": "insert", "path": str(f), "after": 1, "content": "b"}
)
assert result.success
assert f.read_text() == "a\nb\nc\n"
async def test_copy_file(self, writer, tmp_path, allow_all):
src = tmp_path / "src.txt"
src.write_text("copy me")
dst = tmp_path / "dst.txt"
result = await writer.execute(
{"action": "copy", "path": str(src), "destination": str(dst)}
)
assert result.success
assert dst.read_text() == "copy me"
async def test_copy_refuses_existing_without_overwrite(self, writer, tmp_path, allow_all):
src = tmp_path / "src.txt"
src.write_text("new")
dst = tmp_path / "dst.txt"
dst.write_text("old")
result = await writer.execute(
{"action": "copy", "path": str(src), "destination": str(dst)}
)
assert not result.success
assert result.error == "destination_exists"
assert dst.read_text() == "old"
async def test_copy_overwrite(self, writer, tmp_path, allow_all):
src = tmp_path / "src.txt"
src.write_text("new")
dst = tmp_path / "dst.txt"
dst.write_text("old")
result = await writer.execute(
{"action": "copy", "path": str(src), "destination": str(dst), "overwrite": True}
)
assert result.success
assert dst.read_text() == "new"
async def test_copy_missing_source(self, writer, tmp_path, allow_all):
result = await writer.execute(
{"action": "copy", "path": str(tmp_path / "missing.txt"), "destination": str(tmp_path / "dst.txt")}
)
assert not result.success
assert "not found" in result.output.lower()
async def test_move(self, writer, tmp_path, allow_all):
src = tmp_path / "a.txt"
src.write_text("x")
dst = tmp_path / "dir" / "b.txt"
result = await writer.execute(
{"action": "move", "path": str(src), "destination": str(dst)}
)
assert result.success
assert not src.exists()
assert dst.read_text() == "x"
async def test_delete_file(self, writer, tmp_path, allow_all):
f = tmp_path / "del.txt"
f.write_text("bye")
result = await writer.execute({"action": "delete", "path": str(f)})
assert result.success
assert not f.exists()
async def test_delete_non_empty_dir_guard(self, writer, tmp_path, allow_all):
d = tmp_path / "d"
d.mkdir()
(d / "x.txt").write_text("x")
result = await writer.execute({"action": "delete", "path": str(d)})
assert not result.success
assert result.error == "directory_not_empty"
assert d.exists()
async def test_delete_recursive(self, writer, tmp_path, allow_all):
d = tmp_path / "d"
d.mkdir()
(d / "x.txt").write_text("x")
result = await writer.execute({"action": "delete", "path": str(d), "recursive": True})
assert result.success
assert not d.exists()
async def test_mkdir_parents(self, writer, tmp_path, allow_all):
d = tmp_path / "a" / "b" / "c"
result = await writer.execute({"action": "mkdir", "path": str(d)})
assert result.success
assert d.is_dir()
async def test_write_action_redirects_read(self, writer, tmp_path, allow_all):
result = await writer.execute({"action": "grep", "path": str(tmp_path), "pattern": "x"})
assert not result.success
assert result.error == "wrong_tool"
assert "filesystem_read" in result.output
async def test_smart_edit_redirects_to_file_ai(self, writer, tmp_path, allow_all):
result = await writer.execute(
{"action": "smart_edit", "path": str(tmp_path / "x.txt"), "instruction": "x"}
)
assert not result.success
assert result.error == "wrong_tool"
assert "file_ai" in result.output
class _FakeAI:
def __init__(self, answer="answer", ops=None):
self.answer = answer
self.ops = ops or []
self.ask_calls = []
self.ask_json_calls = []
async def ask(self, system, user):
self.ask_calls.append((system, user))
return self.answer
async def ask_json(self, system, user):
self.ask_json_calls.append((system, user))
return self.ops
class TestFileAi:
async def test_without_ai_helper(self, tmp_path, allow_all):
f = tmp_path / "x.txt"
f.write_text("hello")
result = await FileAiTool().execute(
{"action": "ask", "path": str(f), "question": "what?"}
)
assert not result.success
assert result.error == "no_ai_helper"
async def test_ask(self, tmp_path, allow_all):
f = tmp_path / "x.txt"
f.write_text("hello")
ai = _FakeAI(answer="it says hello")
result = await FileAiTool(ai_helper=ai).execute(
{"action": "ask", "path": str(f), "question": "what?"}
)
assert result.success
assert result.output == "it says hello"
assert "hello" in ai.ask_calls[0][1]
async def test_edit(self, tmp_path, allow_all):
f = tmp_path / "x.txt"
f.write_text("alpha\nbeta\n")
ai = _FakeAI(ops=[{"op": "replace", "start": 2, "end": 2, "content": "BETA"}])
result = await FileAiTool(ai_helper=ai).execute(
{"action": "edit", "path": str(f), "instruction": "uppercase beta"}
)
assert result.success
assert f.read_text() == "alpha\nBETA\n"
assert "Applied 1 operation" in result.output
async def test_deterministic_action_redirects(self, tmp_path, allow_all):
ai = _FakeAI()
result = await FileAiTool(ai_helper=ai).execute(
{"action": "read", "path": str(tmp_path / "x.txt")}
)
assert not result.success
assert result.error == "wrong_tool"
assert "filesystem_read" in result.output
class TestNumberedDiff:
def test_unified_diff_has_no_blank_lines_between_content_lines(self, tmp_path):
from pathlib import Path
from navi.tools.filesystem import _unified_diff
out = _unified_diff(
["import os", "OLD = 1", "", "def f():"],
["import os", "NEW = 2", "", "def f():"],
Path("f.py"),
)
assert "\n\n" not in out
assert " 1│ import os" in out
assert "- 2│ OLD = 1" in out
assert "+ 2│ NEW = 2" in out
assert " 3│ " in out or " 3│ " in out
def test_number_diff_strips_trailing_newline_from_content(self):
from navi.tools.filesystem import _number_diff
diff_lines = [
"--- a/f.py\n",
"+++ b/f.py\n",
"@@ -1,3 +1,3 @@",
" import os\n",
"-OLD = 1\n",
"+NEW = 2\n",
" def f():\n",
]
out = "\n".join(_number_diff(diff_lines))
assert "\n\n" not in out
assert out.endswith("def f():")
assert " 1│ import os" in out
assert "- 2│ OLD = 1" in out
assert "+ 2│ NEW = 2" in out
def test_number_diff_noop_for_lines_without_trailing_newline(self):
from navi.tools.filesystem import _number_diff
diff_lines = [
"--- a/f.py",
"+++ b/f.py",
"@@ -1,3 +1,3 @@",
" import os",
"-OLD = 1",
"+NEW = 2",
" def f():",
]
out = "\n".join(_number_diff(diff_lines))
assert "\n\n" not in out
assert out.endswith("def f():")
assert " 1│ import os" in out
assert "- 2│ OLD = 1" in out
assert "+ 2│ NEW = 2" in out
