TUI-WM/tests/integration.py

#!/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:
            deadline = time.time() + 3
            while time.time() < deadline:
                disp2.recv()
            disconnected = False
        except (EOFError, TimeoutError, OSError):
            disconnected = True
        check("ctrl+q från klient kopplar ner klienten", disconnected)
        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()