Restore linked input settings and attribute test notifications to Fcitx

test/acceptance.d/input-method-notifications.py

@@ -0,0 +1,30 @@
+"""Select notifications sent by Fcitx, including addons with custom app names."""
+
+from pathlib import Path
+import re
+import sys
+
+
+def input_notifications(text, initial_owner):
+  calls = re.split(r"(?m)^(?=(?:method call|signal|method return|error)\b)", text)
+  owners = {initial_owner}
+  for call in calls:
+    if "interface=org.freedesktop.DBus; member=NameOwnerChanged" in call.split("\n", 1)[0]:
+      names = re.findall(r'^\s*string "([^"]*)"', call, re.MULTILINE)
+      if len(names) == 3 and names[0] == "org.fcitx.Fcitx5":
+        owners.update(names[1:])
+  owners.discard("")
+  for call in calls:
+    header = call.split("\n", 1)[0]
+    if "interface=org.freedesktop.Notifications; member=Notify" not in header:
+      continue
+    sender = re.search(r"\bsender=(\S+)", header)
+    if sender and sender[1] in owners:
+      yield call
+
+
+if __name__ == "__main__":
+  found = list(input_notifications(Path(sys.argv[1]).read_text(errors="replace"), sys.argv[2]))
+  if found:
+    print("\n".join(found))
+  raise SystemExit(1 if found else 0)

test/acceptance.d/input-method-state.py

@@ -0,0 +1,85 @@
+"""Snapshot test-owned config paths, including targets of nested symlinks."""
+
+import filecmp
+import json
+from pathlib import Path
+import shutil
+import sys
+
+
+def copy(source, destination):
+  if source.is_symlink():
+    destination.symlink_to(source.readlink())
+  elif source.is_dir():
+    shutil.copytree(source, destination, symlinks=True)
+  elif source.exists():
+    shutil.copy2(source, destination)
+
+
+def backup(directory, paths):
+  directory.mkdir()
+  recorded = {}
+
+  def capture(path):
+    path = path.absolute()
+    if str(path) in recorded:
+      return
+    name = str(len(recorded))
+    recorded[str(path)] = name
+    copy(path, directory / name)
+    if not path.parent.exists():
+      capture(path.parent)
+    # Setup writes to the resolved target; intermediate links are unchanged.
+    resolved = path.resolve()
+    if resolved != path:
+      capture(resolved)
+    if path.is_dir() and not path.is_symlink():
+      for child in path.rglob("*"):
+        if child.is_symlink():
+          capture(child)
+
+  for path in paths:
+    capture(path)
+  (directory / "paths.json").write_text(json.dumps(recorded))
+
+
+def unchanged(saved, original):
+  if saved.is_symlink() or original.is_symlink():
+    return saved.is_symlink() and original.is_symlink() and saved.readlink() == original.readlink()
+  if not saved.exists() or not original.exists():
+    return not saved.exists() and not original.exists()
+  if saved.stat().st_mode != original.stat().st_mode:
+    return False
+  if saved.is_file() and original.is_file():
+    return filecmp.cmp(saved, original, shallow=False)
+  if saved.is_dir() and original.is_dir():
+    names = {child.name for child in saved.iterdir()}
+    return names == {child.name for child in original.iterdir()} and all(
+      unchanged(saved / name, original / name) for name in names
+    )
+  return False
+
+
+def restore(directory):
+  for original, name in json.loads((directory / "paths.json").read_text()).items():
+    path = Path(original)
+    saved = directory / name
+    if unchanged(saved, path):
+      continue
+    if path.is_symlink() or path.is_file():
+      path.unlink()
+    elif path.is_dir():
+      shutil.rmtree(path)
+    if saved.exists() or saved.is_symlink():
+      path.parent.mkdir(parents=True, exist_ok=True)
+      copy(saved, path)
+
+
+if __name__ == "__main__":
+  action, directory, *paths = sys.argv[1:]
+  if action == "backup":
+    backup(Path(directory), [Path(path) for path in paths])
+  elif action == "restore":
+    restore(Path(directory))
+  else:
+    raise SystemExit("Expected backup or restore")

test/acceptance.d/input-methods-test.sh

@@ -19,36 +19,42 @@ fi
 pass "installer input selection reaches the first desktop session"
 
 work=$(mktemp -d)
-dbus-monitor --session "type='method_call',interface='org.freedesktop.Notifications',member='Notify'" > "$ARTIFACTS/input-method-notifications.log" 2>&1 &
-notification_monitor=$!
 config_home=${XDG_CONFIG_HOME:-$HOME/.config}
 data_home=${XDG_DATA_HOME:-$HOME/.local/share}
 paths=("$config_home/fcitx5" "$config_home/fontconfig/conf.d/50-omarchy-input-method.conf" "$data_home/dbus-1/services/org.fcitx.Fcitx5.service")
-for (( i = 0; i < ${#paths[@]}; i++ )); do
-  if [[ -e ${paths[i]} || -L ${paths[i]} ]]; then
-    cp -a "${paths[i]}" "$work/backup-$i"
-  fi
-done
+python "$ROOT/test/acceptance.d/input-method-state.py" backup "$work/state" "${paths[@]}"
 
+notification_monitor=""
 cleanup() {
-  kill "$notification_monitor" 2>/dev/null || true
-  wait "$notification_monitor" 2>/dev/null || true
+  local status=$?
+  if [[ -n $notification_monitor ]]; then
+    kill "$notification_monitor" 2>/dev/null || true
+    wait "$notification_monitor" 2>/dev/null || true
+  fi
   close_windows '^org\.omarchy\.ime-test$'
   close_windows '^org\.omarchy\.ime-test-gtk$|^org\.qt-project\.qml$|^chrome-.*_omarchy-ime-entry\.html-Default$'
-  systemctl --user stop omarchy-fcitx5.service
-  for (( i = 0; i < ${#paths[@]}; i++ )); do
-    rm -rf "${paths[i]}"
-    if [[ -e $work/backup-$i || -L $work/backup-$i ]]; then
-      mkdir -p "$(dirname "${paths[i]}")"
-      cp -a "$work/backup-$i" "${paths[i]}"
-    fi
-  done
-  busctl --user call org.freedesktop.DBus /org/freedesktop/DBus org.freedesktop.DBus ReloadConfig >/dev/null
-  systemctl --user start omarchy-fcitx5.service
-  rm -rf "$work"
+  systemctl --user stop omarchy-fcitx5.service || status=1
+  python "$ROOT/test/acceptance.d/input-method-state.py" restore "$work/state" || status=1
+  busctl --user call org.freedesktop.DBus /org/freedesktop/DBus org.freedesktop.DBus ReloadConfig >/dev/null || status=1
+  systemctl --user start omarchy-fcitx5.service || status=1
+  if (( status == 0 )); then
+    rm -rf "$work"
+  else
+    printf 'Input test snapshot retained at %s\n' "$work/state" >&2
+  fi
+  exit "$status"
 }
 trap cleanup EXIT
 
+dbus-monitor --session "type='method_call',interface='org.freedesktop.Notifications',member='Notify'" \
+  "type='signal',interface='org.freedesktop.DBus',member='NameOwnerChanged',arg0='org.fcitx.Fcitx5'" > "$ARTIFACTS/input-method-notifications.log" 2>&1 &
+notification_monitor=$!
+monitor_ready() {
+  kill -0 "$notification_monitor" && grep -q 'member=NameAcquired' "$ARTIFACTS/input-method-notifications.log"
+}
+wait_until "notification monitor attaches" 10 monitor_ready
+initial_owner=$(busctl --user call org.freedesktop.DBus /org/freedesktop/DBus org.freedesktop.DBus GetNameOwner s org.fcitx.Fcitx5 | cut -d '"' -f 2)
+
 cat > "$work/reader" <<'SH'
 #!/bin/bash
 IFS= read -r text
@@ -97,7 +103,8 @@ HTML
 compose() {
   fcitx5-remote -s "$method"
   fcitx5-remote -c
-  wtype -M ctrl -M shift -k space -m shift -m ctrl
+  # Use the controller for composition; wtype cannot prove global shortcuts.
+  fcitx5-remote -o
   case "$method" in
     mozc) wtype "nihongo"; wtype -k space ;;
     hangul) wtype "gksrmf" ;;
@@ -168,7 +175,12 @@ for method in mozc hangul pinyin chewing; do
   done
 done
 
-if grep -q 'member=Notify' "$ARTIFACTS/input-method-notifications.log"; then
+# Stop and drain the monitor before reading a complete final call.
+kill -0 "$notification_monitor" 2>/dev/null || fail "notification monitor stays attached"
+kill "$notification_monitor" 2>/dev/null || true
+wait "$notification_monitor" 2>/dev/null || true
+notification_monitor=""
+if ! python "$ROOT/test/acceptance.d/input-method-notifications.py" "$ARTIFACTS/input-method-notifications.log" "$initial_owner"; then
   fail "input setup and composition produce no additional notifications" "$(cat "$ARTIFACTS/input-method-notifications.log")"
 fi
 pass "all four input engines remain quiet during setup and composition"

test/shell.d/input-acceptance-checks.py

@@ -0,0 +1,144 @@
+"""Regression fixtures for acceptance cleanup and notification attribution."""
+
+from pathlib import Path
+import runpy
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+
+ROOT = Path(__file__).resolve().parents[2]
+state = runpy.run_path(str(ROOT / "test/acceptance.d/input-method-state.py"))
+notifications = runpy.run_path(str(ROOT / "test/acceptance.d/input-method-notifications.py"))
+
+
+class AcceptanceHelpers(unittest.TestCase):
+  def setUp(self):
+    self.temp = tempfile.TemporaryDirectory()
+    self.addCleanup(self.temp.cleanup)
+    self.root = Path(self.temp.name)
+
+  def test_nested_linked_settings_restore_targets_links_and_modes(self):
+    config = self.root / "fcitx5"
+    config.mkdir()
+    target = self.root / "dotfiles-config"
+    target.write_text("custom switching keys")
+    target.chmod(0o600)
+    link = config / "config"
+    link.symlink_to("../dotfiles-config")
+    profile = config / "profile"
+    profile.write_text("custom input methods")
+    snapshot = self.root / "snapshot"
+    state["backup"](snapshot, [config])
+    target.write_text("test keys")
+    target.chmod(0o644)
+    profile.write_text("test methods")
+    (config / "new-engine.conf").write_text("created by test")
+    state["restore"](snapshot)
+    self.assertTrue(link.is_symlink())
+    self.assertEqual(link.readlink(), Path("../dotfiles-config"))
+    self.assertEqual(target.read_text(), "custom switching keys")
+    self.assertEqual(target.stat().st_mode & 0o777, 0o600)
+    self.assertEqual(profile.read_text(), "custom input methods")
+    self.assertFalse((config / "new-engine.conf").exists())
+
+  def test_directory_links_and_external_font_preferences_are_restored(self):
+    target = self.root / "dotfiles"
+    target.mkdir()
+    (target / "profile").write_text("original")
+    config = self.root / "fcitx5"
+    config.symlink_to(target)
+    font_target = self.root / "font-preference"
+    font_target.write_text("original font")
+    font = self.root / "font.conf"
+    font.symlink_to(font_target)
+    snapshot = self.root / "snapshot"
+    state["backup"](snapshot, [config, font])
+    (target / "profile").write_text("changed")
+    font_target.write_text("test font")
+    state["restore"](snapshot)
+    self.assertTrue(config.is_symlink())
+    self.assertTrue(font.is_symlink())
+    self.assertEqual((target / "profile").read_text(), "original")
+    self.assertEqual(font_target.read_text(), "original font")
+
+  def test_missing_targets_and_paths_remain_missing(self):
+    absent = self.root / "absent"
+    target = self.root / "missing-target"
+    link = self.root / "dangling"
+    link.symlink_to(target)
+    snapshot = self.root / "snapshot"
+    state["backup"](snapshot, [absent, link])
+    absent.mkdir()
+    (absent / "test").write_text("new")
+    target.write_text("created through link")
+    state["restore"](snapshot)
+    self.assertFalse(absent.exists())
+    self.assertTrue(link.is_symlink())
+    self.assertFalse(target.exists())
+
+  def test_failed_restore_still_restarts_the_service(self):
+    script = (ROOT / "test/acceptance.d/input-methods-test.sh").read_text()
+    cleanup = "cleanup() {" + script.split("cleanup() {", 1)[1].split("\n}\ntrap cleanup", 1)[0] + "\n}"
+    commands = """
+notification_monitor=""
+work=$1
+close_windows() { :; }
+systemctl() { printf '%s\n' "$*" >> "$work/service.log"; }
+python() { return 1; }
+busctl() { :; }
+"""
+    result = subprocess.run(["bash", "-ec", commands + cleanup + "\ncleanup", "test", str(self.root)], capture_output=True, text=True)
+    self.assertEqual(result.returncode, 1)
+    self.assertIn("start omarchy-fcitx5.service", (self.root / "service.log").read_text())
+    self.assertIn("snapshot retained", result.stderr)
+
+  def test_unchanged_link_targets_are_not_rewritten(self):
+    target = self.root / "read-only-config"
+    target.write_text("unchanged")
+    target.chmod(0o444)
+    link = self.root / "config"
+    link.symlink_to(target)
+    snapshot = self.root / "snapshot"
+    state["backup"](snapshot, [link])
+    with patch.object(Path, "unlink", side_effect=PermissionError("read-only")):
+      state["restore"](snapshot)
+    self.assertTrue(link.is_symlink())
+    self.assertEqual(target.read_text(), "unchanged")
+
+  def test_new_parent_directories_are_removed(self):
+    service = self.root / "dbus-1/services/org.fcitx.Fcitx5.service"
+    snapshot = self.root / "snapshot"
+    state["backup"](snapshot, [service])
+    service.parent.mkdir(parents=True)
+    service.write_text("created by setup")
+    state["restore"](snapshot)
+    self.assertFalse((self.root / "dbus-1").exists())
+
+  def test_notifications_follow_process_ownership_across_restarts(self):
+    def call(sender, app, body):
+      return f'''method call time=1 sender={sender} -> destination=org.freedesktop.Notifications serial=1 path=/org/freedesktop/Notifications; interface=org.freedesktop.Notifications; member=Notify
+   string "{app}"
+   uint32 0
+   string "icon"
+   string "summary"
+   string "{body}"
+'''
+    reminder = call(":1.99", "Fcitx", "unrelated reminder")
+    first = call(":1.42", "Input Method", "input setup prompt")
+    second = call(":1.43", "拼音", "cloud input prompt")
+    changed = '''signal time=2 sender=org.freedesktop.DBus -> destination=(null destination) serial=2 path=/org/freedesktop/DBus; interface=org.freedesktop.DBus; member=NameOwnerChanged
+   string "org.fcitx.Fcitx5"
+   string ":1.42"
+   string ":1.43"
+'''
+    selected = list(notifications["input_notifications"](reminder + first + changed + second, ":1.42"))
+    self.assertEqual(selected, [first, second])
+    self.assertEqual(list(notifications["input_notifications"](reminder, ":1.42")), [])
+    self.assertEqual(list(notifications["input_notifications"]("", ":1.42")), [])
+
+
+
+if __name__ == "__main__":
+  unittest.main()

test/shell.d/input-acceptance-test.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+set -euo pipefail
+source "$(dirname "${BASH_SOURCE[0]}")/base-test.sh"
+python "$ROOT/test/shell.d/input-acceptance-checks.py"