#!/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}") # 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"]}) # ── 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}") # ── 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()