Files
TUI-WM/tests/integration.py
Bjorn Blomberg 35eb33b9a1 Fix SGR mouse release button + reflow wrapped lines on resize
- SGR (1006) release events now carry the same button number as the
  press ('m' suffix marks release); the legacy 'button 3 = release'
  is only used for X10/UTF8 encodings. ncurses apps (htop etc.) could
  not pair press+release before, so clicks were ignored while scroll
  (press-only wheel events) worked
- Each window keeps its recent raw PTY output (capped at 512 KB); on
  resize the history is replayed into a fresh parser at the new width
  so long lines re-wrap correctly instead of keeping old break points
- Integration: release-button and reflow checks, 33/33 passing

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-15 20:09:45 +02:00

454 lines
21 KiB
Python

#!/usr/bin/env python3
"""Integrationstest för TUI-WM:s daemon-läge och socket-API.
Startar `tui-wm -d` med en isolerad socket, ansluter som display- och
app-klient och verifierar:
1. hello/hello_ok-handskakning och att frames strömmas
2. spawn_window + list_windows
3. musforwarding: ett klick skickat som `mouse_event` översätts till
SGR-mus-protokoll och landar i den virtuella terminalens PTY
(verifieras genom att fönstret kör `cat -v` med mus-tracking på,
så klicket ekas synligt i frame-innehållet)
4. key_event: tangenter går genom WM:ns event-hantering
5. spawn_popup + popup_result via Enter
6. TUI-FM renderar inne i ett TUI-WM-fönster (om binär finns)
Körning: python3 tests/integration.py
Miljö: TUI_WM_BIN (default target/debug/tui-wm)
TUI_FM_BIN (default ../TUI-FM/target/debug/tui-fm, hoppas över om saknas)
"""
import json
import os
import shutil
import re
import socket
import struct
import subprocess
import sys
import tempfile
import time
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TUI_WM_BIN = os.environ.get("TUI_WM_BIN", os.path.join(REPO, "target/debug/tui-wm"))
TUI_FM_BIN = os.environ.get(
"TUI_FM_BIN",
os.path.normpath(os.path.join(REPO, "../TUI-FM/target/debug/tui-fm")),
)
ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)")
passed, failed = [], []
def check(name, ok, detail=""):
(passed if ok else failed).append(name)
print(f" {'PASS' if ok else 'FAIL'}: {name}" + (f"{detail}" if detail and not ok else ""))
class Conn:
def __init__(self, path):
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.connect(path)
self.sock.settimeout(5.0)
self.buf = b""
def send(self, msg):
data = json.dumps(msg).encode()
self.sock.sendall(struct.pack(">I", len(data)) + data)
def recv(self):
while len(self.buf) < 4:
data = self.sock.recv(65536)
if not data:
raise EOFError("servern stängde anslutningen")
self.buf += data
(length,) = struct.unpack(">I", self.buf[:4])
while len(self.buf) < 4 + length:
data = self.sock.recv(65536)
if not data:
raise EOFError("servern stängde anslutningen")
self.buf += data
payload = self.buf[4 : 4 + length]
self.buf = self.buf[4 + length :]
return json.loads(payload)
def recv_until(self, msg_type, timeout=5.0, predicate=None):
"""Läs meddelanden tills ett av rätt typ (och ev. predikat) dyker upp."""
deadline = time.time() + timeout
while time.time() < deadline:
msg = self.recv()
if msg.get("type") == msg_type and (predicate is None or predicate(msg)):
return msg
raise TimeoutError(f"fick aldrig {msg_type}")
def frame_text(self, timeout=5.0, contains=None):
"""Läs frames tills texten (ANSI-strippad) innehåller `contains`.
Returnerar den strippade texten från senaste framen."""
deadline = time.time() + timeout
last = ""
while time.time() < deadline:
msg = self.recv_until("frame", timeout=max(0.1, deadline - time.time()))
raw = bytes.fromhex(msg["data"]).decode("utf-8", "replace")
last = ANSI_RE.sub("", raw)
if contains is None or contains in last:
return last
return last
def key_event(ch=None, code=None, modifiers=""):
c = {"Char": ch} if ch else code
return {
"type": "key_event",
"event": {"code": c, "modifiers": modifiers, "kind": "Press", "state": ""},
}
def mouse_event(kind, col, row):
return {
"type": "mouse_event",
"event": {"kind": kind, "column": col, "row": row, "modifiers": ""},
}
def main():
tmp = tempfile.mkdtemp(prefix="tui-wm-test-")
env = dict(os.environ, XDG_RUNTIME_DIR=tmp)
sock_path = os.path.join(tmp, "tui-wm.sock")
# Egen konfig-katalog så testerna (t.ex. hot-reload) kan skriva i
# konfigfiler utan att röra repot.
cfgdir = os.path.join(tmp, "cfg")
os.makedirs(cfgdir)
shutil.copy(os.path.join(REPO, "config.toml"), cfgdir)
shutil.copytree(os.path.join(REPO, "panels"), os.path.join(cfgdir, "panels"))
shutil.copytree(os.path.join(REPO, "themes"), os.path.join(cfgdir, "themes"))
# bakgrundsbilden finns inte i test-miljön — ta bort raden
cfg = open(os.path.join(cfgdir, "config.toml")).read()
cfg = "\n".join(l for l in cfg.splitlines() if not l.startswith("background_image"))
open(os.path.join(cfgdir, "config.toml"), "w").write(cfg)
daemon = subprocess.Popen(
[TUI_WM_BIN, "-d"], cwd=cfgdir, env=env,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
try:
# Vänta på att socketen binds
for _ in range(50):
if os.path.exists(sock_path):
break
time.sleep(0.1)
check("daemon binder socket", os.path.exists(sock_path))
# ── 1. Display-klient: handskakning + frames ────────────────────
disp = Conn(sock_path)
disp.send({"type": "hello", "role": "display", "width": 100, "height": 30})
hello = disp.recv()
check("hello_ok till display", hello.get("type") == "hello_ok")
app = Conn(sock_path)
app.send({"type": "hello", "role": "app", "width": 0, "height": 0})
check("hello_ok till app", app.recv().get("type") == "hello_ok")
frame = disp.recv_until("frame")
check("frames strömmas till display", frame.get("width", 0) > 0)
# ── 1b. Exit från en klient kopplar bara ner klienten ───────────
disp2 = Conn(sock_path)
disp2.send({"type": "hello", "role": "display", "width": 80, "height": 24})
disp2.recv_until("hello_ok")
disp2.recv_until("frame")
disp2.send(key_event(ch="q", modifiers="CONTROL"))
try:
got_bye = False
deadline = time.time() + 3
while time.time() < deadline:
if disp2.recv().get("type") == "bye":
got_bye = True
break
except (EOFError, TimeoutError, OSError):
pass # EOF duger också som nedkoppling
check("ctrl+q från klient ger bye till klienten", got_bye)
frame = disp.recv_until("frame")
check("daemonen överlever klientens exit", frame.get("width", 0) > 0)
# ── 2. spawn_window med mus-tracking-testapp ────────────────────
# printf slår på AnyMotion + SGR-mus-tracking, cat -v ekar allt
# PTY:n tar emot — inklusive forwardade mus-sekvenser.
app.send({
"type": "spawn_window",
"command": "printf '\\033[?1003h\\033[?1006h'; exec cat -v",
"request_id": "w1",
})
opened = app.recv_until("window_opened")
win_id = opened["id"]
check("spawn_window svarar window_opened", isinstance(win_id, int))
app.send({"type": "list_windows", "request_id": "l1"})
wl = app.recv_until("window_list")
win = next((w for w in wl["windows"] if w["id"] == win_id), None)
check("fönstret finns i list_windows", win is not None)
# ── 3. Musforwarding in i den virtuella terminalen ──────────────
time.sleep(0.5) # låt printf hinna sätta mus-läget
cx, cy = win["x"] + 4, win["y"] + 3 # inne i innehållsytan (border=1)
disp.send(mouse_event({"Down": "Left"}, cx, cy))
disp.send(mouse_event({"Up": "Left"}, cx, cy))
text = disp.frame_text(timeout=5.0, contains="[<0;")
check("musklick forwardas som SGR till PTY", "[<0;" in text,
f"frame-text: {text[-200:]!r}")
# Släppet ska bära SAMMA knappnummer med 'm'-suffix (SGR), inte
# legacy-koden 3 — annars registrerar ncurses-appar inga klick.
text = disp.frame_text(timeout=5.0, contains="[<0;4;3m")
check("mus-släpp forwardas med korrekt knappnummer", "[<0;4;3m" in text,
f"frame-text: {text[-200:]!r}")
# Scroll ska också forwardas (SGR-knapp 64/65)
disp.send(mouse_event("ScrollDown", cx, cy))
text = disp.frame_text(timeout=5.0, contains="[<65;")
check("scroll forwardas som SGR till PTY", "[<65;" in text,
f"frame-text: {text[-200:]!r}")
# ── 4. key_event genom WM:t (skriv i terminalen) ────────────────
disp.send(key_event(ch="x"))
text = disp.frame_text(timeout=5.0, contains="x")
check("key_event når den virtuella terminalen", "x" in text)
# ── 4b. Breda tecken (emoji) får inte förskjuta raden ───────────
# vt100 lämnar en tom fortsättningscell efter breda tecken; om
# renderaren gör den till mellanslag blir "📁X" → "📁 X" och allt
# efter emojin skjuts åt höger.
# OBS: fönstertiteln visar kommandosträngen (innehåller "X-MARKER"
# som text) — vänta därför på emojin+markören som bara finns i
# fönstrets INNEHÅLL.
app.send({"type": "spawn_window",
"command": "printf '\\360\\237\\223\\201X-MARKER\\n'; exec cat",
"request_id": "w-emoji"})
emoji_open = app.recv_until("window_opened")
text = disp.frame_text(timeout=5.0, contains="📁X-MARKER")
check("emoji förskjuter inte raden", "📁X-MARKER" in text,
f"frame-text: {text[-200:]!r}")
app.send({"type": "close_window", "window_id": emoji_open["id"]})
# ── 4b2. Resize får inte radera fönstrets innehåll ───────────────
app.send({"type": "spawn_window",
"command": "printf 'RESIZE-KEEP\\n'; exec cat",
"request_id": "w-resize"})
rz_open = app.recv_until("window_opened")
disp.frame_text(timeout=5.0, contains="RESIZE-KEEP")
app.send({"type": "list_windows", "request_id": "lr"})
wlr = app.recv_until("window_list")
rw = next(w for w in wlr["windows"] if w["id"] == rz_open["id"])
# dra i nedre högra hörnet: krymp fönstret 4 kolumner/2 rader
corner_x = rw["x"] + rw["width"] - 1
corner_y = rw["y"] + rw["height"] - 1
disp.send(mouse_event({"Down": "Left"}, corner_x, corner_y))
disp.send(mouse_event({"Drag": "Left"}, corner_x - 4, corner_y - 2))
disp.send(mouse_event({"Up": "Left"}, corner_x - 4, corner_y - 2))
time.sleep(0.5)
text = disp.frame_text(timeout=5.0, contains="RESIZE-KEEP")
check("innehållet överlever fönster-resize", "RESIZE-KEEP" in text,
f"frame-tail: {text[-200:]!r}")
app.send({"type": "close_window", "window_id": rz_open["id"]})
# ── 4b3. Reflow: långa rader radbryts om efter resize ────────────
long = "A" * 90
app.send({"type": "spawn_window",
"command": f"printf '{long}\\n'; exec cat",
"request_id": "w-reflow"})
rf_open = app.recv_until("window_opened")
disp.frame_text(timeout=5.0, contains="AAAA")
app.send({"type": "list_windows", "request_id": "lrf"})
wrf = next(w for w in app.recv_until("window_list")["windows"]
if w["id"] == rf_open["id"])
# 90 A på 80 kolumner bred terminal → raden är bruten
text = disp.frame_text(timeout=2.0)
check("lång rad är bruten före resize", long not in text)
# dra nedre högra hörnet så innehållet blir ≥ 90 kolumner brett
cx0, cy0 = wrf["x"] + wrf["width"] - 1, wrf["y"] + wrf["height"] - 1
grow = 94 - (wrf["width"] - 2)
disp.send(mouse_event({"Down": "Left"}, cx0, cy0))
disp.send(mouse_event({"Drag": "Left"}, min(cx0 + grow, 99), cy0))
disp.send(mouse_event({"Up": "Left"}, min(cx0 + grow, 99), cy0))
text = disp.frame_text(timeout=5.0, contains=long)
check("raden radbryts om (reflow) efter resize", long in text,
f"bredd efter: {wrf['width']}+{grow}")
app.send({"type": "close_window", "window_id": rf_open["id"]})
# ── 4c. Scrollback: mushjul utan mus-tracking scrollar historiken ─
app.send({"type": "spawn_window",
"command": "seq 1 200; exec cat",
"request_id": "w-scroll"})
scroll_open = app.recv_until("window_opened")
text = disp.frame_text(timeout=5.0, contains="200")
app.send({"type": "list_windows", "request_id": "ls"})
wls = app.recv_until("window_list")
sw = next(w for w in wls["windows"] if w["id"] == scroll_open["id"])
scx, scy = sw["x"] + 10, sw["y"] + 5
visible_before = "163" in text
for _ in range(8):
disp.send(mouse_event("ScrollUp", scx, scy))
text = disp.frame_text(timeout=5.0, contains="163")
check("mushjul scrollar terminalens historik",
"163" in text and not visible_before,
f"före: {visible_before}, frame-tail: {text[-150:]!r}")
# Tangenttryck hoppar tillbaka till botten av historiken
disp.send(key_event(ch="x"))
text = disp.frame_text(timeout=5.0, contains="200")
check("tangenttryck hoppar till historikens botten", "200" in text)
app.send({"type": "close_window", "window_id": scroll_open["id"]})
# ── 4d. Kommandodialogen kör kommandot direkt ────────────────────
disp.send(key_event(ch=" ", modifiers="CONTROL")) # ctrl+space
time.sleep(0.3)
for ch in "top":
disp.send(key_event(ch=ch))
disp.send(key_event(code="Enter"))
text = disp.frame_text(timeout=8.0, contains="PID")
check("kommandodialogen kör programmet (top)", "PID" in text,
f"frame-tail: {text[-200:]!r}")
app.send({"type": "list_windows", "request_id": "lt"})
wlt = app.recv_until("window_list")
top_win = next((w for w in wlt["windows"] if w["title"].strip() == "top"), None)
check("fönstret har programmets namn som titel", top_win is not None,
f"fönster: {[w['title'] for w in wlt['windows']]}")
if top_win:
app.send({"type": "close_window", "window_id": top_win["id"]})
# ── 4e. Alt+Tab-växlare + snap/maximera ──────────────────────────
# Två fönster finns (mouse-test + top stängdes; spawna ett till)
app.send({"type": "spawn_window", "command": "exec cat",
"request_id": "w-sw"})
sw_open = app.recv_until("window_opened")
disp.send(key_event(code="Tab", modifiers="ALT"))
text = disp.frame_text(timeout=5.0, contains="Fönster")
check("alt+tab öppnar fönsterväxlaren", "Fönster" in text)
disp.send(key_event(code="Esc"))
time.sleep(0.2)
# Snap vänster: fokuserat fönster ska bli halva bredden
disp.send(key_event(code="Left", modifiers="ALT"))
time.sleep(0.3)
app.send({"type": "list_windows", "request_id": "lsnap"})
wsn = app.recv_until("window_list")
snapped = next(w for w in wsn["windows"] if w["id"] == sw_open["id"])
# innehållsytan är 100 bred (minus panel högst upp) → halvan ~50
check("alt+left fäster fönstret som halvskärm",
45 <= snapped["width"] <= 55 and snapped["x"] <= 1,
f"geometri: {snapped}")
# Maximera/återställ: alt+up två gånger ska ge tillbaka snappad storlek
disp.send(key_event(code="Up", modifiers="ALT"))
time.sleep(0.3)
app.send({"type": "list_windows", "request_id": "lmax"})
wmx = app.recv_until("window_list")
maxed = next(w for w in wmx["windows"] if w["id"] == sw_open["id"])
check("alt+up maximerar fönstret", maxed["width"] >= 95,
f"geometri: {maxed}")
app.send({"type": "close_window", "window_id": sw_open["id"]})
# ── 4f. Copy-mode: alt+c, sök, avsluta ───────────────────────────
app.send({"type": "spawn_window", "command": "seq 1 200; exec cat",
"request_id": "w-copy"})
cm_open = app.recv_until("window_opened")
disp.frame_text(timeout=5.0, contains="200")
disp.send(key_event(ch="c", modifiers="ALT"))
text = disp.frame_text(timeout=5.0, contains="KOPIERA")
check("alt+c öppnar copy-mode", "KOPIERA" in text)
# sök efter rad 42
disp.send(key_event(ch="/"))
for ch in "42\n":
if ch == "\n":
disp.send(key_event(code="Enter"))
else:
disp.send(key_event(ch=ch))
text = disp.frame_text(timeout=5.0, contains="42")
check("copy-mode-sökning hittar träffen", "42" in text)
disp.send(key_event(ch="q"))
text = disp.frame_text(timeout=5.0, contains="200")
check("q lämnar copy-mode och hoppar till botten", "200" in text)
app.send({"type": "close_window", "window_id": cm_open["id"]})
# ── 4g. Hot-reload: ändra panelfilen → panelen uppdateras ────────
with open(os.path.join(cfgdir, "panels", "topbar.toml"), "a") as f:
f.write('\n[[item]]\nlabel = "HOTRELOAD"\naction = { type = "no_op" }\n')
text = disp.frame_text(timeout=6.0, contains="HOTRELOAD")
check("hot-reload plockar upp panel-ändring", "HOTRELOAD" in text,
f"frame-tail: {text[:150]!r}")
# ── 4h. Inbyggda inställningar via kugghjulet i panelen ──────────
text = disp.frame_text(timeout=5.0, contains="Inställningar")
row0 = text[:100]
gx = row0.find("Inställningar")
check("kugghjulet syns i panelen", gx > 0)
disp.send(mouse_event({"Down": "Left"}, gx, 0))
disp.send(mouse_event({"Up": "Left"}, gx, 0))
text = disp.frame_text(timeout=5.0, contains="Tema:")
check("inställningsdialogen öppnas från panelen", "Tema:" in text
and "Skugga" in text and "config.toml" in text)
# Ned två rader till Skugga och toggla av
disp.send(key_event(code="Down"))
disp.send(key_event(code="Down"))
disp.send(key_event(code="Enter"))
time.sleep(0.5)
cfg_now = open(os.path.join(cfgdir, "config.toml")).read()
check("skugg-toggle sparas till config.toml", "shadow = false" in cfg_now,
f"config innehåller: {[l for l in cfg_now.splitlines() if 'shadow' in l]!r}")
disp.send(key_event(code="Esc"))
time.sleep(0.3)
# ── 5. Popup + popup_result via Enter ───────────────────────────
app.send({
"type": "spawn_popup", "message": "Integrationstest?",
"buttons": ["Ja", "Nej"], "request_id": "p1",
})
disp.frame_text(timeout=5.0, contains="Integrationstest?")
disp.send(key_event(code="Enter"))
pr = app.recv_until("popup_result")
check("popup_result efter Enter", pr.get("button") == "Ja",
f"fick: {pr}")
# ── 6. TUI-FM inne i TUI-WM ─────────────────────────────────────
if os.path.exists(TUI_FM_BIN):
app.send({"type": "spawn_window", "command": TUI_FM_BIN,
"request_id": "w2"})
fm_open = app.recv_until("window_opened")
text = disp.frame_text(timeout=10.0, contains="Namn")
check("TUI-FM renderar i TUI-WM-fönster", "Namn" in text,
f"frame-text: {text[-300:]!r}")
# Klick + scroll in i TUI-FM får inte krascha något
app.send({"type": "list_windows", "request_id": "l2"})
wl2 = app.recv_until("window_list")
fmwin = next((w for w in wl2["windows"] if w["id"] == fm_open["id"]), None)
if fmwin:
fx, fy = fmwin["x"] + 30, fmwin["y"] + 5
disp.send(mouse_event({"Down": "Left"}, fx, fy))
disp.send(mouse_event({"Up": "Left"}, fx, fy))
disp.send(mouse_event("ScrollDown", fx, fy))
time.sleep(0.5)
frame = disp.recv_until("frame")
check("frames fortsätter efter mus i TUI-FM", frame.get("width", 0) > 0)
else:
print(f" SKIP: TUI-FM-binär saknas ({TUI_FM_BIN})")
# ── Städa: stäng fönster ────────────────────────────────────────
app.send({"type": "close_window", "window_id": win_id})
finally:
daemon.terminate()
try:
daemon.wait(timeout=3)
except subprocess.TimeoutExpired:
daemon.kill()
print(f"\n{len(passed)} PASS, {len(failed)} FAIL")
if failed:
print("Misslyckade: " + ", ".join(failed))
sys.exit(1)
if __name__ == "__main__":
main()