254 lines
9.8 KiB
Python
254 lines
9.8 KiB
Python
"""Unit tests for FileExecutor (FILE_WRITE)."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Patch the logger before importing the executor
|
|
with patch("app.utils.logger.get_logger", return_value=MagicMock()):
|
|
from app.executors.file_executor import FileExecutor
|
|
|
|
|
|
class TestFileExecutor:
|
|
"""Test suite for FileExecutor."""
|
|
|
|
@pytest.fixture
|
|
def executor(self):
|
|
"""Create executor instance with mocked logger."""
|
|
with patch("app.executors.base.get_logger", return_value=MagicMock()):
|
|
return FileExecutor()
|
|
|
|
@pytest.fixture
|
|
def temp_file_root(self, tmp_path):
|
|
"""Create a temporary directory to act as /opt/letsbe."""
|
|
file_dir = tmp_path / "opt" / "letsbe"
|
|
file_dir.mkdir(parents=True)
|
|
return file_dir
|
|
|
|
@pytest.fixture
|
|
def mock_settings(self, temp_file_root):
|
|
"""Mock settings with temporary file root."""
|
|
settings = MagicMock()
|
|
settings.allowed_file_root = str(temp_file_root)
|
|
settings.allowed_env_root = str(temp_file_root / "env")
|
|
settings.max_file_size = 1_048_576 # 1MB
|
|
return settings
|
|
|
|
# ==================== Happy Path Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_happy_path(self, executor, temp_file_root, mock_settings):
|
|
"""Test writing a simple file."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "test.txt"
|
|
content = "Hello, World!"
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": content,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["written"] is True
|
|
assert result.data["path"] == str(file_path)
|
|
assert result.data["size"] == len(content.encode("utf-8"))
|
|
assert file_path.read_text() == content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_in_config_directory(self, executor, temp_file_root, mock_settings):
|
|
"""Test writing file in /opt/letsbe/config subdirectory.
|
|
|
|
This verifies that config paths are valid under allowed_file_root.
|
|
"""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
config_dir = temp_file_root / "config" / "app"
|
|
config_dir.mkdir(parents=True)
|
|
file_path = config_dir / "settings.json"
|
|
content = '{"debug": false}'
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": content,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["written"] is True
|
|
assert file_path.read_text() == content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_nested_config_directory(self, executor, temp_file_root, mock_settings):
|
|
"""Test writing file in nested config subdirectory."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
config_dir = temp_file_root / "config" / "nginx" / "sites-available"
|
|
config_dir.mkdir(parents=True)
|
|
file_path = config_dir / "default.conf"
|
|
content = "server { listen 80; }"
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": content,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert file_path.read_text() == content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_creates_parent_directories(self, executor, temp_file_root, mock_settings):
|
|
"""Test that parent directories are created automatically."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "new" / "nested" / "dir" / "file.txt"
|
|
content = "content"
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": content,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert file_path.exists()
|
|
assert file_path.read_text() == content
|
|
|
|
# ==================== Write Mode Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.skipif(
|
|
__import__("os").name == "nt",
|
|
reason="os.rename on Windows fails when target exists; executor uses os.rename instead of os.replace"
|
|
)
|
|
async def test_write_mode_overwrites(self, executor, temp_file_root, mock_settings):
|
|
"""Test that write mode overwrites existing content."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "test.txt"
|
|
file_path.write_text("old content")
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": "new content",
|
|
"mode": "write",
|
|
})
|
|
|
|
assert result.success is True
|
|
assert file_path.read_text() == "new content"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_mode(self, executor, temp_file_root, mock_settings):
|
|
"""Test append mode adds to existing content."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "test.txt"
|
|
file_path.write_text("line1\n")
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": "line2\n",
|
|
"mode": "append",
|
|
})
|
|
|
|
assert result.success is True
|
|
assert file_path.read_text() == "line1\nline2\n"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_mode_rejected(self, executor, temp_file_root, mock_settings):
|
|
"""Test rejection of invalid mode."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
result = await executor.execute({
|
|
"path": str(temp_file_root / "test.txt"),
|
|
"content": "content",
|
|
"mode": "invalid",
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Invalid mode" in result.error
|
|
|
|
# ==================== Path Validation Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_outside_allowed_root_rejected(self, executor, temp_file_root, mock_settings):
|
|
"""Test rejection of paths outside allowed root."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
result = await executor.execute({
|
|
"path": "/etc/passwd",
|
|
"content": "hack",
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Validation failed" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_path_traversal_rejected(self, executor, temp_file_root, mock_settings):
|
|
"""Test rejection of directory traversal attempts."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
result = await executor.execute({
|
|
"path": str(temp_file_root / ".." / ".." / "etc" / "passwd"),
|
|
"content": "hack",
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Validation failed" in result.error or "traversal" in result.error.lower()
|
|
|
|
# ==================== Payload Validation Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_path_payload(self, executor):
|
|
"""Test rejection of payload without path."""
|
|
with pytest.raises(ValueError, match="Missing required fields: path"):
|
|
await executor.execute({
|
|
"content": "content",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_content_payload(self, executor):
|
|
"""Test rejection of payload without content."""
|
|
with pytest.raises(ValueError, match="Missing required fields: content"):
|
|
await executor.execute({
|
|
"path": "/some/path",
|
|
})
|
|
|
|
# ==================== Task Type and Registry Tests ====================
|
|
|
|
def test_task_type_property(self, executor):
|
|
"""Test that task_type returns FILE_WRITE."""
|
|
assert executor.task_type == "FILE_WRITE"
|
|
|
|
def test_registry_integration(self):
|
|
"""Test that FILE_WRITE is registered in executor registry."""
|
|
from app.executors import get_executor
|
|
|
|
executor = get_executor("FILE_WRITE")
|
|
assert executor is not None
|
|
assert executor.task_type == "FILE_WRITE"
|
|
|
|
# ==================== Duration Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duration_ms_populated(self, executor, temp_file_root, mock_settings):
|
|
"""Test that duration_ms is populated in result."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "test.txt"
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": "content",
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.duration_ms is not None
|
|
assert result.duration_ms >= 0
|
|
|
|
# ==================== UTF-8 Tests ====================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_utf8_content(self, executor, temp_file_root, mock_settings):
|
|
"""Test writing UTF-8 encoded content."""
|
|
with patch("app.executors.file_executor.get_settings", return_value=mock_settings):
|
|
file_path = temp_file_root / "utf8.txt"
|
|
content = "Hello 世界! こんにちは 🎉"
|
|
|
|
result = await executor.execute({
|
|
"path": str(file_path),
|
|
"content": content,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert file_path.read_text(encoding="utf-8") == content
|