agentscope-ai--agentscope/tests/builtin_glob_test.py

# -*- coding: utf-8 -*-
"""Glob tool test case."""
import os
import tempfile
from unittest.async_case import IsolatedAsyncioTestCase

from utils import AnyString
from agentscope.tool import Glob
from agentscope.permission import (
    PermissionContext,
    PermissionBehavior,
    PermissionRule,
)


class GlobToolTest(IsolatedAsyncioTestCase):
    """The glob tool test case."""

    async def asyncSetUp(self) -> None:
        """The async setup method."""
        self.glob_tool = Glob()
        # Create a temporary directory with test files
        self.temp_dir = tempfile.mkdtemp()

        # Create test files
        with open(
            os.path.join(self.temp_dir, "test1.py"),
            "w",
            encoding="utf-8",
        ):
            pass
        with open(
            os.path.join(self.temp_dir, "test2.py"),
            "w",
            encoding="utf-8",
        ):
            pass
        with open(
            os.path.join(self.temp_dir, "test.txt"),
            "w",
            encoding="utf-8",
        ):
            pass

        # Create subdirectory
        sub_dir = os.path.join(self.temp_dir, "subdir")
        os.makedirs(sub_dir)
        with open(os.path.join(sub_dir, "test3.py"), "w", encoding="utf-8"):
            pass

    async def asyncTearDown(self) -> None:
        """Clean up temporary files."""
        import shutil

        if os.path.exists(self.temp_dir):
            shutil.rmtree(self.temp_dir)

    async def test_tool_properties(self) -> None:
        """Test glob tool properties."""
        self.assertEqual(self.glob_tool.name, "Glob")
        self.assertIsInstance(self.glob_tool.description, str)
        self.assertIsInstance(self.glob_tool.input_schema, dict)
        self.assertFalse(self.glob_tool.is_mcp)
        self.assertTrue(self.glob_tool.is_read_only)
        self.assertTrue(self.glob_tool.is_concurrency_safe)

    async def test_check_permissions(self) -> None:
        """Test glob tool permission checking."""
        context = PermissionContext()
        tool_input = {"pattern": "*.py"}
        decision = await self.glob_tool.check_permissions(tool_input, context)

        # Read/Glob/Grep are read-only, return PASSTHROUGH
        self.assertEqual(decision.behavior, PermissionBehavior.PASSTHROUGH)

    async def test_simple_pattern(self) -> None:
        """Test simple glob pattern."""
        chunk = await self.glob_tool(
            pattern="*.py",
            path=self.temp_dir,
        )

        self.assertEqual(chunk.state, "running")

        # Should find test1.py and test2.py
        content = chunk.content[0].text
        self.assertIn("test1.py", content)
        self.assertIn("test2.py", content)
        self.assertNotIn("test.txt", content)

    async def test_recursive_pattern(self) -> None:
        """Test recursive glob pattern."""
        chunk = await self.glob_tool(
            pattern="**/*.py",
            path=self.temp_dir,
        )

        content = chunk.content[0].text

        # Should find all .py files including in subdirectory
        self.assertIn("test1.py", content)
        self.assertIn("test2.py", content)
        self.assertIn("test3.py", content)

    async def test_windows_style_separator_pattern(self) -> None:
        """Test glob patterns that use backslashes as path separators."""
        chunk = await self.glob_tool(
            pattern=r"subdir\*.py",
            path=self.temp_dir,
        )

        self.assertDictEqual(
            chunk.model_dump(),
            {
                "content": [
                    {
                        "type": "text",
                        "text": os.path.join(
                            self.temp_dir,
                            "subdir",
                            "test3.py",
                        ),
                        "id": AnyString(),
                    },
                ],
                "state": "running",
                "is_last": True,
                "metadata": {},
                "id": AnyString(),
            },
        )

    async def test_no_matches(self) -> None:
        """Test pattern with no matches."""
        chunk = await self.glob_tool(
            pattern="*.nonexistent",
            path=self.temp_dir,
        )

        self.assertEqual(chunk.state, "running")
        self.assertIn("No files found", chunk.content[0].text)

    async def test_match_rule_path(self) -> None:
        """Test match_rule with path patterns."""
        # Test matching explicit path
        self.assertTrue(
            await self.glob_tool.match_rule(
                self.temp_dir,
                {"path": self.temp_dir, "pattern": "*.py"},
            ),
        )

        # Test wildcard pattern matching path
        parent_dir = os.path.dirname(self.temp_dir)
        self.assertTrue(
            await self.glob_tool.match_rule(
                parent_dir + "/**",
                {"path": self.temp_dir, "pattern": "*.py"},
            ),
        )

        # Test non-matching path
        self.assertFalse(
            await self.glob_tool.match_rule(
                "/some/other/path/**",
                {"path": self.temp_dir, "pattern": "*.py"},
            ),
        )

    async def test_match_rule_pattern(self) -> None:
        """Test match_rule with pattern matching."""
        # Test matching against the pattern itself
        self.assertTrue(
            await self.glob_tool.match_rule(
                "*.py",
                {"pattern": "*.py"},
            ),
        )

        # Test wildcard pattern matching
        self.assertTrue(
            await self.glob_tool.match_rule(
                "**/*.py",
                {"pattern": "src/**/*.py"},
            ),
        )

        # Test non-matching pattern
        self.assertFalse(
            await self.glob_tool.match_rule(
                "*.txt",
                {"pattern": "*.py"},
            ),
        )

    async def test_match_rule_path_priority(self) -> None:
        """Test that path matching takes priority over pattern matching."""
        # If path matches, should return True even if pattern doesn't
        self.assertTrue(
            await self.glob_tool.match_rule(
                self.temp_dir,
                {"path": self.temp_dir, "pattern": "*.txt"},
            ),
        )

    async def test_generate_suggestions_with_path(self) -> None:
        """Test generate_suggestions for glob with explicit path."""

        suggestions = await self.glob_tool.generate_suggestions(
            {"path": self.temp_dir, "pattern": "*.py"},
        )

        self.assertIsInstance(suggestions, list)
        self.assertGreater(len(suggestions), 0)
        self.assertIsInstance(suggestions[0], PermissionRule)

        # Should suggest directory pattern
        abs_path = os.path.abspath(self.temp_dir)
        expected_pattern = abs_path.rstrip("/") + "/**"
        suggestion_contents = [s.rule_content for s in suggestions]
        self.assertIn(expected_pattern, suggestion_contents)

    async def test_generate_suggestions_defaults_to_cwd(self) -> None:
        """Test generate_suggestions defaults to cwd when no path provided."""

        suggestions = await self.glob_tool.generate_suggestions(
            {"pattern": "*.py"},
        )

        self.assertIsInstance(suggestions, list)
        self.assertGreater(len(suggestions), 0)

        cwd = os.getcwd()
        expected_pattern = os.path.abspath(cwd).rstrip("/") + "/**"
        suggestion_contents = [s.rule_content for s in suggestions]
        self.assertIn(expected_pattern, suggestion_contents)