mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 05:24:49 +09:00
Juan Denis
06de896175
Refactor services to use centralized system helpers (ServiceControl, PackageManager, run_privileged, is_command_available) instead of ad-hoc subprocess/os calls. Updated deployment, firewall, ftp, nginx, php, python, security, and ssl services to call the new utilities and added error/timeout handling improvements and package manager abstraction. Added backend/app/utils/system.py implementing the helpers and backend/tests/test_utils_system.py with unit tests. This centralizes privileged command execution and service/package management, simplifying tests and platform differences.
368 lines
16 KiB
Python
368 lines
16 KiB
Python
"""Tests for backend/app/utils/system.py.
|
|
|
|
Uses a direct import path so the test can run without Flask dependencies
|
|
(system.py itself has no Flask imports).
|
|
"""
|
|
|
|
import importlib
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from unittest import mock
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Import system.py directly to avoid triggering app/__init__.py (Flask deps).
|
|
# We register stub entries for 'app' and 'app.utils' so that
|
|
# unittest.mock.patch('app.utils.system.X') can resolve the path without
|
|
# importing the real app package (which needs Flask).
|
|
import types
|
|
|
|
_backend = os.path.join(os.path.dirname(__file__), os.pardir)
|
|
_mod_path = os.path.join(_backend, 'app', 'utils', 'system.py')
|
|
_spec = importlib.util.spec_from_file_location('app.utils.system', _mod_path)
|
|
_module = importlib.util.module_from_spec(_spec)
|
|
|
|
# Stub parent packages so patch() resolution never hits Flask imports.
|
|
if 'app' not in sys.modules:
|
|
sys.modules['app'] = types.ModuleType('app')
|
|
if 'app.utils' not in sys.modules:
|
|
_utils = types.ModuleType('app.utils')
|
|
sys.modules['app.utils'] = _utils
|
|
sys.modules['app'].utils = _utils # type: ignore[attr-defined]
|
|
|
|
sys.modules['app.utils.system'] = _module
|
|
sys.modules['app.utils'].system = _module # type: ignore[attr-defined]
|
|
_spec.loader.exec_module(_module)
|
|
|
|
run_privileged = _module.run_privileged
|
|
is_command_available = _module.is_command_available
|
|
PackageManager = _module.PackageManager
|
|
ServiceControl = _module.ServiceControl
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_privileged
|
|
# ---------------------------------------------------------------------------
|
|
class TestRunPrivileged:
|
|
"""Tests for :func:`run_privileged`."""
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_prepends_sudo_when_not_root(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['systemctl', 'restart', 'nginx'])
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'restart', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=0, create=True)
|
|
def test_skips_sudo_when_root(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['systemctl', 'restart', 'nginx'])
|
|
mock_run.assert_called_once_with(
|
|
['systemctl', 'restart', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_no_double_sudo(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['sudo', 'systemctl', 'restart', 'nginx'])
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'restart', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_defaults_applied(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['ls'])
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs['capture_output'] is True
|
|
assert kwargs['text'] is True
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_kwargs_passed_through(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['apt', 'install', '-y', 'foo'], timeout=120, check=True)
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs['timeout'] == 120
|
|
assert kwargs['check'] is True
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_caller_can_override_defaults(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged(['ls'], capture_output=False)
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs['capture_output'] is False
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_string_command_gets_sudo(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged('systemctl restart nginx')
|
|
args, _ = mock_run.call_args
|
|
assert args[0] == 'sudo systemctl restart nginx'
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_string_command_no_double_sudo(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
run_privileged('sudo systemctl restart nginx')
|
|
args, _ = mock_run.call_args
|
|
assert args[0] == 'sudo systemctl restart nginx'
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_returns_completed_process(self, _euid, mock_run):
|
|
expected = subprocess.CompletedProcess([], 0, stdout='ok')
|
|
mock_run.return_value = expected
|
|
result = run_privileged(['echo', 'ok'])
|
|
assert result is expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# is_command_available
|
|
# ---------------------------------------------------------------------------
|
|
class TestIsCommandAvailable:
|
|
"""Tests for :func:`is_command_available`."""
|
|
|
|
@patch('app.utils.system.shutil.which', return_value='/usr/bin/nginx')
|
|
def test_found_in_path(self, _which):
|
|
assert is_command_available('nginx') is True
|
|
|
|
@patch('app.utils.system.os.path.exists', return_value=True)
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_found_in_common_paths(self, _which, _exists):
|
|
assert is_command_available('firewall-cmd') is True
|
|
|
|
@patch('app.utils.system.os.path.exists', return_value=False)
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_not_found(self, _which, _exists):
|
|
assert is_command_available('nonexistent') is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PackageManager
|
|
# ---------------------------------------------------------------------------
|
|
class TestPackageManager:
|
|
"""Tests for :class:`PackageManager`."""
|
|
|
|
def setup_method(self):
|
|
PackageManager.reset_cache()
|
|
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_detect_apt(self, _which):
|
|
assert PackageManager.detect() == 'apt'
|
|
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/dnf' if c == 'dnf' else None)
|
|
def test_detect_dnf(self, _which):
|
|
assert PackageManager.detect() == 'dnf'
|
|
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_detect_none(self, _which):
|
|
assert PackageManager.detect() is None
|
|
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_is_available_true(self, _which):
|
|
assert PackageManager.is_available() is True
|
|
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_is_available_false(self, _which):
|
|
assert PackageManager.is_available() is False
|
|
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_detect_caches(self, mock_which):
|
|
PackageManager.detect()
|
|
PackageManager.detect()
|
|
# shutil.which should only be called during the first detect()
|
|
assert mock_which.call_count <= 3 # at most apt/dnf/yum on first call
|
|
|
|
# -- is_installed --
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_is_installed_apt_true(self, _which, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess(
|
|
[], 0, stdout='Status: install ok installed\n',
|
|
)
|
|
assert PackageManager.is_installed('nginx') is True
|
|
mock_run.assert_called_once_with(
|
|
['dpkg', '-s', 'nginx'], capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_is_installed_apt_false(self, _which, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 1, stdout='')
|
|
assert PackageManager.is_installed('nginx') is False
|
|
|
|
@patch('app.utils.system.subprocess.run', side_effect=FileNotFoundError)
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_is_installed_apt_dpkg_missing(self, _which, _run):
|
|
assert PackageManager.is_installed('nginx') is False
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/dnf' if c == 'dnf' else None)
|
|
def test_is_installed_rpm_true(self, _which, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout='nginx-1.0\n')
|
|
assert PackageManager.is_installed('nginx') is True
|
|
|
|
@patch('app.utils.system.subprocess.run', side_effect=FileNotFoundError)
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/dnf' if c == 'dnf' else None)
|
|
def test_is_installed_rpm_missing(self, _which, _run):
|
|
assert PackageManager.is_installed('nginx') is False
|
|
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_is_installed_no_manager(self, _which):
|
|
assert PackageManager.is_installed('nginx') is False
|
|
|
|
# -- install --
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/apt' if c == 'apt' else None)
|
|
def test_install_apt(self, _which, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
result = PackageManager.install(['nginx', 'curl'])
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'apt', 'install', '-y', 'nginx', 'curl'],
|
|
capture_output=True, text=True, timeout=300,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
@patch('app.utils.system.shutil.which', side_effect=lambda c: '/usr/bin/dnf' if c == 'dnf' else None)
|
|
def test_install_dnf(self, _which, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
PackageManager.install('nginx')
|
|
args = mock_run.call_args[0][0]
|
|
assert args == ['sudo', 'dnf', 'install', '-y', 'nginx']
|
|
|
|
@patch('app.utils.system.shutil.which', return_value=None)
|
|
def test_install_no_manager_raises(self, _which):
|
|
with pytest.raises(RuntimeError, match='No supported package manager'):
|
|
PackageManager.install('nginx')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ServiceControl
|
|
# ---------------------------------------------------------------------------
|
|
class TestServiceControl:
|
|
"""Tests for :class:`ServiceControl`."""
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_start(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.start('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'start', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_stop(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.stop('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'stop', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_restart(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.restart('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'restart', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_reload(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.reload('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'reload', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_enable(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.enable('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'enable', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_disable(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.disable('nginx')
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'disable', 'nginx'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_daemon_reload(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.daemon_reload()
|
|
mock_run.assert_called_once_with(
|
|
['sudo', 'systemctl', 'daemon-reload'],
|
|
capture_output=True, text=True,
|
|
)
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
@patch('app.utils.system.os.geteuid', return_value=1000, create=True)
|
|
def test_restart_with_kwargs(self, _euid, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0)
|
|
ServiceControl.restart('nginx', check=True, timeout=30)
|
|
_, kwargs = mock_run.call_args
|
|
assert kwargs['check'] is True
|
|
assert kwargs['timeout'] == 30
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
def test_is_active_true(self, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout='active\n')
|
|
assert ServiceControl.is_active('nginx') is True
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
def test_is_active_false(self, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 3, stdout='inactive\n')
|
|
assert ServiceControl.is_active('nginx') is False
|
|
|
|
@patch('app.utils.system.subprocess.run', side_effect=FileNotFoundError)
|
|
def test_is_active_missing_systemctl(self, _run):
|
|
assert ServiceControl.is_active('nginx') is False
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
def test_is_enabled_true(self, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout='enabled\n')
|
|
assert ServiceControl.is_enabled('nginx') is True
|
|
|
|
@patch('app.utils.system.subprocess.run')
|
|
def test_is_enabled_false(self, mock_run):
|
|
mock_run.return_value = subprocess.CompletedProcess([], 1, stdout='disabled\n')
|
|
assert ServiceControl.is_enabled('nginx') is False
|
|
|
|
@patch('app.utils.system.subprocess.run', side_effect=FileNotFoundError)
|
|
def test_is_enabled_missing_systemctl(self, _run):
|
|
assert ServiceControl.is_enabled('nginx') is False
|