Restart active OWE user services after upgrades (#888)

tests/owe-package-restart.py

@@ -0,0 +1,94 @@
+#!/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()