468 lines
18 KiB
Python
468 lines
18 KiB
Python
"""Unit tests for DockerExecutor."""
|
|
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
import pytest
|
|
|
|
|
|
# Patch the logger before importing the executor
|
|
with patch("app.utils.logger.get_logger", return_value=MagicMock()):
|
|
from app.executors.docker_executor import DockerExecutor
|
|
|
|
|
|
class TestDockerExecutor:
|
|
"""Tests for DockerExecutor."""
|
|
|
|
@pytest.fixture
|
|
def executor(self):
|
|
"""Create executor with mocked logger."""
|
|
with patch("app.executors.base.get_logger", return_value=MagicMock()):
|
|
return DockerExecutor()
|
|
|
|
@pytest.fixture
|
|
def temp_stacks_root(self, tmp_path):
|
|
"""Create a temporary stacks root directory."""
|
|
stacks_dir = tmp_path / "opt" / "letsbe" / "stacks"
|
|
stacks_dir.mkdir(parents=True)
|
|
return stacks_dir
|
|
|
|
@pytest.fixture
|
|
def mock_settings(self, temp_stacks_root):
|
|
"""Mock settings with temporary paths."""
|
|
settings = MagicMock()
|
|
settings.allowed_stacks_root = str(temp_stacks_root)
|
|
return settings
|
|
|
|
@pytest.fixture
|
|
def mock_get_settings(self, mock_settings):
|
|
"""Patch get_settings to return mock settings."""
|
|
with patch("app.executors.docker_executor.get_settings", return_value=mock_settings):
|
|
yield mock_settings
|
|
|
|
@pytest.fixture
|
|
def sample_compose_content(self):
|
|
"""Sample docker-compose.yml content."""
|
|
return """version: '3.8'
|
|
services:
|
|
app:
|
|
image: nginx:latest
|
|
ports:
|
|
- "80:80"
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def stack_with_docker_compose_yml(self, temp_stacks_root, sample_compose_content):
|
|
"""Create a stack with docker-compose.yml."""
|
|
stack_dir = temp_stacks_root / "myapp"
|
|
stack_dir.mkdir()
|
|
compose_file = stack_dir / "docker-compose.yml"
|
|
compose_file.write_text(sample_compose_content)
|
|
return stack_dir
|
|
|
|
@pytest.fixture
|
|
def stack_with_compose_yml(self, temp_stacks_root, sample_compose_content):
|
|
"""Create a stack with compose.yml (no docker-compose.yml)."""
|
|
stack_dir = temp_stacks_root / "otherapp"
|
|
stack_dir.mkdir()
|
|
compose_file = stack_dir / "compose.yml"
|
|
compose_file.write_text(sample_compose_content)
|
|
return stack_dir
|
|
|
|
@pytest.fixture
|
|
def stack_without_compose(self, temp_stacks_root):
|
|
"""Create a stack without any compose file."""
|
|
stack_dir = temp_stacks_root / "emptyapp"
|
|
stack_dir.mkdir()
|
|
return stack_dir
|
|
|
|
# =========================================================================
|
|
# SUCCESS CASES
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_with_docker_compose_yml(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test successful reload with docker-compose.yml."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "Container started", "")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["compose_dir"] == str(stack_with_docker_compose_yml)
|
|
assert result.data["compose_file"] == str(stack_with_docker_compose_yml / "docker-compose.yml")
|
|
assert result.data["pull_ran"] is False
|
|
assert "up" in result.data["logs"]
|
|
|
|
# Verify only 'up' command was called
|
|
mock_run.assert_called_once()
|
|
call_args = mock_run.call_args
|
|
assert call_args[0][2] == ["up", "-d", "--remove-orphans"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_success_with_compose_yml_fallback(
|
|
self, executor, mock_get_settings, stack_with_compose_yml
|
|
):
|
|
"""Test successful reload with compose.yml fallback."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "Container started", "")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_compose_yml),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["compose_file"] == str(stack_with_compose_yml / "compose.yml")
|
|
assert result.data["pull_ran"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_docker_compose_yml_preferred_over_compose_yml(
|
|
self, executor, mock_get_settings, temp_stacks_root, sample_compose_content
|
|
):
|
|
"""Test that docker-compose.yml is preferred over compose.yml."""
|
|
stack_dir = temp_stacks_root / "bothfiles"
|
|
stack_dir.mkdir()
|
|
(stack_dir / "docker-compose.yml").write_text(sample_compose_content)
|
|
(stack_dir / "compose.yml").write_text(sample_compose_content)
|
|
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "", "")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_dir),
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["compose_file"] == str(stack_dir / "docker-compose.yml")
|
|
|
|
# =========================================================================
|
|
# PULL PARAMETER TESTS
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_false_only_up_called(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that pull=false only runs 'up' command."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "", "")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"pull": False,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["pull_ran"] is False
|
|
assert "pull" not in result.data["logs"]
|
|
assert "up" in result.data["logs"]
|
|
|
|
# Only one call (up)
|
|
assert mock_run.call_count == 1
|
|
call_args = mock_run.call_args
|
|
assert call_args[0][2] == ["up", "-d", "--remove-orphans"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_true_both_commands_called(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that pull=true runs both 'pull' and 'up' commands."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "output", "")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"pull": True,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert result.data["pull_ran"] is True
|
|
assert "pull" in result.data["logs"]
|
|
assert "up" in result.data["logs"]
|
|
|
|
# Two calls: pull then up
|
|
assert mock_run.call_count == 2
|
|
calls = mock_run.call_args_list
|
|
assert calls[0][0][2] == ["pull"]
|
|
assert calls[1][0][2] == ["up", "-d", "--remove-orphans"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_fails_stops_execution(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that pull failure stops execution before 'up'."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (1, "", "Error pulling images")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"pull": True,
|
|
})
|
|
|
|
assert result.success is False
|
|
assert result.data["pull_ran"] is True
|
|
assert "pull" in result.data["logs"]
|
|
assert "up" not in result.data["logs"]
|
|
assert "pull failed" in result.error.lower()
|
|
|
|
# Only one call (pull)
|
|
assert mock_run.call_count == 1
|
|
|
|
# =========================================================================
|
|
# FAILURE CASES
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_compose_file(
|
|
self, executor, mock_get_settings, stack_without_compose
|
|
):
|
|
"""Test failure when no compose file is found."""
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_without_compose),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "No compose file found" in result.error
|
|
assert "docker-compose.yml" in result.error
|
|
assert "compose.yml" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_up_command_fails(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test failure when 'up' command fails."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (1, "", "Error: container crashed")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Docker up failed" in result.error
|
|
assert "up" in result.data["logs"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_compose_dir_parameter(self, executor, mock_get_settings):
|
|
"""Test failure when compose_dir is missing from payload."""
|
|
with pytest.raises(ValueError, match="Missing required fields: compose_dir"):
|
|
await executor.execute({})
|
|
|
|
# =========================================================================
|
|
# PATH SECURITY TESTS
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_path_outside_allowed_root(
|
|
self, executor, mock_get_settings, tmp_path
|
|
):
|
|
"""Test rejection of compose_dir outside allowed stacks root."""
|
|
outside_dir = tmp_path / "outside"
|
|
outside_dir.mkdir()
|
|
(outside_dir / "docker-compose.yml").write_text("version: '3'\n")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(outside_dir),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "validation failed" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_path_traversal_attack(
|
|
self, executor, mock_get_settings, temp_stacks_root
|
|
):
|
|
"""Test rejection of path traversal attempts."""
|
|
malicious_path = str(temp_stacks_root / ".." / ".." / "etc")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": malicious_path,
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "traversal" in result.error.lower() or "validation" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_nonexistent_directory(
|
|
self, executor, mock_get_settings, temp_stacks_root
|
|
):
|
|
"""Test rejection of nonexistent directory."""
|
|
result = await executor.execute({
|
|
"compose_dir": str(temp_stacks_root / "doesnotexist"),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "validation failed" in result.error.lower() or "does not exist" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reject_file_instead_of_directory(
|
|
self, executor, mock_get_settings, temp_stacks_root
|
|
):
|
|
"""Test rejection when compose_dir points to a file instead of directory."""
|
|
file_path = temp_stacks_root / "notadir.yml"
|
|
file_path.write_text("version: '3'\n")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(file_path),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "not a directory" in result.error.lower() or "validation" in result.error.lower()
|
|
|
|
# =========================================================================
|
|
# TIMEOUT AND ERROR HANDLING TESTS
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_timeout_handling(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test timeout handling."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.side_effect = asyncio.TimeoutError()
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"timeout": 10,
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "timed out" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unexpected_exception_handling(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test handling of unexpected exceptions."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.side_effect = RuntimeError("Unexpected error")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
})
|
|
|
|
assert result.success is False
|
|
assert "Unexpected error" in result.error
|
|
|
|
# =========================================================================
|
|
# OUTPUT STRUCTURE TESTS
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_result_structure_on_success(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that result has correct structure on success."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "stdout content", "stderr content")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"pull": True,
|
|
})
|
|
|
|
assert result.success is True
|
|
assert "compose_dir" in result.data
|
|
assert "compose_file" in result.data
|
|
assert "pull_ran" in result.data
|
|
assert "logs" in result.data
|
|
assert isinstance(result.data["logs"], dict)
|
|
assert "pull" in result.data["logs"]
|
|
assert "up" in result.data["logs"]
|
|
assert result.duration_ms is not None
|
|
assert result.error is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_combine_stdout_stderr(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that logs contain both stdout and stderr."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "stdout line", "stderr line")
|
|
|
|
result = await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
})
|
|
|
|
assert "stdout line" in result.data["logs"]["up"]
|
|
assert "stderr line" in result.data["logs"]["up"]
|
|
|
|
# =========================================================================
|
|
# INTERNAL METHOD TESTS
|
|
# =========================================================================
|
|
|
|
def test_find_compose_file_docker_compose_yml(
|
|
self, executor, stack_with_docker_compose_yml
|
|
):
|
|
"""Test _find_compose_file finds docker-compose.yml."""
|
|
result = executor._find_compose_file(stack_with_docker_compose_yml)
|
|
assert result == stack_with_docker_compose_yml / "docker-compose.yml"
|
|
|
|
def test_find_compose_file_compose_yml(
|
|
self, executor, stack_with_compose_yml
|
|
):
|
|
"""Test _find_compose_file finds compose.yml."""
|
|
result = executor._find_compose_file(stack_with_compose_yml)
|
|
assert result == stack_with_compose_yml / "compose.yml"
|
|
|
|
def test_find_compose_file_not_found(
|
|
self, executor, stack_without_compose
|
|
):
|
|
"""Test _find_compose_file returns None when not found."""
|
|
result = executor._find_compose_file(stack_without_compose)
|
|
assert result is None
|
|
|
|
def test_combine_output_both_present(self, executor):
|
|
"""Test _combine_output with both stdout and stderr."""
|
|
result = executor._combine_output("stdout", "stderr")
|
|
assert result == "stdout\nstderr"
|
|
|
|
def test_combine_output_stdout_only(self, executor):
|
|
"""Test _combine_output with only stdout."""
|
|
result = executor._combine_output("stdout", "")
|
|
assert result == "stdout"
|
|
|
|
def test_combine_output_stderr_only(self, executor):
|
|
"""Test _combine_output with only stderr."""
|
|
result = executor._combine_output("", "stderr")
|
|
assert result == "stderr"
|
|
|
|
def test_combine_output_both_empty(self, executor):
|
|
"""Test _combine_output with empty strings."""
|
|
result = executor._combine_output("", "")
|
|
assert result == ""
|
|
|
|
# =========================================================================
|
|
# TASK TYPE TEST
|
|
# =========================================================================
|
|
|
|
def test_task_type(self, executor):
|
|
"""Test task_type property."""
|
|
assert executor.task_type == "DOCKER_RELOAD"
|
|
|
|
# =========================================================================
|
|
# CUSTOM TIMEOUT TEST
|
|
# =========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_timeout_passed_to_command(
|
|
self, executor, mock_get_settings, stack_with_docker_compose_yml
|
|
):
|
|
"""Test that custom timeout is passed to subprocess."""
|
|
with patch.object(executor, "_run_compose_command") as mock_run:
|
|
mock_run.return_value = (0, "", "")
|
|
|
|
await executor.execute({
|
|
"compose_dir": str(stack_with_docker_compose_yml),
|
|
"timeout": 120,
|
|
})
|
|
|
|
call_args = mock_run.call_args
|
|
assert call_args[0][3] == 120 # timeout argument
|