Alex Blank
5deee8c65c
- Add ToolDefinitionManifest model with base image versioning - Add manifest compiler: Dockerfile + Compose generation from JSON manifests - Add permission fixer: post-start chown/chmod for mount policies - Add tool definition CRUD API with live compile preview endpoint - Integrate manifest-based startup flow in start_instance - Add Alembic migration with data conversion for pi-agent - Add 48 unit tests for manifest compiler, permission fixer, docker service - Keep backward compatibility with legacy dockerfile_template/compose_template Migration: applied successfully. Pi-agent converted to manifest. Quality gates: pytest (146 passed, 4 pre-existing unrelated failures)
146 lines
5.2 KiB
Python
146 lines
5.2 KiB
Python
"""Unit tests for the permission fixer."""
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.services.permission_fixer import (
|
|
PermissionFixError,
|
|
apply_mount_permissions,
|
|
check_root_user_available,
|
|
_run_in_container,
|
|
)
|
|
|
|
|
|
class TestApplyMountPermissions:
|
|
"""Tests for apply_mount_permissions."""
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_applies_chown_when_owner_declared(self, mock_run) -> None:
|
|
mounts = [
|
|
{"name": "workspace", "target": "/workspace", "owner": "user"},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["mount_name"] == "workspace"
|
|
assert results[0]["success"] is True
|
|
mock_run.assert_called_once()
|
|
args = mock_run.call_args[0]
|
|
assert args[0] == "abc123"
|
|
assert args[1] == ["chown", "-R", "user:user", "/workspace"]
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_applies_chmod_when_mode_declared(self, mock_run) -> None:
|
|
mounts = [
|
|
{"name": "ssh", "target": "/home/user/.ssh", "mode": "0700"},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert results[0]["success"] is True
|
|
# Only chmod called (no owner, so no chown)
|
|
assert mock_run.call_count == 1
|
|
chmod_call = mock_run.call_args_list[0]
|
|
assert chmod_call[0][1] == ["chmod", "0700", "/home/user/.ssh"]
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_applies_file_mode_when_declared(self, mock_run) -> None:
|
|
mounts = [
|
|
{
|
|
"name": "ssh",
|
|
"target": "/home/user/.ssh",
|
|
"file_mode": "0600",
|
|
},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert results[0]["success"] is True
|
|
# Only file_mode called (no owner, no mode)
|
|
assert mock_run.call_count == 1
|
|
file_mode_call = mock_run.call_args_list[0]
|
|
assert file_mode_call[0][1][0] == "sh"
|
|
assert (
|
|
"find /home/user/.ssh -type f -exec chmod 0600" in file_mode_call[0][1][2]
|
|
)
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_skips_mount_with_no_policy(self, mock_run) -> None:
|
|
mounts = [
|
|
{"name": "workspace", "target": "/workspace", "writable": True},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["success"] is True
|
|
mock_run.assert_not_called()
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_reports_failure_on_command_error(self, mock_run) -> None:
|
|
mock_run.side_effect = PermissionFixError("chown failed")
|
|
|
|
mounts = [
|
|
{"name": "workspace", "target": "/workspace", "owner": "user"},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert results[0]["success"] is False
|
|
assert "chown failed" in results[0]["error"]
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_stops_on_first_failure(self, mock_run) -> None:
|
|
"""If chown fails, chmod and file_mode should not run."""
|
|
mock_run.side_effect = PermissionFixError("chown failed")
|
|
|
|
mounts = [
|
|
{
|
|
"name": "workspace",
|
|
"target": "/workspace",
|
|
"owner": "user",
|
|
"mode": "0755",
|
|
"file_mode": "0644",
|
|
},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert results[0]["success"] is False
|
|
assert mock_run.call_count == 1 # Only chown attempted
|
|
|
|
|
|
class TestRunInContainer:
|
|
"""Tests for _run_in_container."""
|
|
|
|
@patch("subprocess.run")
|
|
def test_success(self, mock_run) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
_run_in_container("abc123", ["echo", "hello"], 10)
|
|
mock_run.assert_called_once()
|
|
cmd = mock_run.call_args[0][0]
|
|
assert cmd == ["docker", "exec", "--user", "root", "abc123", "echo", "hello"]
|
|
|
|
@patch("subprocess.run")
|
|
def test_failure_raises(self, mock_run) -> None:
|
|
mock_run.return_value = MagicMock(returncode=1, stderr="permission denied")
|
|
with pytest.raises(PermissionFixError, match="permission denied"):
|
|
_run_in_container("abc123", ["chown", "x"], 10)
|
|
|
|
@patch("subprocess.run")
|
|
def test_timeout_raises(self, mock_run) -> None:
|
|
import subprocess
|
|
|
|
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["docker"], timeout=10)
|
|
with pytest.raises(PermissionFixError, match="timed out"):
|
|
_run_in_container("abc123", ["chown", "x"], 10)
|
|
|
|
|
|
class TestCheckRootUserAvailable:
|
|
"""Tests for check_root_user_available."""
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_returns_true_when_root_exists(self, mock_run) -> None:
|
|
assert check_root_user_available("abc123") is True
|
|
|
|
@patch("src.services.permission_fixer._run_in_container")
|
|
def test_returns_false_when_root_missing(self, mock_run) -> None:
|
|
mock_run.side_effect = PermissionFixError("no such user")
|
|
assert check_root_user_available("abc123") is False
|