78 lines
2.2 KiB
Python
78 lines
2.2 KiB
Python
|
import pytest
|
||
|
from pathlib import Path
|
||
|
|
||
|
from pyr.tools.file_ops import ReadFileTool, WriteFileTool, DirectoryGlobTool, MkdirTool
|
||
|
|
||
|
|
||
|
@pytest.mark.asyncio
|
||
|
async def test_write_and_read_file(temp_dir):
|
||
|
write_tool = WriteFileTool()
|
||
|
read_tool = ReadFileTool()
|
||
|
|
||
|
test_file = temp_dir / "test.txt"
|
||
|
test_content = "Hello, PYR!"
|
||
|
|
||
|
result = await write_tool.execute(str(test_file), test_content)
|
||
|
assert "successfully" in result.lower()
|
||
|
|
||
|
result = await read_tool.execute(str(test_file))
|
||
|
assert test_content in result
|
||
|
|
||
|
|
||
|
@pytest.mark.asyncio
|
||
|
async def test_write_file_append(temp_dir):
|
||
|
write_tool = WriteFileTool()
|
||
|
read_tool = ReadFileTool()
|
||
|
|
||
|
test_file = temp_dir / "append_test.txt"
|
||
|
|
||
|
await write_tool.execute(str(test_file), "Line 1\n")
|
||
|
await write_tool.execute(str(test_file), "Line 2\n", append=True)
|
||
|
|
||
|
result = await read_tool.execute(str(test_file))
|
||
|
assert "Line 1" in result
|
||
|
assert "Line 2" in result
|
||
|
|
||
|
|
||
|
@pytest.mark.asyncio
|
||
|
async def test_directory_glob(temp_dir):
|
||
|
glob_tool = DirectoryGlobTool()
|
||
|
|
||
|
(temp_dir / "test1.txt").write_text("content1")
|
||
|
(temp_dir / "test2.txt").write_text("content2")
|
||
|
(temp_dir / "other.log").write_text("log content")
|
||
|
|
||
|
result = await glob_tool.execute(f"{temp_dir}/*.txt")
|
||
|
assert "test1.txt" in result
|
||
|
assert "test2.txt" in result
|
||
|
assert "other.log" not in result
|
||
|
|
||
|
|
||
|
@pytest.mark.asyncio
|
||
|
async def test_mkdir_tool(temp_dir):
|
||
|
mkdir_tool = MkdirTool()
|
||
|
|
||
|
new_dir = temp_dir / "new_directory" / "nested"
|
||
|
|
||
|
result = await mkdir_tool.execute(str(new_dir))
|
||
|
assert "created" in result.lower()
|
||
|
assert new_dir.exists()
|
||
|
assert new_dir.is_dir()
|
||
|
|
||
|
|
||
|
@pytest.mark.asyncio
|
||
|
async def test_read_nonexistent_file():
|
||
|
read_tool = ReadFileTool()
|
||
|
|
||
|
result = await read_tool.execute("/nonexistent/file.txt")
|
||
|
assert "error" in result.lower()
|
||
|
|
||
|
|
||
|
def test_tool_definitions():
|
||
|
read_tool = ReadFileTool()
|
||
|
definition = read_tool.get_definition()
|
||
|
|
||
|
assert definition.function["name"] == "read_file"
|
||
|
assert "path" in definition.function["parameters"]["properties"]
|
||
|
assert "path" in definition.function["parameters"]["required"]
|