Rebase onto upstream (a4d95fd)
#12
@@ -160,13 +160,9 @@ class MemoryTool20250818(BaseAnthropicTool):
|
||||
path_str = path # Keep original for error messages
|
||||
validated_path = self._validate_memory_path(path)
|
||||
|
||||
# Directory listing (will implement in Task 4)
|
||||
# Directory listing
|
||||
if validated_path.is_dir():
|
||||
return CLIResult(
|
||||
exit_code=1,
|
||||
output="",
|
||||
error="Directory listing not implemented yet"
|
||||
)
|
||||
return await self._view_directory(validated_path, path_str)
|
||||
|
||||
# File viewing
|
||||
if not validated_path.exists():
|
||||
@@ -225,6 +221,86 @@ class MemoryTool20250818(BaseAnthropicTool):
|
||||
error=""
|
||||
)
|
||||
|
||||
async def _view_directory(self, path: Path, path_str: str) -> CLIResult:
|
||||
"""View directory listing up to 2 levels deep.
|
||||
|
||||
Args:
|
||||
path: Validated Path object
|
||||
path_str: Original path string for display
|
||||
|
||||
Returns:
|
||||
CLIResult with directory listing
|
||||
"""
|
||||
import os
|
||||
|
||||
def format_size(size_bytes: int) -> str:
|
||||
"""Convert bytes to human-readable format."""
|
||||
for unit in ['B', 'K', 'M', 'G', 'T']:
|
||||
if size_bytes < 1024:
|
||||
return f"{size_bytes:.1f}{unit}"
|
||||
size_bytes /= 1024
|
||||
return f"{size_bytes:.1f}P"
|
||||
|
||||
lines = []
|
||||
header = f"Here're the files and directories up to 2 levels deep in {path_str}, excluding hidden items and node_modules:"
|
||||
lines.append(header)
|
||||
|
||||
# Walk directory tree (max depth 2)
|
||||
base_depth = str(path).count(os.sep)
|
||||
|
||||
for root, dirs, files in os.walk(path):
|
||||
# Calculate current depth
|
||||
current_depth = str(root).count(os.sep) - base_depth
|
||||
|
||||
# Filter out hidden items and node_modules at this level
|
||||
dirs[:] = [d for d in dirs if not d.startswith('.') and d != 'node_modules']
|
||||
|
||||
# Stop if we've gone too deep
|
||||
if current_depth >= 2:
|
||||
dirs.clear() # Don't recurse further
|
||||
continue
|
||||
|
||||
# Get size and add directory entry
|
||||
root_path = Path(root)
|
||||
try:
|
||||
# Directory size (sum of all files within, or 4K default)
|
||||
dir_size = sum(f.stat().st_size for f in root_path.rglob('*') if f.is_file())
|
||||
if dir_size == 0:
|
||||
dir_size = 4096 # Default directory size
|
||||
size_str = format_size(dir_size)
|
||||
|
||||
# Convert absolute path to /memories/... format
|
||||
relative = root_path.relative_to(self.workspace)
|
||||
display_path = "/" + str(relative).replace(os.sep, "/")
|
||||
|
||||
lines.append(f"{size_str}\t{display_path}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Add file entries at this level
|
||||
for filename in sorted(files):
|
||||
if filename.startswith('.'):
|
||||
continue # Skip hidden files
|
||||
|
||||
file_path = root_path / filename
|
||||
try:
|
||||
file_size = file_path.stat().st_size
|
||||
size_str = format_size(file_size)
|
||||
|
||||
# Convert to /memories/... format
|
||||
relative = file_path.relative_to(self.workspace)
|
||||
display_path = "/" + str(relative).replace(os.sep, "/")
|
||||
|
||||
lines.append(f"{size_str}\t{display_path}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return CLIResult(
|
||||
exit_code=0,
|
||||
output="\n".join(lines),
|
||||
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")
|
||||
|
||||
@@ -77,3 +77,34 @@ async def test_view_file_not_exists(memory_tool):
|
||||
assert result.exit_code == 1
|
||||
assert result.output == ""
|
||||
assert "The path /memories/nonexistent.txt does not exist" in result.error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_directory(memory_tool, temp_workspace):
|
||||
"""Test viewing a directory listing."""
|
||||
# Create test directory structure
|
||||
memories = temp_workspace / "memories"
|
||||
(memories / "notes.txt").write_text("content")
|
||||
(memories / "project").mkdir()
|
||||
(memories / "project" / "status.xml").write_text("<status>ok</status>")
|
||||
(memories / ".hidden").write_text("hidden") # Should be excluded
|
||||
|
||||
result = await memory_tool(command="view", path="/memories")
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert result.error == ""
|
||||
assert "Here're the files and directories up to 2 levels deep in /memories" in result.output
|
||||
assert "/memories" in result.output
|
||||
assert "/memories/notes.txt" in result.output
|
||||
assert "/memories/project" in result.output
|
||||
assert "/memories/project/status.xml" in result.output
|
||||
assert ".hidden" not in result.output # Hidden files excluded
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_view_empty_directory(memory_tool):
|
||||
"""Test viewing an empty directory."""
|
||||
result = await memory_tool(command="view", path="/memories")
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "/memories" in result.output
|
||||
|
||||
Reference in New Issue
Block a user