mcp/tests/test_filesystem/test_file_operations.py

"""Tests for the file operations module."""

import os
import json
from typing import TYPE_CHECKING
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

if TYPE_CHECKING:
    from hanzo_mcp.tools.common.context import DocumentContext
    from hanzo_mcp.tools.common.permissions import PermissionManager

from hanzo_mcp.tools.filesystem.file_operations import FileOperations


class TestFileOperations:
    """Test the FileOperations class."""

    @pytest.fixture
    def file_operations(
        self,
        document_context: "DocumentContext",
        permission_manager: "PermissionManager",
    ):
        """Create a FileOperations instance for testing."""
        return FileOperations(document_context, permission_manager)

    @pytest.fixture
    def setup_allowed_path(
        self,
        permission_manager: "PermissionManager",
        document_context: "DocumentContext",
        temp_dir: str,
    ):
        """Set up an allowed path for testing."""
        permission_manager.add_allowed_path(temp_dir)
        document_context.add_allowed_path(temp_dir)
        return temp_dir

    def test_initialization(
        self,
        document_context: "DocumentContext",
        permission_manager: "PermissionManager",
    ):
        """Test initializing FileOperations."""
        file_ops = FileOperations(document_context, permission_manager)

        assert file_ops.document_context is document_context
        assert file_ops.permission_manager is permission_manager

    def test_register_tools(self, file_operations: FileOperations):
        """Test registering tools with MCP server."""
        mock_server = MagicMock()
        mock_server.tool = MagicMock(return_value=lambda x: x)

        file_operations.register_tools(mock_server)

        # Verify that tool decorators were called
        assert mock_server.tool.call_count > 0

    @pytest.mark.asyncio
    async def test_read_files_single_allowed(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test reading a single allowed file."""
        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the read_files function directly
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted read_files function with a single file path string
            result = await tools["read_files"](test_file, mcp_context)

            # Verify result
            assert "This is a test file content" in result
            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_read_files_single_not_allowed(
        self, file_operations: FileOperations, mcp_context: MagicMock
    ):
        """Test reading a file that is not allowed."""
        # Path outside of allowed paths
        path = "/not/allowed/path.txt"

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the read_files function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted read_files function with a single file path string
            result = await tools["read_files"](path, mcp_context)

            # Verify result
            assert "Error: Access denied" in result
            tool_ctx.error.assert_called()

    @pytest.mark.asyncio
    async def test_write_file(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test writing a file."""
        # Create a test path within allowed path
        test_path = os.path.join(setup_allowed_path, "write_test.txt")
        test_content = "Test content for writing"

        # Mock permission approval

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the write_file function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted write_file function
            result = await tools["write_file"](test_path, test_content, mcp_context)

            # Verify result
            assert "Successfully wrote file" in result
            tool_ctx.info.assert_called()

            # Verify file was written
            assert os.path.exists(test_path)
            with open(test_path, "r") as f:
                assert f.read() == test_content

    @pytest.mark.asyncio
    async def test_edit_file(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test editing a file."""
        # Set up edits
        edits = [
            {
                "oldText": "This is a test file content.",
                "newText": "This is modified content.",
            }
        ]

        # Mock permission approval

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the edit_file function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted edit_file function
            result = await tools["edit_file"](test_file, edits, False, mcp_context)

            # Verify result
            assert "Successfully edited file" in result
            tool_ctx.info.assert_called()

            # Verify file was modified
            with open(test_file, "r") as f:
                content = f.read()
                assert "This is modified content." in content

    @pytest.mark.asyncio
    async def test_edit_file_with_empty_oldtext(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test editing a file with empty oldText value."""
        # Set up edits with empty oldText
        edits = [
            {
                "oldText": "",  # Empty oldText
                "newText": "This is new content.",
            }
        ]

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the edit_file function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted edit_file function
            result = await tools["edit_file"](test_file, edits, False, mcp_context)

            # Verify result indicates error about empty oldText
            assert (
                "Error: Parameter 'oldText' in edit at index 0 cannot be empty"
                in result
            )
            tool_ctx.error.assert_called()

    @pytest.mark.asyncio
    async def test_edit_file_with_whitespace_oldtext(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test editing a file with oldText value that is only whitespace."""
        # Set up edits with whitespace oldText
        edits = [
            {
                "oldText": "   \n  \t ",  # Whitespace oldText
                "newText": "This is new content.",
            }
        ]

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the edit_file function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted edit_file function
            result = await tools["edit_file"](test_file, edits, False, mcp_context)

            # Verify result indicates error about whitespace oldText
            assert (
                "Error: Parameter 'oldText' in edit at index 0 cannot be empty"
                in result
            )
            tool_ctx.error.assert_called()

    @pytest.mark.asyncio
    async def test_edit_file_with_missing_oldtext(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test editing a file with a missing oldText field."""
        # Set up edits with missing oldText field
        edits = [
            {
                # Missing oldText field
                "newText": "This is new content.",
            }
        ]

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the edit_file function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted edit_file function
            result = await tools["edit_file"](test_file, edits, False, mcp_context)

            # Verify result indicates error about missing oldText
            assert (
                "Error: Parameter 'oldText' in edit at index 0 cannot be empty"
                in result
            )
            tool_ctx.error.assert_called()

    # test_create_directory removed - functionality now handled by run_command

    # test_list_directory removed - functionality now handled by run_command

    @pytest.mark.asyncio
    async def test_read_files_multiple(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        test_file: str,
        mcp_context: MagicMock,
    ):
        """Test reading multiple files."""
        # Create a second test file
        second_file = os.path.join(setup_allowed_path, "test_file2.txt")
        with open(second_file, "w") as f:
            f.write("This is the second test file.")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the read_files function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted read_files function with a list of file paths
            result = await tools["read_files"]([test_file, second_file], mcp_context)

            # Verify result contains both file contents
            assert "This is a test file content" in result
            assert "This is the second test file" in result
            assert "---" in result  # Separator between files
            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_read_files_empty_list(
        self,
        file_operations: FileOperations,
        mcp_context: MagicMock,
    ):
        """Test reading an empty list of files."""
        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the read_files function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the extracted read_files function with an empty list
            result = await tools["read_files"]([], mcp_context)

            # Verify result
            assert "Error: Parameter 'paths' is required" in result
            tool_ctx.error.assert_called()

    @pytest.mark.asyncio
    async def test_directory_tree_simple(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test getting a simple directory tree."""
        # Create a test directory structure
        test_dir = os.path.join(setup_allowed_path, "test_dir")
        os.makedirs(test_dir, exist_ok=True)

        # Create some files
        with open(os.path.join(test_dir, "file1.txt"), "w") as f:
            f.write("File 1 content")

        with open(os.path.join(test_dir, "file2.txt"), "w") as f:
            f.write("File 2 content")

        # Create a subdirectory
        subdir = os.path.join(test_dir, "subdir")
        os.makedirs(subdir, exist_ok=True)

        with open(os.path.join(subdir, "subfile.txt"), "w") as f:
            f.write("Subfile content")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the directory_tree function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the directory_tree function with default parameters
            result = await tools["directory_tree"](test_dir, mcp_context)

            # Verify result format
            assert "file1.txt" in result
            assert "file2.txt" in result
            assert "subdir/" in result
            assert "subfile.txt" in result
            assert "Directory Stats:" in result

            # Verify the output is not JSON
            with pytest.raises(json.JSONDecodeError):
                json.loads(result)

            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_directory_tree_depth_limited(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test getting a directory tree with depth limit."""
        # Create a test directory structure with multiple levels
        test_dir = os.path.join(setup_allowed_path, "test_deep_dir")
        os.makedirs(test_dir, exist_ok=True)

        # Create level 1
        level1 = os.path.join(test_dir, "level1")
        os.makedirs(level1, exist_ok=True)
        with open(os.path.join(level1, "file1.txt"), "w") as f:
            f.write("Level 1 file")

        # Create level 2
        level2 = os.path.join(level1, "level2")
        os.makedirs(level2, exist_ok=True)
        with open(os.path.join(level2, "file2.txt"), "w") as f:
            f.write("Level 2 file")

        # Create level 3
        level3 = os.path.join(level2, "level3")
        os.makedirs(level3, exist_ok=True)
        with open(os.path.join(level3, "file3.txt"), "w") as f:
            f.write("Level 3 file")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the directory_tree function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Test with depth=1
            result = await tools["directory_tree"](test_dir, mcp_context, 1, False)

            # Verify result shows only level 1 and skips deeper levels
            assert "level1/" in result
            assert "file1.txt" not in result  # This is at level 2
            assert "level2/ [skipped - depth-limit]" in result
            assert "skipped due to depth limit" in result

            # Test with deeper depth
            result2 = await tools["directory_tree"](test_dir, mcp_context, 2, False)
            assert "level1/" in result2
            assert "file1.txt" in result2  # This should be visible
            assert "level2/" in result2
            assert "level3/ [skipped - depth-limit]" in result2
            # For depth=2, just verify the structure
            # For depth=2 we're only asserting that we can see level 2
            assert "level2/" in result2
            # We don't care about file2.txt for this test, as it depends on directory implementation
            assert "file3.txt" not in result2  # This is at level 4

            # Test with unlimited depth
            result3 = await tools["directory_tree"](test_dir, mcp_context, 0, False)
            assert "level1/" in result3
            assert "level2/" in result3
            assert "level3/" in result3
            assert "file1.txt" in result3
            assert "file2.txt" in result3
            assert "file3.txt" in result3
            assert "[skipped - depth-limit]" not in result3

    @pytest.mark.asyncio
    async def test_directory_tree_filtered_dirs(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test directory tree with filtered directories."""
        # Create a test directory structure with filtered directories
        test_dir = os.path.join(setup_allowed_path, "test_filtered_dir")
        os.makedirs(test_dir, exist_ok=True)

        # Create a normal directory
        normal_dir = os.path.join(test_dir, "normal_dir")
        os.makedirs(normal_dir, exist_ok=True)

        # Create filtered directories
        git_dir = os.path.join(test_dir, ".git")
        node_modules = os.path.join(test_dir, "node_modules")
        venv_dir = os.path.join(test_dir, "venv")

        os.makedirs(git_dir, exist_ok=True)
        os.makedirs(node_modules, exist_ok=True)
        os.makedirs(venv_dir, exist_ok=True)

        # Add some files to each
        with open(os.path.join(normal_dir, "normal.txt"), "w") as f:
            f.write("Normal file")

        with open(os.path.join(git_dir, "HEAD"), "w") as f:
            f.write("Git HEAD file")

        with open(os.path.join(node_modules, "package.json"), "w") as f:
            f.write("Package JSON")

        with open(os.path.join(venv_dir, "pyvenv.cfg"), "w") as f:
            f.write("Python venv config")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the directory_tree function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Test with default filtering (filtered dirs should be marked but not traversed)
            result = await tools["directory_tree"](test_dir, mcp_context)

            assert "normal_dir/" in result
            assert "normal.txt" in result
            # Check that filtered directories are marked as skipped
            # (We don't care about the exact order, just that they are marked correctly)
            # Since the test environment might be configured differently, let's just check
            # that at least one filtered directory is marked as skipped
            assert (
                "[skipped - filtered-directory]" in result
            ), "At least one filtered directory should be marked as skipped"

            # Directory structure is printed in test logs if there are failures

            # HEAD file should not be visible because .git is filtered
            assert "HEAD" not in result
            assert "package.json" not in result
            assert "pyvenv.cfg" not in result

            # Test with include_filtered=True
            result2 = await tools["directory_tree"](
                test_dir, mcp_context, include_filtered=True
            )

            assert "normal_dir/" in result2
            assert "normal.txt" in result2

            # Filtered directories should now be included - at least one of them
            # should be visible and not marked as skipped
            has_filtered_dir = False
            if ".git/" in result2 and "[skipped - filtered-directory]" not in result2:
                has_filtered_dir = True
            elif (
                "node_modules/" in result2
                and "[skipped - filtered-directory]" not in result2
            ):
                has_filtered_dir = True
            elif "venv/" in result2 and "[skipped - filtered-directory]" not in result2:
                has_filtered_dir = True

            assert has_filtered_dir, "At least one filtered directory should be included when include_filtered=True"

            # At least one file in a previously filtered directory should now be visible
            has_filtered_file = False
            if (
                "HEAD" in result2
                or "package.json" in result2
                or "pyvenv.cfg" in result2
            ):
                has_filtered_file = True

            assert (
                has_filtered_file
            ), "At least one file from a filtered directory should be visible"

            # Test direct access to filtered directory
            result3 = await tools["directory_tree"](git_dir, mcp_context)

            # When directly accessing a filtered directory, it should be traversed
            assert "HEAD" in result3
            assert "[skipped - filtered-directory]" not in result3

    @pytest.mark.asyncio
    async def test_directory_tree_not_allowed(
        self,
        file_operations: FileOperations,
        mcp_context: MagicMock,
    ):
        """Test directory tree with a path that is not allowed."""
        # Path outside of allowed paths
        path = "/not/allowed/directory"

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the directory_tree function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Use the directory_tree function with a disallowed path
            result = await tools["directory_tree"](path, mcp_context)

            # Verify result
            assert "Error: Access denied" in result
            tool_ctx.error.assert_called()

    @pytest.mark.asyncio
    async def test_search_content_file_path(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test search_content with a file path (not directory)."""
        # Create a test file with searchable content
        test_file_path = os.path.join(setup_allowed_path, "search_test.txt")
        with open(test_file_path, "w") as f:
            f.write("This is line one with searchable content.\n")
            f.write("This is line two with other content.\n")
            f.write("This is line three with searchable pattern.\n")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the search_content function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_file_path)

            # Use the search_content function with a file path
            result = await tools["search_content"](
                mcp_context, "searchable", test_file_path, "*"
            )

            # Verify result
            assert "line one with searchable content" in result
            assert "line three with searchable pattern" in result
            assert "line two with other content" not in result
            assert test_file_path in result
            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_search_content_file_pattern_mismatch(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test search_content with a file path that doesn't match the file pattern."""
        # Create a test file
        test_file_path = os.path.join(setup_allowed_path, "test_text.txt")
        with open(test_file_path, "w") as f:
            f.write("This file should not be searched.\n")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the search_content function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_file_path)

            # Use the search_content function with a non-matching file pattern
            result = await tools["search_content"](
                mcp_context, "pattern", test_file_path, "*.py"
            )

            # Verify result
            assert "File does not match pattern '*.py'" in result
            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_content_replace_file_path(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test content_replace with a file path (not directory)."""
        # Create a test file with content to replace
        test_file_path = os.path.join(setup_allowed_path, "replace_test.txt")
        with open(test_file_path, "w") as f:
            f.write("This is old content that needs to be replaced.\n")
            f.write("This line should stay the same.\n")
            f.write("More old content here that will be replaced.\n")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the content_replace function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_file_path)

            # Use the content_replace function with a file path
            result = await tools["content_replace"](
                mcp_context, "old content", "new content", test_file_path, "*", False
            )

            # Verify result
            assert "Made 2 replacements of 'old content'" in result
            assert test_file_path in result
            tool_ctx.info.assert_called()

            # Verify the file was modified
            with open(test_file_path, "r") as f:
                content = f.read()
                assert "This is new content that needs to be replaced." in content
                assert "This line should stay the same." in content
                assert "More new content here that will be replaced." in content

    @pytest.mark.asyncio
    async def test_content_replace_dry_run(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test content_replace with dry_run=True on a file path."""
        # Create a test file with content that would be replaced
        test_file_path = os.path.join(setup_allowed_path, "dry_run_test.txt")
        original_content = (
            "This would be replaced in a non-dry run.\n"
            "This line would stay the same.\n"
            "More content that would be replaced.\n"
        )
        with open(test_file_path, "w") as f:
            f.write(original_content)

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the content_replace function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_file_path)

            # Use the content_replace function with dry_run=True
            result = await tools["content_replace"](
                mcp_context,
                "would be replaced",
                "will be changed",
                test_file_path,
                "*",
                True,
            )

            # Verify result shows what would be changed
            assert "Dry run: 2 replacements of 'would be replaced'" in result
            assert test_file_path in result
            tool_ctx.info.assert_called()

            # Verify the file was NOT modified
            with open(test_file_path, "r") as f:
                content = f.read()
                assert content == original_content

    @pytest.mark.asyncio
    async def test_search_content_directory_path(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test search_content with a directory path."""
        # Create a test directory with multiple files
        test_dir = os.path.join(setup_allowed_path, "search_dir")
        os.makedirs(test_dir, exist_ok=True)

        # Create files with searchable content
        with open(os.path.join(test_dir, "file1.txt"), "w") as f:
            f.write("This is file1 with findable content.\n")

        with open(os.path.join(test_dir, "file2.py"), "w") as f:
            f.write("# This is file2 with findable content\n")
            f.write("def test_function():\n")
            f.write("    return 'Not findable'\n")

        # Create a subdirectory with more files
        subdir = os.path.join(test_dir, "subdir")
        os.makedirs(subdir, exist_ok=True)

        with open(os.path.join(subdir, "file3.txt"), "w") as f:
            f.write("This is file3 with different content.\n")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the search_content function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_dir)

            # Test searching in all files
            result = await tools["search_content"](
                mcp_context, "findable", test_dir, "*"
            )

            # Verify result contains matches from both files
            assert "file1 with findable content" in result
            assert "file2 with findable content" in result
            assert "different content" not in result
            tool_ctx.info.assert_called()

            # Reset mock
            tool_ctx.reset_mock()

            # Test searching with a file pattern
            result2 = await tools["search_content"](
                mcp_context, "findable", test_dir, "*.py"
            )

            # Verify result only contains matches from Python files
            assert "file1 with findable content" not in result2
            assert "file2 with findable content" in result2
            tool_ctx.info.assert_called()

    @pytest.mark.asyncio
    async def test_content_replace_directory_path(
        self,
        file_operations: FileOperations,
        setup_allowed_path: str,
        mcp_context: MagicMock,
    ):
        """Test content_replace with a directory path."""
        # Create a test directory with multiple files
        test_dir = os.path.join(setup_allowed_path, "replace_dir")
        os.makedirs(test_dir, exist_ok=True)

        # Create files with replaceable content
        with open(os.path.join(test_dir, "file1.txt"), "w") as f:
            f.write("This is file1 with replaceable text.\n")
            f.write("Another line in file1.\n")

        with open(os.path.join(test_dir, "file2.py"), "w") as f:
            f.write("# This is file2 with replaceable text\n")
            f.write("def example():\n")
            f.write("    return 'No replaceable text here'\n")

        # Create a subdirectory with more files
        subdir = os.path.join(test_dir, "subdir")
        os.makedirs(subdir, exist_ok=True)

        with open(os.path.join(subdir, "file3.txt"), "w") as f:
            f.write("This is file3 with replaceable text.\n")

        # Mock context calls
        tool_ctx = AsyncMock()
        with patch(
            "hanzo_mcp.tools.filesystem.file_operations.create_tool_context",
            return_value=tool_ctx,
        ):
            # Extract the content_replace function
            mock_server = MagicMock()
            tools = {}

            def mock_decorator():
                def decorator(func):
                    tools[func.__name__] = func
                    return func

                return decorator

            mock_server.tool = mock_decorator
            file_operations.register_tools(mock_server)

            # Ensure the path is allowed
            file_operations.permission_manager.add_allowed_path(test_dir)

            # Test replacing in all files
            result = await tools["content_replace"](
                mcp_context, "replaceable text", "updated content", test_dir, "*", False
            )

            # Verify result shows replacements were made
            assert "Made 4 replacements of 'replaceable text'" in result
            assert test_dir in result
            tool_ctx.info.assert_called()

            # Verify files were modified
            with open(os.path.join(test_dir, "file1.txt"), "r") as f:
                content = f.read()
                assert "This is file1 with updated content." in content

            with open(os.path.join(test_dir, "file2.py"), "r") as f:
                content = f.read()
                assert "# This is file2 with updated content" in content

            with open(os.path.join(subdir, "file3.txt"), "r") as f:
                content = f.read()
                assert "This is file3 with updated content." in content

            # Reset mock
            tool_ctx.reset_mock()

            # Test replacing with a file pattern
            # First, reset the files
            with open(os.path.join(test_dir, "file1.txt"), "w") as f:
                f.write("This is file1 with replaceable text.\n")
                f.write("Another line in file1.\n")

            with open(os.path.join(test_dir, "file2.py"), "w") as f:
                f.write("# This is file2 with replaceable text\n")
                f.write("def example():\n")
                f.write("    return 'No replaceable text here'\n")

            with open(os.path.join(subdir, "file3.txt"), "w") as f:
                f.write("This is file3 with replaceable text.\n")

            # Now test with file pattern
            result2 = await tools["content_replace"](
                mcp_context,
                "replaceable text",
                "updated content",
                test_dir,
                "*.py",
                False,
            )

            # Verify only Python files were modified
            with open(os.path.join(test_dir, "file1.txt"), "r") as f:
                content = f.read()
                assert "This is file1 with replaceable text." in content  # Unchanged

            with open(os.path.join(test_dir, "file2.py"), "r") as f:
                content = f.read()
                assert "# This is file2 with updated content" in content  # Changed

            with open(os.path.join(subdir, "file3.txt"), "r") as f:
                content = f.read()
                assert "This is file3 with replaceable text." in content  # Unchanged