niri-scratchpad: fix close-while-stashed eviction + add live acceptance test
Refresh the patch with the destroy-path fix (a client dying while stashed is now evicted instead of leaked/zombied), and add pkg/niri-scratchpad/tests/ acceptance.py: a headless IPC-driven regression test that drives a nested patched niri and asserts on `niri msg -j windows`. It found the above bug (unit tests miss it — the harness calls remove_window directly). Run it after rebasing the patch onto a new niri release. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
funman300
@@ -0,0 +1,310 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Live acceptance / regression test for the niri native scratchpad patch.
|
||||
|
||||
Drives a real (patched) niri through its own IPC — the exact `do_action`
|
||||
path a keybind hits — and asserts on `niri msg -j windows`. No keyboard
|
||||
injection, so it can't type into the host session's windows. Each test gets
|
||||
its own fresh nested niri, so cases are hermetic and order-independent.
|
||||
|
||||
Run this after rebasing the patch onto a new niri release to confirm the
|
||||
scratchpad still behaves. It needs:
|
||||
- a running Wayland (or X) session to host the nested winit window,
|
||||
- the patched niri binary (built from niri-patches/ via the PKGBUILD, or a
|
||||
dev checkout),
|
||||
- `alacritty` as the throwaway test client.
|
||||
|
||||
Usage:
|
||||
python3 acceptance.py [PATH_TO_NIRI]
|
||||
Binary resolution order: argv[1], $NIRI_BIN, `niri` on PATH,
|
||||
~/build/niri-scratchpad/target/release/niri. The binary MUST be patched
|
||||
(its `niri msg action --help` must list `scratchpad-show`).
|
||||
|
||||
Exit code 0 iff every non-skipped test passes.
|
||||
"""
|
||||
import glob, json, os, shutil, signal, subprocess, sys, tempfile, time
|
||||
|
||||
MINIMAL_CONFIG = """\
|
||||
// Hermetic config for scratchpad acceptance: no startup spawns.
|
||||
input { keyboard { xkb { } } }
|
||||
binds {
|
||||
Mod+M { move-window-to-scratchpad; }
|
||||
Mod+Shift+M { scratchpad-show; }
|
||||
Mod+Q { close-window; }
|
||||
}
|
||||
"""
|
||||
|
||||
|
||||
def resolve_niri():
|
||||
for cand in (sys.argv[1] if len(sys.argv) > 1 else None,
|
||||
os.environ.get("NIRI_BIN"),
|
||||
shutil.which("niri"),
|
||||
os.path.expanduser("~/build/niri-scratchpad/target/release/niri")):
|
||||
if cand and os.path.exists(cand):
|
||||
return cand
|
||||
sys.exit("ERROR: no niri binary found (pass a path, set $NIRI_BIN, or "
|
||||
"install niri-scratchpad).")
|
||||
|
||||
|
||||
NIRI = resolve_niri()
|
||||
RT = os.environ["XDG_RUNTIME_DIR"]
|
||||
|
||||
|
||||
class Niri:
|
||||
"""A fresh nested niri instance driven over IPC."""
|
||||
|
||||
def __init__(self, config_path):
|
||||
before = set(glob.glob(f"{RT}/niri.*.sock"))
|
||||
self.proc = subprocess.Popen(
|
||||
[NIRI, "-c", config_path],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
self.sock = None
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 15:
|
||||
new = set(glob.glob(f"{RT}/niri.*.sock")) - before
|
||||
if new:
|
||||
self.sock = new.pop()
|
||||
break
|
||||
time.sleep(0.2)
|
||||
if not self.sock:
|
||||
self.close()
|
||||
raise RuntimeError("nested niri socket never appeared")
|
||||
self.env = dict(os.environ, NIRI_SOCKET=self.sock)
|
||||
for _ in range(40): # wait for IPC to answer
|
||||
if self.alive():
|
||||
break
|
||||
time.sleep(0.2)
|
||||
|
||||
def msg(self, *args):
|
||||
r = subprocess.run([NIRI, "msg", "-j", *args],
|
||||
capture_output=True, text=True, env=self.env)
|
||||
return json.loads(r.stdout) if r.stdout.strip() else None
|
||||
|
||||
def action(self, *args):
|
||||
return subprocess.run([NIRI, "msg", "action", *args],
|
||||
capture_output=True, text=True, env=self.env)
|
||||
|
||||
def alive(self):
|
||||
return self.msg("version") is not None
|
||||
|
||||
def windows(self):
|
||||
return self.msg("windows") or []
|
||||
|
||||
def wait_count(self, n, timeout=8):
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < timeout:
|
||||
if len(self.windows()) == n:
|
||||
return self.windows()
|
||||
time.sleep(0.15)
|
||||
return self.windows()
|
||||
|
||||
def spawn_window(self):
|
||||
"""Spawn an alacritty window and return its id once it appears."""
|
||||
before = {w["id"] for w in self.windows()}
|
||||
self.action("spawn", "--", "alacritty")
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < 10:
|
||||
new = [w for w in self.windows() if w["id"] not in before]
|
||||
if new:
|
||||
return new[0]["id"]
|
||||
time.sleep(0.15)
|
||||
raise RuntimeError("spawned window never appeared")
|
||||
|
||||
def by_id(self, i):
|
||||
return next((w for w in self.windows() if w["id"] == i), None)
|
||||
|
||||
def focused(self):
|
||||
return next((w for w in self.windows() if w["is_focused"]), None)
|
||||
|
||||
def floating_pos(self, i, tries=12):
|
||||
# tile_pos_in_workspace_view is transiently null right after a show.
|
||||
for _ in range(tries):
|
||||
w = self.by_id(i)
|
||||
p = w["layout"].get("tile_pos_in_workspace_view") if w else None
|
||||
if p is not None:
|
||||
return p
|
||||
time.sleep(0.08)
|
||||
return None
|
||||
|
||||
def output_size(self):
|
||||
outs = self.msg("outputs") or {}
|
||||
for o in outs.values():
|
||||
lg = o.get("logical") or {}
|
||||
if lg.get("width") and lg.get("height"):
|
||||
return lg["width"], lg["height"]
|
||||
return None
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
self.proc.send_signal(signal.SIGTERM)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
self.proc.wait(timeout=3)
|
||||
except Exception:
|
||||
try:
|
||||
self.proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ---- tests: each takes a fresh Niri and returns (ok, detail) ----
|
||||
|
||||
def t_stash_hides_and_moves_focus(n):
|
||||
a = n.spawn_window()
|
||||
b = n.spawn_window()
|
||||
n.action("focus-window", "--id", str(b))
|
||||
time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad")
|
||||
n.wait_count(1)
|
||||
f = n.focused()
|
||||
ok = (len(n.windows()) == 1 and n.by_id(b) is None
|
||||
and f is not None and f["id"] == a)
|
||||
return ok, f"remaining={len(n.windows())}, b_gone={n.by_id(b) is None}, focus={f['id'] if f else None}/want {a}"
|
||||
|
||||
|
||||
def t_show_floating_focused_centered(n):
|
||||
b = n.spawn_window()
|
||||
n.action("focus-window", "--id", str(b)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad"); n.wait_count(0)
|
||||
n.action("scratchpad-show"); n.wait_count(1)
|
||||
w = n.by_id(b)
|
||||
pos = n.floating_pos(b)
|
||||
centered = "n/a"
|
||||
ok = bool(w and w["is_floating"] and w["is_focused"] and pos and pos[0] > 5)
|
||||
osz = n.output_size()
|
||||
if ok and osz and w:
|
||||
ww, wh = w["layout"]["window_size"]
|
||||
cx, cy = (osz[0] - ww) / 2, (osz[1] - wh) / 2
|
||||
centered = abs(pos[0] - cx) < osz[0] * 0.15 and abs(pos[1] - cy) < osz[1] * 0.15
|
||||
ok = ok and bool(centered)
|
||||
return ok, f"floating={w['is_floating'] if w else None}, focused={w['is_focused'] if w else None}, pos={pos}, centered={centered}"
|
||||
|
||||
|
||||
def t_reshow_hides(n):
|
||||
b = n.spawn_window()
|
||||
n.action("focus-window", "--id", str(b)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad"); n.wait_count(0)
|
||||
n.action("scratchpad-show"); n.wait_count(1) # show (focused)
|
||||
n.action("scratchpad-show"); n.wait_count(0) # hide
|
||||
ok = len(n.windows()) == 0
|
||||
return ok, f"count after re-show={len(n.windows())} (want 0)"
|
||||
|
||||
|
||||
def t_cycle_round_robin(n):
|
||||
a = n.spawn_window()
|
||||
b = n.spawn_window()
|
||||
for wid in (b, a):
|
||||
n.action("focus-window", "--id", str(wid)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad")
|
||||
n.wait_count(0)
|
||||
n.action("scratchpad-show"); w1 = n.wait_count(1); first = w1[0]["id"] if w1 else None
|
||||
n.action("scratchpad-show"); n.wait_count(0) # hide (focused)
|
||||
n.action("scratchpad-show"); w2 = n.wait_count(1); second = w2[0]["id"] if w2 else None
|
||||
ok = first is not None and second is not None and first != second
|
||||
return ok, f"first={first}, second={second} (want different)"
|
||||
|
||||
|
||||
def t_geometry_remembered(n):
|
||||
w = n.spawn_window()
|
||||
n.action("focus-window", "--id", str(w)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad"); n.wait_count(0)
|
||||
n.action("scratchpad-show"); n.wait_count(1)
|
||||
n.action("focus-window", "--id", str(w)); time.sleep(0.2)
|
||||
n.action("move-floating-window", "-x", "+130", "-y", "+90"); time.sleep(0.4)
|
||||
pos1 = n.floating_pos(w)
|
||||
n.action("scratchpad-show"); n.wait_count(0) # hide
|
||||
n.action("scratchpad-show"); n.wait_count(1) # show again (single window)
|
||||
pos2 = n.floating_pos(w)
|
||||
back = n.by_id(w)
|
||||
ok = (back is not None and back["is_floating"] and pos1 and pos2
|
||||
and abs(pos1[0] - pos2[0]) < 2 and abs(pos1[1] - pos2[1]) < 2)
|
||||
return ok, f"pos before hide={pos1}, after show={pos2}, floating={back['is_floating'] if back else None}"
|
||||
|
||||
|
||||
def t_shown_not_focused_gets_focus(n):
|
||||
p = n.spawn_window() # persistent tiled window
|
||||
s = n.spawn_window() # to be stashed + shown
|
||||
n.action("focus-window", "--id", str(s)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad"); n.wait_count(1)
|
||||
n.action("scratchpad-show"); n.wait_count(2) # s shown floating, focused
|
||||
n.action("focus-window", "--id", str(p)); time.sleep(0.3)
|
||||
npre = len(n.windows())
|
||||
n.action("scratchpad-show"); time.sleep(0.4) # should just focus s back
|
||||
f = n.focused()
|
||||
ok = len(n.windows()) == npre and f is not None and f["id"] == s
|
||||
return ok, f"count {npre}->{len(n.windows())} (want same), focus={f['id'] if f else None}/want {s}"
|
||||
|
||||
|
||||
def t_close_while_stashed(n):
|
||||
w = n.spawn_window()
|
||||
pid = n.by_id(w)["pid"]
|
||||
n.action("focus-window", "--id", str(w)); time.sleep(0.2)
|
||||
n.action("move-window-to-scratchpad"); n.wait_count(0)
|
||||
os.kill(pid, signal.SIGKILL) # client dies WHILE stashed
|
||||
time.sleep(0.6)
|
||||
r = n.action("scratchpad-show"); time.sleep(0.5)
|
||||
ok = n.alive() and n.by_id(w) is None and r.returncode == 0
|
||||
return ok, f"alive={n.alive()}, dead_shown={n.by_id(w) is not None}, rc={r.returncode}"
|
||||
|
||||
|
||||
def t_empty_stash_noop(n):
|
||||
n.spawn_window()
|
||||
r = n.action("scratchpad-show"); time.sleep(0.3)
|
||||
ok = len(n.windows()) == 1 and n.alive() and r.returncode == 0
|
||||
return ok, f"count stayed {len(n.windows())}, alive={n.alive()}, rc={r.returncode}"
|
||||
|
||||
|
||||
TESTS = [
|
||||
("1 stash hides + focus moves", t_stash_hides_and_moves_focus),
|
||||
("2 show floating+focused+centered", t_show_floating_focused_centered),
|
||||
("3 re-show hides", t_reshow_hides),
|
||||
("4 cycle round-robin", t_cycle_round_robin),
|
||||
("5 geometry remembered", t_geometry_remembered),
|
||||
("6 shown-not-focused -> focus it", t_shown_not_focused_gets_focus),
|
||||
("7 close-while-stashed no crash", t_close_while_stashed),
|
||||
("10 empty-stash no-op", t_empty_stash_noop),
|
||||
]
|
||||
|
||||
|
||||
def main():
|
||||
if not shutil.which("alacritty"):
|
||||
sys.exit("ERROR: alacritty is required as the test client.")
|
||||
# confirm the binary is patched
|
||||
h = subprocess.run([NIRI, "msg", "action", "--help"],
|
||||
capture_output=True, text=True)
|
||||
if "scratchpad-show" not in h.stdout:
|
||||
sys.exit(f"ERROR: {NIRI} is not patched (no scratchpad-show action).")
|
||||
|
||||
print(f"niri: {NIRI}\n")
|
||||
cfg = tempfile.NamedTemporaryFile("w", suffix=".kdl", delete=False)
|
||||
cfg.write(MINIMAL_CONFIG); cfg.close()
|
||||
results = []
|
||||
try:
|
||||
for name, fn in TESTS:
|
||||
n = None
|
||||
try:
|
||||
n = Niri(cfg.name)
|
||||
ok, detail = fn(n)
|
||||
except Exception as e:
|
||||
ok, detail = False, f"exception: {e}"
|
||||
finally:
|
||||
if n:
|
||||
n.close()
|
||||
time.sleep(0.3)
|
||||
results.append((name, ok, detail))
|
||||
print(f" [{'PASS' if ok else 'FAIL'}] {name}: {detail}")
|
||||
finally:
|
||||
os.unlink(cfg.name)
|
||||
|
||||
# #8 multi-monitor cannot be exercised nested (single winit output).
|
||||
print(" [SKIP] 8 multi-monitor: needs a physical 2nd output; "
|
||||
"code uses active_output (correct by construction).")
|
||||
|
||||
passed = sum(1 for _, ok, _ in results if ok)
|
||||
failed = sum(1 for _, ok, _ in results if not ok)
|
||||
print(f"\n{passed} passed, {failed} failed, 1 skipped")
|
||||
sys.exit(1 if failed else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user