Alex Blank
b483a34517
- apply_mount_permissions now skips mounts with readonly=true to avoid 'Read-only file system' warnings on post-start chown/chmod - Removed the ssh_keys mount from the pi-agent manifest definition; instance-level SSH key mounting now handles this exclusively - Added unit test for read-only mount skipping Quality gates: pytest (15 passed)
238 lines
8.0 KiB
Python
238 lines
8.0 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,
|
|
apply_ssh_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_readonly_mount(self, mock_run) -> None:
|
|
mounts = [
|
|
{
|
|
"name": "ssh_keys",
|
|
"target": "/home/user/.ssh",
|
|
"readonly": True,
|
|
"mode": "0700",
|
|
"file_mode": "0600",
|
|
},
|
|
]
|
|
results = apply_mount_permissions("abc123", mounts)
|
|
|
|
assert len(results) == 1
|
|
assert results[0]["mount_name"] == "ssh_keys"
|
|
assert results[0]["success"] is True
|
|
mock_run.assert_not_called()
|
|
|
|
@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 TestApplySshPermissions:
|
|
"""Tests for apply_ssh_permissions."""
|
|
|
|
@patch("subprocess.run")
|
|
def test_applies_chown_chmod_and_file_mode(self, mock_run) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
|
result = apply_ssh_permissions("abc123", "/home/user/.ssh", "user")
|
|
|
|
assert result["success"] is True
|
|
# 3 fix commands + 3 verification commands
|
|
assert mock_run.call_count == 6
|
|
chown_cmd = mock_run.call_args_list[0][0][0]
|
|
chmod_cmd = mock_run.call_args_list[1][0][0]
|
|
file_mode_cmd = mock_run.call_args_list[2][0][0]
|
|
|
|
assert chown_cmd == [
|
|
"docker",
|
|
"exec",
|
|
"--user",
|
|
"root",
|
|
"abc123",
|
|
"chown",
|
|
"-R",
|
|
"user:user",
|
|
"/home/user/.ssh",
|
|
]
|
|
assert chmod_cmd == [
|
|
"docker",
|
|
"exec",
|
|
"--user",
|
|
"root",
|
|
"abc123",
|
|
"chmod",
|
|
"700",
|
|
"/home/user/.ssh",
|
|
]
|
|
assert file_mode_cmd[0] == "docker"
|
|
assert (
|
|
"find /home/user/.ssh -name 'id_*' -type f -exec chmod 600"
|
|
in file_mode_cmd[-1]
|
|
)
|
|
|
|
@patch("subprocess.run")
|
|
def test_uses_root_user(self, mock_run) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
|
result = apply_ssh_permissions("abc123", "/root/.ssh", "root")
|
|
|
|
assert result["success"] is True
|
|
chown_cmd = mock_run.call_args_list[0][0][0]
|
|
assert chown_cmd == [
|
|
"docker",
|
|
"exec",
|
|
"--user",
|
|
"root",
|
|
"abc123",
|
|
"chown",
|
|
"-R",
|
|
"root:root",
|
|
"/root/.ssh",
|
|
]
|
|
|
|
@patch("subprocess.run")
|
|
def test_reports_failure(self, mock_run) -> None:
|
|
mock_run.return_value = MagicMock(
|
|
returncode=1, stdout="", stderr="chown failed"
|
|
)
|
|
|
|
result = apply_ssh_permissions("abc123", "/home/user/.ssh", "user")
|
|
|
|
assert result["success"] is False
|
|
assert "chown failed" in result["error"]
|
|
|
|
|
|
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
|