241 lines
8.7 KiB
Python
241 lines
8.7 KiB
Python
"""Tests for filesystem tools."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from agent.tools import filesystem as fs_tools
|
|
from infrastructure.persistence import get_memory
|
|
|
|
|
|
class TestSetPathForFolder:
|
|
"""Tests for set_path_for_folder tool."""
|
|
|
|
def test_success(self, memory, real_folder):
|
|
"""Should set folder path successfully."""
|
|
result = fs_tools.set_path_for_folder("download", str(real_folder["downloads"]))
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["folder_name"] == "download"
|
|
assert result["path"] == str(real_folder["downloads"])
|
|
|
|
def test_saves_to_ltm(self, memory, real_folder):
|
|
"""Should save path to LTM config."""
|
|
fs_tools.set_path_for_folder("download", str(real_folder["downloads"]))
|
|
|
|
mem = get_memory()
|
|
assert mem.ltm.get_config("download_folder") == str(real_folder["downloads"])
|
|
|
|
def test_all_folder_types(self, memory, real_folder):
|
|
"""Should accept all valid folder types."""
|
|
for folder_type in ["download", "movie", "tvshow", "torrent"]:
|
|
result = fs_tools.set_path_for_folder(
|
|
folder_type, str(real_folder["downloads"])
|
|
)
|
|
assert result["status"] == "ok"
|
|
|
|
def test_invalid_folder_type(self, memory, real_folder):
|
|
"""Should reject invalid folder type."""
|
|
result = fs_tools.set_path_for_folder("invalid", str(real_folder["downloads"]))
|
|
|
|
assert result["error"] == "validation_failed"
|
|
|
|
def test_path_not_exists(self, memory):
|
|
"""Should reject non-existent path."""
|
|
result = fs_tools.set_path_for_folder("download", "/nonexistent/path/12345")
|
|
|
|
assert result["error"] == "invalid_path"
|
|
assert "does not exist" in result["message"]
|
|
|
|
def test_path_is_file(self, memory, real_folder):
|
|
"""Should reject file path."""
|
|
file_path = real_folder["downloads"] / "test_movie.mkv"
|
|
|
|
result = fs_tools.set_path_for_folder("download", str(file_path))
|
|
|
|
assert result["error"] == "invalid_path"
|
|
assert "not a directory" in result["message"]
|
|
|
|
def test_resolves_path(self, memory, real_folder):
|
|
"""Should resolve relative paths."""
|
|
# Create a symlink or use relative path
|
|
relative_path = real_folder["downloads"]
|
|
|
|
result = fs_tools.set_path_for_folder("download", str(relative_path))
|
|
|
|
assert result["status"] == "ok"
|
|
# Path should be absolute
|
|
assert Path(result["path"]).is_absolute()
|
|
|
|
|
|
class TestListFolder:
|
|
"""Tests for list_folder tool."""
|
|
|
|
def test_success(self, memory, real_folder):
|
|
"""Should list folder contents."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download")
|
|
|
|
assert result["status"] == "ok"
|
|
assert "test_movie.mkv" in result["entries"]
|
|
assert "test_series" in result["entries"]
|
|
assert result["count"] == 2
|
|
|
|
def test_subfolder(self, memory, real_folder):
|
|
"""Should list subfolder contents."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "test_series")
|
|
|
|
assert result["status"] == "ok"
|
|
assert "episode1.mkv" in result["entries"]
|
|
|
|
def test_folder_not_configured(self, memory):
|
|
"""Should return error if folder not configured."""
|
|
result = fs_tools.list_folder("download")
|
|
|
|
assert result["error"] == "folder_not_set"
|
|
|
|
def test_invalid_folder_type(self, memory):
|
|
"""Should reject invalid folder type."""
|
|
result = fs_tools.list_folder("invalid")
|
|
|
|
assert result["error"] == "validation_failed"
|
|
|
|
def test_path_traversal_dotdot(self, memory, real_folder):
|
|
"""Should block path traversal with .."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "../")
|
|
|
|
assert result["error"] == "forbidden"
|
|
|
|
def test_path_traversal_absolute(self, memory, real_folder):
|
|
"""Should block absolute paths."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "/etc/passwd")
|
|
|
|
assert result["error"] == "forbidden"
|
|
|
|
def test_path_traversal_encoded(self, memory, real_folder):
|
|
"""Should block encoded traversal attempts."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "..%2F..%2Fetc")
|
|
|
|
# Should either be forbidden or not found (depending on normalization)
|
|
assert result.get("error") in ["forbidden", "not_found"]
|
|
|
|
def test_path_not_exists(self, memory, real_folder):
|
|
"""Should return error for non-existent path."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "nonexistent_folder")
|
|
|
|
assert result["error"] == "not_found"
|
|
|
|
def test_path_is_file(self, memory, real_folder):
|
|
"""Should return error if path is a file."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "test_movie.mkv")
|
|
|
|
assert result["error"] == "not_a_directory"
|
|
|
|
def test_empty_folder(self, memory, real_folder):
|
|
"""Should handle empty folder."""
|
|
empty_dir = real_folder["downloads"] / "empty"
|
|
empty_dir.mkdir()
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "empty")
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["entries"] == []
|
|
assert result["count"] == 0
|
|
|
|
def test_sorted_entries(self, memory, real_folder):
|
|
"""Should return sorted entries."""
|
|
# Create files with different names
|
|
(real_folder["downloads"] / "zebra.txt").touch()
|
|
(real_folder["downloads"] / "alpha.txt").touch()
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download")
|
|
|
|
assert result["status"] == "ok"
|
|
# Check that entries are sorted
|
|
entries = result["entries"]
|
|
assert entries == sorted(entries)
|
|
|
|
|
|
class TestFileManagerSecurity:
|
|
"""Security-focused tests for FileManager."""
|
|
|
|
def test_null_byte_injection(self, memory, real_folder):
|
|
"""Should block null byte injection."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "test\x00.txt")
|
|
|
|
assert result["error"] == "forbidden"
|
|
|
|
def test_path_outside_root(self, memory, real_folder):
|
|
"""Should block paths that escape root."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
# Try to access parent directory
|
|
result = fs_tools.list_folder("download", "test_series/../../")
|
|
|
|
assert result["error"] == "forbidden"
|
|
|
|
def test_symlink_escape(self, memory, real_folder):
|
|
"""Should handle symlinks that point outside root."""
|
|
# Create a symlink pointing outside
|
|
symlink = real_folder["downloads"] / "escape_link"
|
|
try:
|
|
symlink.symlink_to("/tmp")
|
|
except OSError:
|
|
pytest.skip("Cannot create symlinks")
|
|
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "escape_link")
|
|
|
|
# Should either be forbidden or work (depending on policy)
|
|
# The important thing is it doesn't crash
|
|
assert "error" in result or "status" in result
|
|
|
|
def test_special_characters_in_path(self, memory, real_folder):
|
|
"""Should handle special characters in path."""
|
|
special_dir = real_folder["downloads"] / "special !@#$%"
|
|
special_dir.mkdir()
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "special !@#$%")
|
|
|
|
assert result["status"] == "ok"
|
|
|
|
def test_unicode_path(self, memory, real_folder):
|
|
"""Should handle unicode in path."""
|
|
unicode_dir = real_folder["downloads"] / "日本語フォルダ"
|
|
unicode_dir.mkdir()
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
result = fs_tools.list_folder("download", "日本語フォルダ")
|
|
|
|
assert result["status"] == "ok"
|
|
|
|
def test_very_long_path(self, memory, real_folder):
|
|
"""Should handle very long paths gracefully."""
|
|
memory.ltm.set_config("download_folder", str(real_folder["downloads"]))
|
|
|
|
long_path = "a" * 1000
|
|
|
|
result = fs_tools.list_folder("download", long_path)
|
|
|
|
# Should return an error, not crash
|
|
assert "error" in result
|