tenseleyflow/loader / f91124a

+"""Tests for loader."""

+"""Pytest configuration and fixtures."""

+

+import pytest

+import tempfile

+from pathlib import Path

+

+

+@pytest.fixture

+def temp_dir():

+    """Create a temporary directory for tests."""

+    with tempfile.TemporaryDirectory() as tmpdir:

+        yield Path(tmpdir)

+

+

+@pytest.fixture

+def sample_file(temp_dir):

+    """Create a sample file for testing."""

+    file_path = temp_dir / "sample.txt"

+    file_path.write_text("Line 1\nLine 2\nLine 3\n")

+    return file_path

+

+

+@pytest.fixture

+def sample_python_file(temp_dir):

+    """Create a sample Python file for testing."""

+    file_path = temp_dir / "sample.py"

+    file_path.write_text('''"""Sample module."""

+

+def hello():

+    """Say hello."""

+    return "Hello, world!"

+

+def add(a, b):

+    """Add two numbers."""

+    return a + b

+''')

+    return file_path

+"""Tests for the ReAct parsing module."""

+

+import pytest

+from loader.agent.parsing import parse_tool_calls, format_tool_result

+

+

+class TestParseToolCalls:

+    """Tests for parse_tool_calls function."""

+

+    def test_parse_tool_call_xml_style(self):

+        text = '''I need to read the file.

+<tool_call>

+{"name": "read", "arguments": {"file_path": "/tmp/test.txt"}}

+</tool_call>

+'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 1

+        assert result.tool_calls[0].name == "read"

+        assert result.tool_calls[0].arguments == {"file_path": "/tmp/test.txt"}

+        assert not result.is_final_answer

+

+    def test_parse_multiple_tool_calls(self):

+        text = '''<tool_call>

+{"name": "read", "arguments": {"file_path": "a.txt"}}

+</tool_call>

+<tool_call>

+{"name": "read", "arguments": {"file_path": "b.txt"}}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 2

+        assert result.tool_calls[0].arguments["file_path"] == "a.txt"

+        assert result.tool_calls[1].arguments["file_path"] == "b.txt"

+

+    def test_parse_final_answer(self):

+        text = '''Thought: I have all the information needed.

+Final Answer: The file contains a hello world program.'''

+        result = parse_tool_calls(text)

+        assert result.is_final_answer

+        assert len(result.tool_calls) == 0

+        assert "hello world" in result.content.lower()

+

+    def test_parse_no_tool_calls(self):

+        text = "Just some regular text without any tool calls."

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 0

+        assert not result.is_final_answer

+        assert "regular text" in result.content

+

+    def test_parse_bare_json(self):

+        text = '''Let me read that file.

+{"name": "read", "arguments": {"file_path": "/test.txt"}}'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 1

+        assert result.tool_calls[0].name == "read"

+

+    def test_parse_removes_react_labels(self):

+        text = '''Thought: I need to check this.

+Action: <tool_call>

+{"name": "read", "arguments": {"file_path": "test.txt"}}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        assert "Thought:" not in result.content

+        assert "Action:" not in result.content

+

+    def test_parse_invalid_json_ignored(self):

+        text = '''<tool_call>

+{invalid json here}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 0

+

+    def test_parse_empty_arguments(self):

+        text = '''<tool_call>

+{"name": "pwd", "arguments": {}}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 1

+        assert result.tool_calls[0].arguments == {}

+

+    def test_parse_parameters_alias(self):

+        """Test that 'parameters' is accepted as alias for 'arguments'."""

+        text = '''<tool_call>

+{"name": "read", "parameters": {"file_path": "/tmp/test.txt"}}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        assert len(result.tool_calls) == 1

+        assert result.tool_calls[0].name == "read"

+        assert result.tool_calls[0].arguments == {"file_path": "/tmp/test.txt"}

+

+    def test_parse_malformed_closing_tag(self):

+        """Test handling of malformed </tool_call> at start."""

+        text = '''</tool_call> {"name": "read", "parameters": {"file_path": "test.txt"}}

+</tool_call>'''

+        result = parse_tool_calls(text)

+        # Should clean up the malformed tags

+        assert "</tool_call>" not in result.content

+

+    def test_parse_cleans_orphaned_tags(self):

+        """Test that orphaned tool_call tags are removed from content."""

+        text = '''Some text </tool_call> more text <tool_call> end'''

+        result = parse_tool_calls(text)

+        assert "<tool_call>" not in result.content

+        assert "</tool_call>" not in result.content

+

+

+class TestFormatToolResult:

+    """Tests for format_tool_result function."""

+

+    def test_format_success(self):

+        result = format_tool_result("read", "file contents here")

+        assert "Observation" in result

+        assert "read" in result

+        assert "Result" in result

+        assert "file contents here" in result

+

+    def test_format_error(self):

+        result = format_tool_result("write", "Permission denied", is_error=True)

+        assert "Observation" in result

+        assert "write" in result

+        assert "Error" in result

+        assert "Permission denied" in result

+"""Tests for tool implementations."""

+

+import pytest

+from loader.tools import (

+    ReadTool, WriteTool, EditTool, GlobTool,

+    BashTool, GrepTool, ConfirmationRequired,

+)

+from loader.tools.base import ToolRegistry, create_default_registry

+

+

+class TestReadTool:

+    """Tests for ReadTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return ReadTool()

+

+    @pytest.mark.asyncio

+    async def test_read_file(self, tool, sample_file):

+        result = await tool.execute(file_path=str(sample_file))

+        assert not result.is_error

+        assert "Line 1" in result.output

+        assert "Line 2" in result.output

+

+    @pytest.mark.asyncio

+    async def test_read_nonexistent(self, tool, temp_dir):

+        result = await tool.execute(file_path=str(temp_dir / "nonexistent.txt"))

+        assert result.is_error

+        assert "not found" in result.output.lower()

+

+    @pytest.mark.asyncio

+    async def test_read_with_offset(self, tool, sample_file):

+        result = await tool.execute(file_path=str(sample_file), offset=2, limit=1)

+        assert not result.is_error

+        assert "Line 2" in result.output

+        assert "Line 1" not in result.output

+

+    def test_is_not_destructive(self, tool):

+        assert not tool.is_destructive

+

+

+class TestWriteTool:

+    """Tests for WriteTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return WriteTool()

+

+    @pytest.mark.asyncio

+    async def test_write_file(self, tool, temp_dir):

+        file_path = temp_dir / "new_file.txt"

+        result = await tool.execute(file_path=str(file_path), content="Hello, world!")

+        assert not result.is_error

+        assert file_path.exists()

+        assert file_path.read_text() == "Hello, world!"

+

+    @pytest.mark.asyncio

+    async def test_write_creates_parents(self, tool, temp_dir):

+        file_path = temp_dir / "subdir" / "deep" / "file.txt"

+        result = await tool.execute(file_path=str(file_path), content="nested")

+        assert not result.is_error

+        assert file_path.exists()

+

+    def test_is_destructive(self, tool):

+        assert tool.is_destructive

+

+    def test_requires_confirmation(self, tool):

+        with pytest.raises(ConfirmationRequired) as exc_info:

+            tool.check_confirmation(

+                skip_confirmation=False,

+                file_path="/tmp/test.txt",

+                content="test",

+            )

+        assert "Write to file" in exc_info.value.message

+

+    def test_skip_confirmation(self, tool):

+        # Should not raise

+        tool.check_confirmation(

+            skip_confirmation=True,

+            file_path="/tmp/test.txt",

+            content="test",

+        )

+

+

+class TestEditTool:

+    """Tests for EditTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return EditTool()

+

+    @pytest.mark.asyncio

+    async def test_edit_file(self, tool, sample_file):

+        result = await tool.execute(

+            file_path=str(sample_file),

+            old_string="Line 2",

+            new_string="Modified Line 2",

+        )

+        assert not result.is_error

+        assert "Modified Line 2" in sample_file.read_text()

+

+    @pytest.mark.asyncio

+    async def test_edit_nonexistent(self, tool, temp_dir):

+        result = await tool.execute(

+            file_path=str(temp_dir / "nonexistent.txt"),

+            old_string="foo",

+            new_string="bar",

+        )

+        assert result.is_error

+

+    @pytest.mark.asyncio

+    async def test_edit_string_not_found(self, tool, sample_file):

+        result = await tool.execute(

+            file_path=str(sample_file),

+            old_string="Not in file",

+            new_string="replacement",

+        )

+        assert result.is_error

+        assert "not found" in result.output.lower()

+

+

+class TestGlobTool:

+    """Tests for GlobTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return GlobTool()

+

+    @pytest.mark.asyncio

+    async def test_glob_finds_files(self, tool, temp_dir):

+        (temp_dir / "file1.py").write_text("# python")

+        (temp_dir / "file2.py").write_text("# python")

+        (temp_dir / "file3.txt").write_text("text")

+

+        result = await tool.execute(pattern="*.py", path=str(temp_dir))

+        assert not result.is_error

+        assert "file1.py" in result.output

+        assert "file2.py" in result.output

+        assert "file3.txt" not in result.output

+

+    @pytest.mark.asyncio

+    async def test_glob_no_matches(self, tool, temp_dir):

+        result = await tool.execute(pattern="*.xyz", path=str(temp_dir))

+        assert not result.is_error

+        assert "No files matching" in result.output

+

+

+class TestBashTool:

+    """Tests for BashTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return BashTool()

+

+    @pytest.mark.asyncio

+    async def test_bash_simple_command(self, tool):

+        result = await tool.execute(command="echo 'hello world'")

+        assert not result.is_error

+        assert "hello world" in result.output

+

+    @pytest.mark.asyncio

+    async def test_bash_pwd(self, tool):

+        result = await tool.execute(command="pwd")

+        assert not result.is_error

+        assert "/" in result.output

+

+    @pytest.mark.asyncio

+    async def test_bash_failed_command(self, tool):

+        result = await tool.execute(command="exit 1")

+        assert result.is_error

+        assert "Exit code 1" in result.output

+

+    def test_is_destructive(self, tool):

+        assert tool.is_destructive

+

+    def test_safe_command_no_confirmation(self, tool):

+        # ls is safe

+        tool.check_confirmation(skip_confirmation=False, command="ls -la")

+        # git status is safe

+        tool.check_confirmation(skip_confirmation=False, command="git status")

+

+    def test_unsafe_command_requires_confirmation(self, tool):

+        with pytest.raises(ConfirmationRequired):

+            tool.check_confirmation(skip_confirmation=False, command="rm -rf /tmp/test")

+

+

+class TestGrepTool:

+    """Tests for GrepTool."""

+

+    @pytest.fixture

+    def tool(self):

+        return GrepTool()

+

+    @pytest.mark.asyncio

+    async def test_grep_finds_pattern(self, tool, sample_python_file):

+        result = await tool.execute(

+            pattern="def.*hello",

+            path=str(sample_python_file),

+        )

+        assert not result.is_error

+        assert "hello" in result.output

+

+    @pytest.mark.asyncio

+    async def test_grep_no_matches(self, tool, sample_file):

+        result = await tool.execute(

+            pattern="nonexistent_pattern",

+            path=str(sample_file),

+        )

+        assert not result.is_error

+        assert "No matches" in result.output

+

+

+class TestToolRegistry:

+    """Tests for ToolRegistry."""

+

+    def test_create_default_registry(self):

+        registry = create_default_registry()

+        assert registry.get("read") is not None

+        assert registry.get("write") is not None

+        assert registry.get("edit") is not None

+        assert registry.get("glob") is not None

+        assert registry.get("bash") is not None

+        assert registry.get("grep") is not None

+

+    def test_unknown_tool(self):

+        registry = create_default_registry()

+        assert registry.get("nonexistent") is None

+

+    @pytest.mark.asyncio

+    async def test_execute_unknown_tool(self):

+        registry = create_default_registry()

+        result = await registry.execute("nonexistent")

+        assert result.is_error

+        assert "Unknown tool" in result.output

+

+    def test_skip_confirmation_flag(self):

+        registry = create_default_registry()

+        assert not registry.skip_confirmation

+        registry.skip_confirmation = True

+        assert registry.skip_confirmation