95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Package restart fixtures; no real systemd managers are contacted."""
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PACKAGE = ROOT / "pkgbuilds/owe"
|
|
|
|
|
|
class Restart(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix="owe-restart-")
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.log = self.root / "calls"
|
|
self.env = dict(os.environ, PATH=f"{self.root}:{os.environ['PATH']}", CALLS=str(self.log))
|
|
self.command("systemd-detect-virt", 'exit "${CHROOT:-1}"')
|
|
self.command("systemd-notify", 'exit "${BOOTED:-0}"')
|
|
self.command("systemctl", '''
|
|
printf '%s\\n' "$*" >> "$CALLS"
|
|
case "$1" in
|
|
list-units)
|
|
printf '%s\\n' "${MANAGERS:-}"
|
|
exit "${LIST_FAILED:-0}"
|
|
;;
|
|
esac
|
|
[[ $* != *"--machine=${FAIL_USER:-none}@.host"* ]]
|
|
''')
|
|
|
|
def command(self, name, body):
|
|
path = self.root / name
|
|
path.write_text("#!/bin/bash\n" + body + "\n")
|
|
path.chmod(0o755)
|
|
|
|
def run_hook(self, **values):
|
|
return subprocess.run(["bash", str(PACKAGE / "package-restart")],
|
|
env=self.env | values, capture_output=True, text=True)
|
|
|
|
def calls(self):
|
|
return self.log.read_text().splitlines() if self.log.exists() else []
|
|
|
|
def test_offline_and_chroot_skip_managers(self):
|
|
for values in ({"CHROOT": "0"}, {"BOOTED": "1"}):
|
|
with self.subTest(values=values):
|
|
self.assertEqual(self.run_hook(**values).returncode, 0)
|
|
self.assertEqual(self.calls(), [])
|
|
|
|
def test_no_running_managers(self):
|
|
self.assertEqual(self.run_hook().returncode, 0)
|
|
self.assertEqual(len(self.calls()), 1)
|
|
|
|
def test_all_running_managers_use_conditional_restart(self):
|
|
result = self.run_hook(MANAGERS="user@1000.service loaded active running User Manager\nuser@1001.service loaded active running User Manager")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(self.calls(), [
|
|
"list-units user@*.service --state=running --no-legend --plain",
|
|
"--user --machine=1000@.host try-restart owed.service",
|
|
"--user --machine=1001@.host try-restart owed.service",
|
|
])
|
|
|
|
def test_failure_reports_user_and_continues(self):
|
|
result = self.run_hook(MANAGERS="user@1000.service loaded active running\nuser@1001.service loaded active running", FAIL_USER="1000")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Could not restart OWE for user 1000", result.stderr)
|
|
self.assertIn("--user --machine=1001@.host try-restart owed.service", self.calls())
|
|
|
|
def test_list_failure_is_not_silently_successful(self):
|
|
self.assertNotEqual(self.run_hook(LIST_FAILED="1").returncode, 0)
|
|
self.assertEqual(len(self.calls()), 1)
|
|
|
|
def test_only_user_manager_units_are_accepted(self):
|
|
self.assertEqual(self.run_hook(MANAGERS="invalid.service loaded active running\nuser@oops.service loaded active running").returncode, 0)
|
|
self.assertEqual(len(self.calls()), 1)
|
|
|
|
def test_packaged_hook_and_sources(self):
|
|
hook = (PACKAGE / "90-owe-restart.hook").read_text()
|
|
self.assertIn("Operation = Upgrade\n", hook)
|
|
self.assertNotIn("Operation = Install", hook)
|
|
self.assertNotIn("Operation = Remove", hook)
|
|
self.assertIn("When = PostTransaction\n", hook)
|
|
self.assertIn("Exec = /usr/lib/owe/package-restart\n", hook)
|
|
recipe = (PACKAGE / "PKGBUILD").read_text()
|
|
for source in ("90-owe-restart.hook", "package-restart"):
|
|
self.assertIn(hashlib.sha256((PACKAGE / source).read_bytes()).hexdigest(), recipe)
|
|
self.assertIn('"$pkgdir/usr/share/libalpm/hooks/90-owe-restart.hook"', recipe)
|
|
self.assertIn('"$pkgdir/usr/lib/owe/package-restart"', recipe)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|