wehub-resource-sync
c3bf08ac8d
K8s Workspace Integration Tests / k8s-workspace-tests (push) Has been cancelled
Pre-commit / run (ubuntu-latest) (push) Has been cancelled
Python Unittest Coverage / test (macos-15, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (ubuntu-latest, 3.11) (push) Has been cancelled
Python Unittest Coverage / test (windows-latest, 3.11) (push) Has been cancelled
Web UI / check (push) Has been cancelled
255 lines
8.1 KiB
Python
255 lines
8.1 KiB
Python
# -*- 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)
|