Your Name
110 lines
3.8 KiB
Python
110 lines
3.8 KiB
Python
"""Tests for file-manager elevation policy helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from server.domain.models import Server
|
|
from server.utils.files_elevation import (
|
|
FilesElevation,
|
|
apply_files_elevation_payload,
|
|
get_files_elevation,
|
|
merge_files_elevation_into_extra_attrs,
|
|
normalize_files_elevation,
|
|
)
|
|
|
|
|
|
def test_normalize_defaults_to_auto():
|
|
assert normalize_files_elevation(None) == FilesElevation.AUTO
|
|
assert normalize_files_elevation("bogus") == FilesElevation.AUTO
|
|
assert normalize_files_elevation("off") == FilesElevation.OFF
|
|
|
|
|
|
def test_get_files_elevation_from_extra_attrs():
|
|
server = Server(name="t", domain="1.2.3.4", extra_attrs={"files_elevation": "always_sudo"})
|
|
assert get_files_elevation(server) == FilesElevation.ALWAYS
|
|
|
|
|
|
def test_apply_files_elevation_payload_merges():
|
|
data = {"name": "x", "files_elevation": "off"}
|
|
apply_files_elevation_payload(data)
|
|
assert "files_elevation" not in data
|
|
assert data["extra_attrs"]["files_elevation"] == "off"
|
|
|
|
|
|
def test_merge_preserves_other_extra_attrs():
|
|
merged = merge_files_elevation_into_extra_attrs({"foo": 1}, "auto_sudo")
|
|
assert merged == {"foo": 1, "files_elevation": "auto_sudo"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_files_capability_root(monkeypatch):
|
|
from server.infrastructure.ssh import files_capability as fc
|
|
|
|
server = Server(name="t", domain="1.2.3.4", username="root")
|
|
|
|
async def fake_exec(*_a, **_k):
|
|
raise AssertionError("should not SSH when user is root")
|
|
|
|
monkeypatch.setattr(fc, "exec_ssh_command", fake_exec)
|
|
result = await fc.probe_files_capability(server)
|
|
assert result["is_root"] is True
|
|
assert result["can_sudo_nopasswd"] is True
|
|
assert result["whitelist_ok"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_files_capability_deploy(monkeypatch):
|
|
from server.infrastructure.ssh import files_capability as fc
|
|
|
|
server = Server(name="t", domain="1.2.3.4", username="deploy")
|
|
|
|
calls: list[str] = []
|
|
|
|
async def fake_exec(_server, command, timeout=15):
|
|
calls.append(command)
|
|
if command == "sudo -n true":
|
|
return {"exit_code": 0, "stdout": "", "stderr": ""}
|
|
return {"exit_code": 0, "stdout": "", "stderr": ""}
|
|
|
|
monkeypatch.setattr(fc, "exec_ssh_command", fake_exec)
|
|
result = await fc.probe_files_capability(server)
|
|
assert result["can_sudo_nopasswd"] is True
|
|
assert result["whitelist_ok"] is True
|
|
assert len(calls) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remote_shell_always_sudo(monkeypatch):
|
|
from server.infrastructure.ssh import remote_shell
|
|
|
|
server = Server(name="t", domain="1.2.3.4", username="deploy", extra_attrs={"files_elevation": "always_sudo"})
|
|
seen: list[str] = []
|
|
|
|
async def fake_exec(_server, command, timeout=120):
|
|
seen.append(command)
|
|
return {"exit_code": 0, "stdout": "ok", "stderr": ""}
|
|
|
|
monkeypatch.setattr(remote_shell, "exec_ssh_command", fake_exec)
|
|
result = await remote_shell.exec_ssh_command_with_fallback(server, "chmod 644 /tmp/x")
|
|
assert result["elevated"] is True
|
|
assert seen[0].startswith("sudo -n bash -c")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remote_shell_off_skips_sudo(monkeypatch):
|
|
from server.infrastructure.ssh import remote_shell
|
|
|
|
server = Server(name="t", domain="1.2.3.4", username="deploy", extra_attrs={"files_elevation": "off"})
|
|
call_count = 0
|
|
|
|
async def fake_exec(_server, command, timeout=120):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
return {"exit_code": 1, "stdout": "", "stderr": "permission denied"}
|
|
|
|
monkeypatch.setattr(remote_shell, "exec_ssh_command", fake_exec)
|
|
result = await remote_shell.exec_ssh_command_with_fallback(server, "chmod 644 /tmp/x")
|
|
assert call_count == 1
|
|
assert result["elevated"] is False
|