#!/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 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: self.buf += self.sock.recv(65536) (length,) = struct.unpack(">I", self.buf[:4]) while len(self.buf) < 4 + length: self.buf += self.sock.recv(65536) 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") daemon = subprocess.Popen( [TUI_WM_BIN, "-d"], cwd=REPO, 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) # ── 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) # ── 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()