103 lines
4.7 KiB
Python
103 lines
4.7 KiB
Python
"""Opt-in regression against real user systemd, using only unique test units.
|
|
|
|
APP_BACKUP_TEST_USER_SYSTEMD=1 python3 -m unittest discover -s tests \
|
|
-p test_systemd_stop.py -v
|
|
No root, production services, or persistent unit files are involved.
|
|
"""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import uuid
|
|
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get('APP_BACKUP_TEST_USER_SYSTEMD') == '1',
|
|
'opt in to isolated real user-systemd test')
|
|
class SystemdStop(unittest.TestCase):
|
|
def setUp(self):
|
|
locale = patch.dict(os.environ, {'LC_ALL': 'C'})
|
|
locale.start()
|
|
self.addCleanup(locale.stop)
|
|
|
|
def exercise(self, protected, competing):
|
|
unit = 'app-backup-stop-test-' + uuid.uuid4().hex + '.service'
|
|
with tempfile.TemporaryDirectory(prefix='app-backup-stop-test-') as directory:
|
|
guard = Path(directory) / 'blocked'
|
|
stopper = None
|
|
try:
|
|
subprocess.run([
|
|
'systemd-run', '--user', '--quiet', '--unit=' + unit,
|
|
'--property=Type=oneshot', '--property=RemainAfterExit=yes',
|
|
'--property=ExecStop=/bin/sleep 3', '--property=TimeoutStopSec=10s',
|
|
'--property=ExecReload=/bin/true',
|
|
'--property=ConditionPathExists=!' + str(guard), '/bin/true',
|
|
], check=True, capture_output=True, text=True, timeout=15)
|
|
guard.touch()
|
|
if protected:
|
|
# Exercise production stop_service. The transient test unit
|
|
# already owns its condition; recovery persistence is covered
|
|
# by the backup integration tests, without root here.
|
|
command = ['bash', '-c', '''
|
|
source "$1/lib/common.sh"
|
|
systemctl() { command systemctl --user "$@"; }
|
|
install_service_guard() { :; }
|
|
save_recovery() { :; }
|
|
STOPPED_SERVICES=()
|
|
stop_service "$2"
|
|
''', 'stop-test', str(REPO), unit]
|
|
else:
|
|
command = ['systemctl', '--user', 'stop', unit]
|
|
stopper = subprocess.Popen(command, stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT, text=True)
|
|
deadline = time.monotonic() + 5
|
|
while True:
|
|
state = subprocess.run([
|
|
'systemctl', '--user', 'show', unit, '-p', 'ActiveState', '--value',
|
|
], capture_output=True, text=True, timeout=5).stdout.strip()
|
|
if state == 'deactivating':
|
|
break
|
|
self.assertIsNone(stopper.poll(), 'stop exited before concurrent request')
|
|
self.assertLess(time.monotonic(), deadline, 'stop never became pending')
|
|
time.sleep(.02)
|
|
competitor = subprocess.run(['systemctl', '--user', competing, unit],
|
|
capture_output=True, text=True, timeout=15)
|
|
output, _ = stopper.communicate(timeout=15)
|
|
if protected:
|
|
self.assertEqual(stopper.returncode, 0, output)
|
|
self.assertNotEqual(competitor.returncode, 0,
|
|
competitor.stdout + competitor.stderr)
|
|
self.assertIn('-- Service stopped: ' + unit, output)
|
|
state = subprocess.check_output([
|
|
'systemctl', '--user', 'show', unit,
|
|
'-p', 'ActiveState', '--value',
|
|
], text=True, timeout=5).strip()
|
|
self.assertEqual(state, 'inactive')
|
|
else:
|
|
self.assertNotEqual(stopper.returncode, 0, output)
|
|
self.assertIn('canceled', output.lower())
|
|
finally:
|
|
if stopper is not None and stopper.poll() is None:
|
|
stopper.terminate()
|
|
stopper.communicate(timeout=15)
|
|
# Only our UUID-named unit is stopped/reset. A canceled stop may
|
|
# still have its ExecStop process running, so wait for cleanup.
|
|
subprocess.run(['systemctl', '--user', 'stop', unit],
|
|
capture_output=True, timeout=15)
|
|
subprocess.run(['systemctl', '--user', 'reset-failed', unit],
|
|
capture_output=True, timeout=15)
|
|
|
|
def test_reload_cancels_unprotected_stop_despite_condition(self):
|
|
self.exercise(False, 'reload')
|
|
|
|
def test_reload_cannot_cancel_protected_stop(self):
|
|
self.exercise(True, 'reload')
|
|
|
|
def test_start_cannot_cancel_protected_stop(self):
|
|
self.exercise(True, 'start')
|