feat!: migrate to OpenAI native tool calls and fix circular deps (#fuck-gemini)
- Fix circular dependencies in agent/tools - Migrate from custom JSON to OpenAI tool calls format - Add async streaming (step_stream, complete_stream) - Simplify prompt system and remove token counting - Add 5 new API endpoints (/health, /v1/models, /api/memory/*) - Add 3 new tools (get_torrent_by_index, add_torrent_by_index, set_language) - Fix all 500 tests and add coverage config (80% threshold) - Add comprehensive docs (README, pytest guide) BREAKING: LLM interface changed, memory injection via get_memory()
This commit is contained in:
240
tests/test_tools_filesystem.py
Normal file
240
tests/test_tools_filesystem.py
Normal file
@@ -0,0 +1,240 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user