Bjorn Blomberg
373294e2b9
Dropping the client's channel sender never closed the socket (the server's reader thread keeps the connection open), so the client froze waiting for frames. The server now sends a bye message; the client exits cleanly on it and its socket close lets the server clean up. Messages from already-disconnected clients are ignored. Verified end-to-end with the real client in a PTY (exits 0, daemon survives). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
309 lines
13 KiB
Python
309 lines
13 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 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")
|
|
|
|
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)
|
|
|
|
# ── 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"]})
|
|
|
|
# ── 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"]})
|
|
|
|
# ── 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()
|