path: root/tests/smoke.py

#!/usr/bin/env python3
"""End-to-end smoke test for gdb-driver (Milestone 5).

Spawns a real gdb+GEF pane on a unix socket via the pane manager, drives the
whole method surface against a known fixture, and asserts on the *structured*
results.  Doubles as a protocol regression lock (blueprint §12.6).

Run:  python3 tests/smoke.py     (must be inside a tmux session)
Exit: 0 = all checks passed, 1 = one or more failed.
"""

import json
import os
import socket
import subprocess
import sys
import tempfile
import threading

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
sys.path.insert(0, ROOT)
import pane  # noqa: E402

FIXTURE_SRC = r"""
#include <stdio.h>
int add(int a, int b) {
    int c = a + b;
    return c;
}
int main(void) {
    int x = add(2, 3);
    printf("%d\n", x);
    return 0;
}
"""

# --------------------------------------------------------------------------
# tiny assert framework
# --------------------------------------------------------------------------

_FAILS = []


def check(name, cond, detail=""):
    ok = bool(cond)
    tag = "PASS" if ok else "FAIL"
    line = f"[{tag}] {name}"
    if not ok and detail:
        line += f"  -- {detail}"
    print(line, flush=True)
    if not ok:
        _FAILS.append(name)
    return ok


# --------------------------------------------------------------------------
# rpc client
# --------------------------------------------------------------------------

class RpcError(Exception):
    pass


def rpc(sock, method, params=None, timeout=130):
    s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    s.settimeout(timeout)
    s.connect(sock)
    try:
        f = s.makefile("rwb")
        f.write((json.dumps({"id": 1, "method": method,
                             "params": params or {}}) + "\n").encode())
        f.flush()
        line = f.readline()
    finally:
        s.close()
    if not line:
        raise RpcError("empty response")
    msg = json.loads(line)
    if msg.get("error"):
        raise RpcError(msg["error"].get("message", "unknown"))
    return msg.get("result")


# --------------------------------------------------------------------------
# the drive
# --------------------------------------------------------------------------

def run_checks(sock, fixture):
    base = os.path.basename(fixture)

    # -- lifecycle ------------------------------------------------------
    p = rpc(sock, "ping")
    check("ping ready", p.get("ready") is True and p.get("proto") == 1, p)

    names = {m["name"] for m in rpc(sock, "methods")}
    for want in ("ping", "state", "regs", "bt", "mem", "screen",
                 "breakpoints"):
        check(f"methods advertises {want}", want in names, names)

    # -- pre-run state / not-ready errors -------------------------------
    st = rpc(sock, "state")
    check("state=no-inferior before run", st.get("status") == "no-inferior", st)

    try:
        rpc(sock, "regs")
        check("regs errors before run", False, "no error raised")
    except RpcError as e:
        check("regs errors before run", "not ready" in str(e), str(e))

    # -- breakpoints ----------------------------------------------------
    r = rpc(sock, "break", {"loc": "add"})
    check("break add output", "Breakpoint" in (r.get("output") or ""), r)

    bps = rpc(sock, "breakpoints")
    check("one breakpoint on add", len(bps) == 1 and bps[0]["loc"].endswith("add"),
          bps)

    # -- run -> settle at breakpoint (the crux) -------------------------
    r = rpc(sock, "run")
    st = r["state"]
    check("run settles STOPPED", st.get("status") == "stopped", st)
    check("run stopped in add", st.get("function") == "add", st)
    check("run stopped in fixture file",
          (st.get("file") or "").endswith(base.replace(base, "smoke_fixture.c")) or
          (st.get("file") or "").endswith("smoke_fixture.c"), st)

    # -- eval args ------------------------------------------------------
    a = rpc(sock, "print", {"expr": "a"})
    b = rpc(sock, "print", {"expr": "b"})
    check("print a == 2", a["output"].endswith("0x2") or a["output"].endswith("2"),
          a)
    check("print b == 3", b["output"].endswith("0x3") or b["output"].endswith("3"),
          b)

    # -- reads ----------------------------------------------------------
    regs = rpc(sock, "regs")
    check("regs has rip", "rip" in regs, list(regs)[:5])
    bt = rpc(sock, "bt")
    check("bt: #0 add #1 main",
          len(bt) >= 2 and bt[0]["func"] == "add" and bt[1]["func"] == "main",
          bt)
    m = rpc(sock, "mem", {"addr": "$sp", "len": 16})
    check("mem returns 16 bytes hex", len(m.get("hex", "")) == 32, m)

    # -- step: line must advance, then c becomes defined ----------------
    line_before = rpc(sock, "state")["line"]
    r = rpc(sock, "next")
    line_after = r["state"]["line"]
    check("next advances source line", line_after == line_before + 1,
          f"{line_before}->{line_after}")
    c = rpc(sock, "print", {"expr": "c"})
    check("print c == 5 after next",
          c["output"].endswith("0x5") or c["output"].endswith("5"), c)

    # -- finish -> back in main ----------------------------------------
    r = rpc(sock, "finish")
    check("finish returns to main", r["state"].get("function") == "main",
          r["state"])

    # -- screen (viewer surface) ---------------------------------------
    scr = rpc(sock, "screen")
    text = scr.get("text", "")
    check("screen returns text", len(text) > 0)
    # capture-pane returns only the visible region; GEF's context fills it, so
    # match on any GEF context marker a viewer sees (not necessarily the prompt)
    markers = ("gef", "registers", "$rip", "$rsp", "\u2500", "code:")
    check("screen shows GEF context",
          any(m in text.lower() if m == "gef" else m in text for m in markers),
          text[:80])

    # -- continue -> program exits -------------------------------------
    r = rpc(sock, "cont")
    check("cont settles no-inferior (exit)",
          r["state"].get("status") == "no-inferior", r["state"])

    # -- concurrency: parallel pings never corrupt framing --------------
    results = []

    def hammer():
        try:
            results.append(("ok", rpc(sock, "ping", timeout=5)))
        except RpcError as e:
            results.append(("busy", str(e)))
        except Exception as e:  # noqa: BLE001
            results.append(("crash", str(e)))
    ts = [threading.Thread(target=hammer) for _ in range(8)]
    [t.start() for t in ts]
    [t.join() for t in ts]
    check("concurrent pings: no crashes/garbage",
          all(k in ("ok", "busy") for k, _ in results), results)


# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------

def main():
    if "TMUX" not in os.environ:
        print("smoke: must run inside a tmux session", file=sys.stderr)
        return 2

    tmp = tempfile.mkdtemp(prefix="gdbdrv-smoke-")
    src = os.path.join(tmp, "smoke_fixture.c")
    binp = os.path.join(tmp, "smoke_fixture")
    with open(src, "w") as f:
        f.write(FIXTURE_SRC)
    subprocess.check_call(["gcc", "-g", "-O0", "-o", binp, src])

    entry = None
    try:
        entry = pane.spawn(binp)
        print(f"spawned pane={entry['pane']} sock={entry['sock']}\n")
        run_checks(entry["sock"], binp)
    finally:
        if entry:
            pane.stop(entry["sock"])
        # verify no leak
        leaked = any(e["sock"] == (entry or {}).get("sock")
                     for e in pane.list_all())
        check("teardown: pane deregistered/killed", not leaked)

    print()
    if _FAILS:
        print(f"SMOKE FAILED: {len(_FAILS)} check(s): {', '.join(_FAILS)}")
        return 1
    print("SMOKE PASSED")
    return 0


if __name__ == "__main__":
    sys.exit(main())