Rebase onto upstream (a4d95fd)
#12
@@ -103,13 +103,152 @@ class MemoryTool20250818(BaseAnthropicTool):
|
||||
Returns:
|
||||
CLIResult with command output or error
|
||||
"""
|
||||
# Placeholder - will implement in later tasks
|
||||
try:
|
||||
if command == "view":
|
||||
return await self._view(path, view_range)
|
||||
elif command == "create":
|
||||
return await self._create(path, file_text)
|
||||
elif command == "str_replace":
|
||||
return await self._str_replace(path, old_str, new_str)
|
||||
elif command == "insert":
|
||||
return await self._insert(path, insert_line, insert_text)
|
||||
elif command == "delete":
|
||||
return await self._delete(path)
|
||||
elif command == "rename":
|
||||
return await self._rename(old_path, new_path)
|
||||
else:
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"Unknown command: {command}"
|
||||
)
|
||||
except ValueError as e:
|
||||
# Path security error
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"Error: {e}"
|
||||
)
|
||||
except Exception as e:
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"Error: {e}"
|
||||
)
|
||||
|
||||
async def _view(
|
||||
self,
|
||||
path: str | None,
|
||||
view_range: list[int] | None = None,
|
||||
) -> CLIResult:
|
||||
"""View directory listing or file contents.
|
||||
|
||||
Args:
|
||||
path: Path to view
|
||||
view_range: Optional [start_line, end_line] for file viewing (1-indexed)
|
||||
|
||||
Returns:
|
||||
CLIResult with directory listing or file contents
|
||||
"""
|
||||
if path is None:
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error="Error: path is required for view command"
|
||||
)
|
||||
|
||||
path_str = path # Keep original for error messages
|
||||
validated_path = self._validate_memory_path(path)
|
||||
|
||||
# Directory listing (will implement in Task 4)
|
||||
if validated_path.is_dir():
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error="Directory listing not implemented yet"
|
||||
)
|
||||
|
||||
# File viewing
|
||||
if not validated_path.exists():
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"The path {path_str} does not exist. Please provide a valid path."
|
||||
)
|
||||
|
||||
# Read file
|
||||
try:
|
||||
content = validated_path.read_text()
|
||||
except Exception as e:
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"Error reading file: {e}"
|
||||
)
|
||||
|
||||
lines = content.splitlines(keepends=True)
|
||||
|
||||
# Check line limit
|
||||
if len(lines) > 999_999:
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error=f"File {path_str} exceeds maximum line limit of 999,999 lines."
|
||||
)
|
||||
|
||||
# Apply view_range if specified
|
||||
if view_range:
|
||||
start, end = view_range
|
||||
# Convert to 0-indexed, clamp to valid range
|
||||
start_idx = max(0, start - 1)
|
||||
end_idx = min(len(lines), end)
|
||||
lines_to_show = lines[start_idx:end_idx]
|
||||
start_num = start
|
||||
else:
|
||||
lines_to_show = lines
|
||||
start_num = 1
|
||||
|
||||
# Format with line numbers (6 chars, right-aligned, tab-separated)
|
||||
formatted_lines = []
|
||||
for i, line in enumerate(lines_to_show):
|
||||
line_num = start_num + i
|
||||
# Remove trailing newline for display
|
||||
line_content = line.rstrip("\n")
|
||||
formatted_lines.append(f"{line_num:6d}\t{line_content}")
|
||||
|
||||
output = f"Here's the content of {path_str} with line numbers:\n"
|
||||
output += "\n".join(formatted_lines)
|
||||
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error="Not implemented yet"
|
||||
exit_code=0,
|
||||
output=output,
|
||||
error=""
|
||||
)
|
||||
|
||||
async def _create(self, path: str | None, file_text: str | None) -> CLIResult:
|
||||
"""Placeholder for create command."""
|
||||
return CLIResult(exit_code=1, output="", error="Not implemented yet")
|
||||
|
||||
async def _str_replace(
|
||||
self, path: str | None, old_str: str | None, new_str: str | None
|
||||
) -> CLIResult:
|
||||
"""Placeholder for str_replace command."""
|
||||
return CLIResult(exit_code=1, output="", error="Not implemented yet")
|
||||
|
||||
async def _insert(
|
||||
self, path: str | None, insert_line: int | None, insert_text: str | None
|
||||
) -> CLIResult:
|
||||
"""Placeholder for insert command."""
|
||||
return CLIResult(exit_code=1, output="", error="Not implemented yet")
|
||||
|
||||
async def _delete(self, path: str | None) -> CLIResult:
|
||||
"""Placeholder for delete command."""
|
||||
return CLIResult(exit_code=1, output="", error="Not implemented yet")
|
||||
|
||||
async def _rename(self, old_path: str | None, new_path: str | None) -> CLIResult:
|
||||
"""Placeholder for rename command."""
|
||||
return CLIResult(exit_code=1, output="", error="Not implemented yet")
|
||||
|
||||
def to_params(self) -> dict[str, Any]:
|
||||
"""Convert to Anthropic API tool parameter format.
|
||||
|
||||
|
||||
@@ -33,3 +33,47 @@ def test_memory_tool_to_params(memory_tool):
|
||||
"type": "memory_20250818",
|
||||
"name": "memory"
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_file(memory_tool, temp_workspace):
|
||||
"""Test viewing a file with line numbers."""
|
||||
# Create test file
|
||||
test_file = temp_workspace / "memories" / "notes.txt"
|
||||
test_file.write_text("Line 1\nLine 2\nLine 3\n")
|
||||
|
||||
result = await memory_tool(command="view", path="/memories/notes.txt")
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert result.error == ""
|
||||
assert "Here's the content of /memories/notes.txt with line numbers:" in result.output
|
||||
assert " 1\tLine 1" in result.output
|
||||
assert " 2\tLine 2" in result.output
|
||||
assert " 3\tLine 3" in result.output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_file_with_range(memory_tool, temp_workspace):
|
||||
"""Test viewing a file with line range."""
|
||||
# Create test file with 10 lines
|
||||
test_file = temp_workspace / "memories" / "test.txt"
|
||||
test_file.write_text("\n".join([f"Line {i}" for i in range(1, 11)]))
|
||||
|
||||
result = await memory_tool(command="view", path="/memories/test.txt", view_range=[3, 5])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert " 3\tLine 3" in result.output
|
||||
assert " 4\tLine 4" in result.output
|
||||
assert " 5\tLine 5" in result.output
|
||||
assert "Line 1" not in result.output
|
||||
assert "Line 10" not in result.output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_file_not_exists(memory_tool):
|
||||
"""Test viewing a nonexistent file."""
|
||||
result = await memory_tool(command="view", path="/memories/nonexistent.txt")
|
||||
|
||||
assert result.exit_code == 1
|
||||
assert result.output == ""
|
||||
assert "The path /memories/nonexistent.txt does not exist" in result.error
|
||||
|
||||
Reference in New Issue
Block a user