nik/FreedomBox
1
0
Fork 0
mirror of https://github.com/freedombox/FreedomBox.git synced 2026-05-27 10:44:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

action_utils: Allow checking result for service operations

Browse Source 
Tests:

- Functional tests for bepasty, nextcloud, torproxy app work.

- Adding/removing a domains works.

- After first setup, nscd service is masked.

- Re-run setup on miniflux app works.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa 2025-05-15 11:57:03 -07:00 committed by James Valleroy
parent 3c9814cb78
commit 16f90d863b
No known key found for this signature in database
GPG Key ID: 77C0C75E7B650808
3 changed files with 60 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
plinth/action_utils.py
View File

@ -111,73 +111,73 @@ def service_is_enabled(service_name, strict_check=False):
return False
def service_enable(service_name):
def service_enable(service_name: str, check: bool = False):
"""Enable and start a service in systemd."""
subprocess.call(['systemctl', 'enable', service_name])
service_start(service_name)
subprocess.run(['systemctl', 'enable', service_name], check=check)
service_start(service_name, check=check)
def service_disable(service_name):
def service_disable(service_name: str, check: bool = False):
"""Disable and stop service in systemd."""
subprocess.call(['systemctl', 'disable', service_name])
subprocess.run(['systemctl', 'disable', service_name], check=check)
try:
service_stop(service_name)
service_stop(service_name, check=check)
except subprocess.CalledProcessError:
pass
def service_mask(service_name):
def service_mask(service_name: str, check: bool = False):
"""Mask a service"""
subprocess.call(['systemctl', 'mask', service_name])
subprocess.run(['systemctl', 'mask', service_name], check=check)
def service_unmask(service_name):
def service_unmask(service_name: str, check: bool = False):
"""Unmask a service"""
subprocess.call(['systemctl', 'unmask', service_name])
subprocess.run(['systemctl', 'unmask', service_name], check=check)
def service_start(service_name):
def service_start(service_name: str, check: bool = False):
"""Start a service with systemd."""
service_action(service_name, 'start')
service_action(service_name, 'start', check=check)
def service_stop(service_name):
def service_stop(service_name: str, check: bool = False):
"""Stop a service with systemd."""
service_action(service_name, 'stop')
service_action(service_name, 'stop', check=check)
def service_restart(service_name):
def service_restart(service_name: str, check: bool = False):
"""Restart a service with systemd."""
service_action(service_name, 'restart')
service_action(service_name, 'restart', check=check)
def service_try_restart(service_name):
def service_try_restart(service_name: str, check: bool = False):
"""Try to restart a service with systemd."""
service_action(service_name, 'try-restart')
service_action(service_name, 'try-restart', check=check)
def service_reload(service_name):
def service_reload(service_name: str, check: bool = False):
"""Reload a service with systemd."""
service_action(service_name, 'reload')
service_action(service_name, 'reload', check=check)
def service_try_reload_or_restart(service_name):
def service_try_reload_or_restart(service_name: str, check: bool = False):
"""Reload a service if it supports reloading, otherwise restart.
Do nothing if service is not running.
"""
service_action(service_name, 'try-reload-or-restart')
service_action(service_name, 'try-reload-or-restart', check=check)
def service_reset_failed(service_name):
def service_reset_failed(service_name: str, check: bool = False):
"""Reset the 'failed' state of units."""
service_action(service_name, 'reset-failed')
service_action(service_name, 'reset-failed', check=check)
def service_action(service_name, action):
def service_action(service_name: str, action: str, check: bool = False):
"""Perform the given action on the service_name."""
subprocess.run(['systemctl', action, service_name],
stdout=subprocess.DEVNULL, check=False)
stdout=subprocess.DEVNULL, check=check)
def webserver_is_enabled(name, kind='config'):

12
plinth/tests/test_action_utils.py
View File

@ -78,17 +78,17 @@ def test_service_enable_and_disable():
def test_service_actions(mock):
"""Trivial white box test for trivial implementations."""
service_start(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'start')
mock.assert_called_with(UNKNOWN, 'start', check=False)
service_stop(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'stop')
mock.assert_called_with(UNKNOWN, 'stop', check=False)
service_restart(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'restart')
mock.assert_called_with(UNKNOWN, 'restart', check=False)
service_try_restart(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'try-restart')
mock.assert_called_with(UNKNOWN, 'try-restart', check=False)
service_reload(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'reload')
mock.assert_called_with(UNKNOWN, 'reload', check=False)
service_try_reload_or_restart(UNKNOWN)
mock.assert_called_with(UNKNOWN, 'try-reload-or-restart')
mock.assert_called_with(UNKNOWN, 'try-reload-or-restart', check=False)
@pytest.mark.usefixtures('needs_root')

60
plinth/tests/test_daemon.py
View File

@ -79,22 +79,24 @@ def test_is_enabled(service_is_enabled, daemon):
@patch('plinth.app.apps_init')
@patch('subprocess.run')
@patch('subprocess.call')
def test_enable(subprocess_call, subprocess_run, apps_init, app_list,
mock_privileged, daemon):
def test_enable(subprocess_run, apps_init, app_list, mock_privileged, daemon):
"""Test that enabling the daemon works."""
daemon.enable()
subprocess_call.assert_has_calls(
[call(['systemctl', 'enable', 'test-unit'])])
subprocess_run.assert_has_calls(
[call(['systemctl', 'enable', 'test-unit'], check=False)])
subprocess_run.assert_any_call(['systemctl', 'start', 'test-unit'],
stdout=subprocess.DEVNULL, check=False)
subprocess_call.reset_mock()
subprocess_run.reset_mock()
daemon.alias = 'test-unit-2'
daemon.enable()
subprocess_call.assert_has_calls([
call(['systemctl', 'enable', 'test-unit']),
call(['systemctl', 'enable', 'test-unit-2'])
subprocess_run.assert_has_calls([
call(['systemctl', 'enable', 'test-unit'], check=False),
call(['systemctl', 'start', 'test-unit'], stdout=subprocess.DEVNULL,
check=False),
call(['systemctl', 'enable', 'test-unit-2'], check=False),
call(['systemctl', 'start', 'test-unit-2'], stdout=subprocess.DEVNULL,
check=False),
])
subprocess_run.assert_any_call(['systemctl', 'start', 'test-unit'],
stdout=subprocess.DEVNULL, check=False)
@ -104,22 +106,24 @@ def test_enable(subprocess_call, subprocess_run, apps_init, app_list,
@patch('plinth.app.apps_init')
@patch('subprocess.run')
@patch('subprocess.call')
def test_disable(subprocess_call, subprocess_run, apps_init, app_list,
mock_privileged, daemon):
def test_disable(subprocess_run, apps_init, app_list, mock_privileged, daemon):
"""Test that disabling the daemon works."""
daemon.disable()
subprocess_call.assert_has_calls(
[call(['systemctl', 'disable', 'test-unit'])])
subprocess_run.assert_has_calls(
[call(['systemctl', 'disable', 'test-unit'], check=False)])
subprocess_run.assert_any_call(['systemctl', 'stop', 'test-unit'],
stdout=subprocess.DEVNULL, check=False)
subprocess_call.reset_mock()
subprocess_run.reset_mock()
daemon.alias = 'test-unit-2'
daemon.disable()
subprocess_call.assert_has_calls([
call(['systemctl', 'disable', 'test-unit']),
call(['systemctl', 'disable', 'test-unit-2'])
subprocess_run.assert_has_calls([
call(['systemctl', 'disable', 'test-unit'], check=False),
call(['systemctl', 'stop', 'test-unit'], stdout=subprocess.DEVNULL,
check=False),
call(['systemctl', 'disable', 'test-unit-2'], check=False),
call(['systemctl', 'stop', 'test-unit-2'], stdout=subprocess.DEVNULL,
check=False),
])
subprocess_run.assert_any_call(['systemctl', 'stop', 'test-unit'],
stdout=subprocess.DEVNULL, check=False)
@ -141,38 +145,30 @@ def test_is_running(service_is_running, daemon):
@patch('plinth.app.apps_init')
@patch('plinth.action_utils.service_is_running')
@patch('subprocess.run')
@patch('subprocess.call')
def test_ensure_running(subprocess_call, subprocess_run, service_is_running,
apps_init, app_list, mock_privileged, daemon):
def test_ensure_running(subprocess_run, service_is_running, apps_init,
app_list, mock_privileged, daemon):
"""Test that checking that the daemon is running works."""
service_is_running.return_value = True
with daemon.ensure_running() as starting_state:
assert starting_state
assert not subprocess_call.called
assert not subprocess_run.called
assert not subprocess_call.called
assert not subprocess_run.called
service_is_running.return_value = False
with daemon.ensure_running() as starting_state:
assert not starting_state
assert subprocess_run.mock_calls == [
call(['systemctl', 'enable', 'test-unit'], check=False),
call(['systemctl', 'start', 'test-unit'],
stdout=subprocess.DEVNULL, check=False)
]
assert subprocess_call.mock_calls == [
call(['systemctl', 'enable', 'test-unit'])
stdout=subprocess.DEVNULL, check=False),
]
subprocess_run.reset_mock()
subprocess_call.reset_mock()
assert subprocess_run.mock_calls == [
call(['systemctl', 'disable', 'test-unit'], check=False),
call(['systemctl', 'stop', 'test-unit'], stdout=subprocess.DEVNULL,
check=False)
]
assert subprocess_call.mock_calls == [
call(['systemctl', 'disable', 'test-unit'])
check=False),
]