"""Unit tests for filesystem tool (non-AI operations)."""
import pytest
from navi.tools.filesystem import FilesystemTool, _check_path
from navi.tools._internal.base import ToolResult
class TestCheckPath:
@pytest.fixture(autouse=True)
def _allow_all(self, monkeypatch):
import navi.tools.filesystem as _fs_mod
monkeypatch.setattr(_fs_mod.settings, "fs_allowed_paths", "*")
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 as _fs_mod
allowed = tmp_path / "allowed"
allowed.mkdir()
blocked = tmp_path / "blocked"
blocked.mkdir()
monkeypatch.setattr(_fs_mod.settings, "fs_allowed_paths", str(allowed))
assert _check_path(str(allowed / "file.txt")) is not None
assert _check_path(str(blocked / "file.txt")) is None
class TestFilesystemToolBasic:
@pytest.fixture(autouse=True)
def _allow_all(self, monkeypatch):
import navi.tools.filesystem as _fs_mod
async def _to_thread(func, *args, **kwargs):
return func(*args, **kwargs)
monkeypatch.setattr(_fs_mod.settings, "fs_allowed_paths", "*")
monkeypatch.setattr(_fs_mod.asyncio, "to_thread", _to_thread)
@pytest.fixture
def tool(self):
return FilesystemTool()
async def test_read_file(self, tool, tmp_path):
f = tmp_path / "hello.txt"
f.write_text("world")
result = await tool.execute({"action": "read", "path": str(f)})
assert result.success
assert "world" in result.output
async def test_read_missing(self, tool, tmp_path):
f = tmp_path / "missing.txt"
result = await tool.execute({"action": "read", "path": str(f)})
assert not result.success
assert "not found" in result.output.lower() or "does not exist" in result.output.lower()
async def test_write_file(self, tool, tmp_path):
f = tmp_path / "write.txt"
result = await tool.execute({"action": "write", "path": str(f), "content": "data"})
assert result.success
assert f.read_text() == "data"
async def test_list_dir(self, tool, tmp_path):
(tmp_path / "a.txt").write_text("a")
(tmp_path / "b.txt").write_text("b")
result = await tool.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(self, tool, tmp_path):
f = tmp_path / "exists.txt"
f.write_text("")
result = await tool.execute({"action": "exists", "path": str(f)})
assert result.success
assert "true" in result.output.lower() or "exists" in result.output.lower()
async def test_exists_false(self, tool, tmp_path):
f = tmp_path / "no.txt"
result = await tool.execute({"action": "exists", "path": str(f)})
# exists returns success=True because the check itself succeeded;
# the answer is in the output text
assert result.success
assert "false" in result.output.lower()
async def test_delete_file(self, tool, tmp_path):
f = tmp_path / "del.txt"
f.write_text("bye")
result = await tool.execute({"action": "delete", "path": str(f)})
assert result.success
assert not f.exists()
async def test_info_file(self, tool, tmp_path):
f = tmp_path / "info.txt"
f.write_text("content")
result = await tool.execute({"action": "info", "path": str(f)})
assert result.success
assert "info.txt" in result.output
async def test_append_file(self, tool, tmp_path):
f = tmp_path / "append.txt"
f.write_text("first")
result = await tool.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, tool, tmp_path):
f = tmp_path / "edit.txt"
f.write_text("alpha\nbeta\ngamma\n")
result = await tool.execute({
"action": "edit",
"path": str(f),
"old": "beta\n",
"new": "BETA\n",
})
assert result.success
assert f.read_text() == "alpha\nBETA\ngamma\n"
async def test_edit_requires_old_text(self, tool, tmp_path):
f = tmp_path / "edit.txt"
f.write_text("alpha")
result = await tool.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, tool, tmp_path):
f = tmp_path / "edit.txt"
f.write_text("alpha")
result = await tool.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, tool, tmp_path):
f = tmp_path / "edit.txt"
f.write_text("alpha")
result = await tool.execute({"action": "edit", "path": str(f), "old": "beta", "new": "gamma"})
assert not result.success
assert result.error == "old_not_found"
assert f.read_text() == "alpha"
async def test_edit_rejects_non_unique_match(self, tool, tmp_path):
f = tmp_path / "edit.txt"
f.write_text("alpha beta beta")
result = await tool.execute({"action": "edit", "path": str(f), "old": "beta", "new": "BETA"})
assert not result.success
assert result.error == "old_not_unique"
assert f.read_text() == "alpha beta beta"
async def test_copy_file(self, tool, tmp_path):
src = tmp_path / "src.txt"
src.write_text("copy me")
dst = tmp_path / "dst.txt"
result = await tool.execute({"action": "copy", "path": str(src), "destination": str(dst)})
assert result.success
assert dst.exists()
assert dst.read_text() == "copy me"
async def test_copy_missing_source(self, tool, tmp_path):
src = tmp_path / "missing.txt"
dst = tmp_path / "dst.txt"
result = await tool.execute({"action": "copy", "path": str(src), "destination": str(dst)})
assert not result.success
assert "not found" in result.output.lower()
async def test_grep_file(self, tool, tmp_path):
f = tmp_path / "grep.txt"
f.write_text("apple\nbanana\napricot\n")
result = await tool.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, tool, tmp_path):
f = tmp_path / "grep.txt"
f.write_text("cat 123\ndog 456\ncat 789\n")
result = await tool.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, tool, tmp_path):
f = tmp_path / "grep.txt"
f.write_text("hello\nworld\n")
result = await tool.execute({"action": "grep", "path": str(f), "pattern": "xyz"})
assert result.success
assert "No matches" in result.output
async def test_grep_missing_pattern(self, tool, tmp_path):
f = tmp_path / "grep.txt"
f.write_text("hello\n")
result = await tool.execute({"action": "grep", "path": str(f)})
assert not result.success
assert "pattern" in result.output.lower()
async def test_diff_identical(self, tool, tmp_path):
a = tmp_path / "a.txt"
a.write_text("same\ncontent\n")
b = tmp_path / "b.txt"
b.write_text("same\ncontent\n")
result = await tool.execute({"action": "diff", "path": str(a), "destination": str(b)})
assert result.success
assert "identical" in result.output.lower()
async def test_diff_different(self, tool, tmp_path):
a = tmp_path / "a.txt"
a.write_text("line1\nline2\n")
b = tmp_path / "b.txt"
b.write_text("line1\nmodified\n")
result = await tool.execute({"action": "diff", "path": str(a), "destination": str(b)})
assert result.success
assert "---" in result.output
assert "+++" in result.output
assert "modified" in result.output
async def test_diff_missing_destination(self, tool, tmp_path):
a = tmp_path / "a.txt"
a.write_text("data")
result = await tool.execute({"action": "diff", "path": str(a)})
assert not result.success
assert "destination" in result.output.lower()
async def test_diff_directory_rejected(self, tool, tmp_path):
a = tmp_path / "a.txt"
a.write_text("data")
d = tmp_path / "dir"
d.mkdir()
result = await tool.execute({"action": "diff", "path": str(a), "destination": str(d)})
assert not result.success
assert "directories" in result.output.lower()