Spaces:
Paused
Paused
File size: 6,414 Bytes
4b03eed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 | # -*- coding: utf-8 -*-
"""Read tool test case."""
import os
import tempfile
from unittest.async_case import IsolatedAsyncioTestCase
from agentscope.tool import ToolChunk, Read
from agentscope.permission import (
PermissionContext,
PermissionBehavior,
PermissionRule,
)
from agentscope.message import TextBlock
class ReadToolTest(IsolatedAsyncioTestCase):
"""The read tool test case."""
async def asyncSetUp(self) -> None:
"""The async setup method."""
self.read_tool = Read()
# Create a temporary file for testing
self.temp_file = tempfile.NamedTemporaryFile(
mode="w",
delete=False,
suffix=".txt",
)
# Write multiple lines
for i in range(1, 11):
self.temp_file.write(f"Line {i}\n")
self.temp_file.close()
async def asyncTearDown(self) -> None:
"""Clean up temporary files."""
if os.path.exists(self.temp_file.name):
os.unlink(self.temp_file.name)
async def test_tool_properties(self) -> None:
"""Test read tool properties."""
self.assertEqual(self.read_tool.name, "Read")
self.assertIsInstance(self.read_tool.description, str)
self.assertIsInstance(self.read_tool.input_schema, dict)
self.assertFalse(self.read_tool.is_mcp)
self.assertTrue(self.read_tool.is_read_only)
self.assertTrue(self.read_tool.is_concurrency_safe)
async def test_check_permissions(self) -> None:
"""Test read tool permission checking."""
context = PermissionContext()
tool_input = {"file_path": "/tmp/test.txt"}
decision = await self.read_tool.check_permissions(tool_input, context)
# Read/Glob/Grep are read-only, return PASSTHROUGH
self.assertEqual(decision.behavior, PermissionBehavior.PASSTHROUGH)
async def test_simple_read(self) -> None:
"""Test simple file reading."""
chunk = await self.read_tool(file_path=self.temp_file.name)
self.assertIsInstance(chunk, ToolChunk)
self.assertEqual(chunk.state, "running")
self.assertEqual(len(chunk.content), 1)
self.assertIsInstance(chunk.content[0], TextBlock)
content = chunk.content[0].text
# Should contain all lines with line numbers
self.assertIn("Line 1", content)
self.assertIn("Line 10", content)
async def test_read_with_offset(self) -> None:
"""Test reading with offset."""
chunk = await self.read_tool(
file_path=self.temp_file.name,
offset=5,
)
self.assertEqual(chunk.state, "running")
content = chunk.content[0].text
# Should start from line 5
self.assertIn("Line 5", content)
# Line 1 should not appear (but Line 10 contains "1",
# so check more specifically)
lines = content.split("\n")
line_numbers = [
int(line.split("\t")[0].strip()) for line in lines if line.strip()
]
self.assertNotIn(1, line_numbers)
self.assertIn(5, line_numbers)
async def test_read_with_limit(self) -> None:
"""Test reading with limit."""
chunk = await self.read_tool(
file_path=self.temp_file.name,
offset=1,
limit=3,
)
self.assertEqual(chunk.state, "running")
content = chunk.content[0].text
# Should only read 3 lines
self.assertIn("Line 1", content)
self.assertIn("Line 2", content)
self.assertIn("Line 3", content)
self.assertNotIn("Line 4", content)
async def test_read_nonexistent_file(self) -> None:
"""Test reading a non-existent file."""
chunk = await self.read_tool(file_path="/nonexistent/file.txt")
self.assertEqual(chunk.state, "error")
self.assertIn("does not exist", chunk.content[0].text)
async def test_read_directory(self) -> None:
"""Test reading a directory (should fail)."""
temp_dir = tempfile.mkdtemp()
try:
chunk = await self.read_tool(file_path=temp_dir)
self.assertEqual(chunk.state, "error")
self.assertIn("directory", chunk.content[0].text.lower())
finally:
os.rmdir(temp_dir)
async def test_match_rule_glob_pattern(self) -> None:
"""Test match_rule with glob patterns."""
# Test exact match
self.assertTrue(
await self.read_tool.match_rule(
"test.py",
{"file_path": "test.py"},
),
)
# Test wildcard pattern
self.assertTrue(
await self.read_tool.match_rule(
"*.py",
{"file_path": "test.py"},
),
)
# Test directory pattern
self.assertTrue(
await self.read_tool.match_rule(
"/tmp/**",
{"file_path": "/tmp/test.py"},
),
)
# Test non-matching pattern
self.assertFalse(
await self.read_tool.match_rule(
"*.txt",
{"file_path": "test.py"},
),
)
# Test empty file_path
self.assertFalse(
await self.read_tool.match_rule(
"*.py",
{"file_path": ""},
),
)
async def test_generate_suggestions(self) -> None:
"""Test generate_suggestions for file operations."""
# Test suggestion for file in subdirectory
suggestions = await self.read_tool.generate_suggestions(
{"file_path": "/tmp/project/src/main.py"},
)
self.assertIsInstance(suggestions, list)
self.assertGreater(len(suggestions), 0)
self.assertIsInstance(suggestions[0], PermissionRule)
# Should suggest parent directory pattern
suggestion_contents = [s.rule_content for s in suggestions]
self.assertIn("/tmp/project/src/**", suggestion_contents)
# Test suggestion for file in root
suggestions = await self.read_tool.generate_suggestions(
{"file_path": "/test.py"},
)
self.assertGreater(len(suggestions), 0)
|