From 32d81c76c68ec491e7a3dc2b0dd1cfbe212e24c6 Mon Sep 17 00:00:00 2001 From: David Heinemeier Hansson Date: Fri, 9 Oct 2026 13:41:17 +0200 Subject: [PATCH] Restore linked input settings and attribute test notifications to Fcitx --- .../input-method-notifications.py | 30 ++++ test/acceptance.d/input-method-state.py | 85 +++++++++++ test/acceptance.d/input-methods-test.sh | 56 ++++--- test/shell.d/input-acceptance-checks.py | 144 ++++++++++++++++++ test/shell.d/input-acceptance-test.sh | 5 + 5 files changed, 298 insertions(+), 22 deletions(-) create mode 100644 test/acceptance.d/input-method-notifications.py create mode 100644 test/acceptance.d/input-method-state.py create mode 100644 test/shell.d/input-acceptance-checks.py create mode 100755 test/shell.d/input-acceptance-test.sh diff --git a/test/acceptance.d/input-method-notifications.py b/test/acceptance.d/input-method-notifications.py new file mode 100644 index 00000000..52e5b295 --- /dev/null +++ b/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) diff --git a/test/acceptance.d/input-method-state.py b/test/acceptance.d/input-method-state.py new file mode 100644 index 00000000..47e2c014 --- /dev/null +++ b/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") diff --git a/test/acceptance.d/input-methods-test.sh b/test/acceptance.d/input-methods-test.sh index f5c414a9..5194cfdf 100644 --- a/test/acceptance.d/input-methods-test.sh +++ b/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" diff --git a/test/shell.d/input-acceptance-checks.py b/test/shell.d/input-acceptance-checks.py new file mode 100644 index 00000000..a1cc3031 --- /dev/null +++ b/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() diff --git a/test/shell.d/input-acceptance-test.sh b/test/shell.d/input-acceptance-test.sh new file mode 100755 index 00000000..ec7df578 --- /dev/null +++ b/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"