Newer
Older
navi-1 / tests / unit / tools / test_filesystem.py
"""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()