# -*- coding: utf-8 -*- """Grep tool test case.""" import os import tempfile from unittest.async_case import IsolatedAsyncioTestCase from agentscope.message import ToolResultState from agentscope.tool import Grep from agentscope.permission import ( PermissionContext, PermissionBehavior, PermissionRule, ) class GrepToolTest(IsolatedAsyncioTestCase): """The grep tool test case.""" async def asyncSetUp(self) -> None: """The async setup method.""" self.grep_tool = Grep() # 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", ) as f: f.write("def hello():\n print('Hello World')\n") with open( os.path.join(self.temp_dir, "test2.py"), "w", encoding="utf-8", ) as f: f.write("def goodbye():\n print('Goodbye')\n") with open( os.path.join(self.temp_dir, "test.txt"), "w", encoding="utf-8", ) as f: f.write("This is a text file\nHello from text\n") # Create subdirectory with files for glob pattern testing subdir = os.path.join(self.temp_dir, "subdir") os.makedirs(subdir) with open( os.path.join(subdir, "nested.py"), "w", encoding="utf-8", ) as f: f.write("def nested():\n print('Nested')\n") 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 grep tool properties.""" self.assertEqual(self.grep_tool.name, "Grep") self.assertIsInstance(self.grep_tool.description, str) self.assertIsInstance(self.grep_tool.input_schema, dict) self.assertFalse(self.grep_tool.is_mcp) self.assertTrue(self.grep_tool.is_read_only) self.assertTrue(self.grep_tool.is_concurrency_safe) async def test_check_permissions(self) -> None: """Test grep tool permission checking.""" context = PermissionContext() tool_input = {"pattern": "hello"} decision = await self.grep_tool.check_permissions(tool_input, context) # Read/Glob/Grep are read-only, return PASSTHROUGH self.assertEqual(decision.behavior, PermissionBehavior.PASSTHROUGH) async def test_simple_search(self) -> None: """Test simple grep search.""" chunk = await self.grep_tool( pattern="Hello", path=self.temp_dir, output_mode="files_with_matches", ) self.assertEqual(chunk.state, ToolResultState.SUCCESS) content = chunk.content[0].text # Should find files containing "Hello" self.assertIn("test1.py", content) self.assertIn("test.txt", content) async def test_content_mode(self) -> None: """Test grep with content output mode.""" chunk = await self.grep_tool( pattern="def", path=self.temp_dir, output_mode="content", type="py", ) content = chunk.content[0].text # Should show matching lines self.assertIn("def hello", content) self.assertIn("def goodbye", content) async def test_case_insensitive(self) -> None: """Test case-insensitive search.""" chunk = await self.grep_tool( pattern="HELLO", path=self.temp_dir, case_insensitive=True, output_mode="files_with_matches", ) content = chunk.content[0].text self.assertIn("test1.py", content) async def test_no_matches(self) -> None: """Test search with no matches.""" chunk = await self.grep_tool( pattern="NonExistentPattern", path=self.temp_dir, ) self.assertIn("No matches found", chunk.content[0].text) async def test_type_filter(self) -> None: """Test filtering by file type.""" chunk = await self.grep_tool( pattern="Hello", path=self.temp_dir, type="py", output_mode="files_with_matches", ) content = chunk.content[0].text # Should only find .py files self.assertIn("test1.py", content) self.assertNotIn("test.txt", content) async def test_invalid_regex(self) -> None: """Test grep with invalid regex pattern.""" chunk = await self.grep_tool( pattern="[invalid(regex", path=self.temp_dir, ) self.assertEqual(chunk.state, "error") # ripgrep returns its own error message for regex parse errors self.assertIn("regex parse error", chunk.content[0].text) async def test_glob_pattern_with_subdirs(self) -> None: """Test glob pattern matching with subdirectories like **/*.py.""" chunk = await self.grep_tool( pattern="def", path=self.temp_dir, glob="**/*.py", output_mode="files_with_matches", ) content = chunk.content[0].text # Should find all .py files including in subdirectories self.assertIn("test1.py", content) self.assertIn("test2.py", content) self.assertIn("nested.py", content) # Should not find .txt files self.assertNotIn("test.txt", content) async def test_match_rule_path(self) -> None: """Test match_rule with search path patterns.""" # Test matching explicit path self.assertTrue( await self.grep_tool.match_rule( self.temp_dir, {"path": self.temp_dir}, ), ) # Test wildcard pattern matching path parent_dir = os.path.dirname(self.temp_dir) self.assertTrue( await self.grep_tool.match_rule( parent_dir + "/**", {"path": self.temp_dir}, ), ) # Test non-matching path self.assertFalse( await self.grep_tool.match_rule( "/some/other/path/**", {"path": self.temp_dir}, ), ) async def test_match_rule_defaults_to_cwd(self) -> None: """Test match_rule defaults to cwd when no path is provided.""" cwd = os.getcwd() # When no path provided, should match against cwd self.assertTrue( await self.grep_tool.match_rule( cwd, {"pattern": "hello"}, ), ) # Should not match a different path self.assertFalse( await self.grep_tool.match_rule( "/some/other/path", {"pattern": "hello"}, ), ) async def test_generate_suggestions_with_path(self) -> None: """Test generate_suggestions for grep with explicit path.""" suggestions = await self.grep_tool.generate_suggestions( {"path": self.temp_dir, "pattern": "hello"}, ) 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.grep_tool.generate_suggestions( {"pattern": "hello"}, ) 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)