bfe-core1106-sdk/qemu/tests/qmp.py

#!/usr/bin/env python3
"""Tiny QMP client for the qemu/ device-sim tests.

    qmp.py <socket> screendump <out.ppm>
    qmp.py <socket> tap <x> <y>          # absolute 0..32767 (virtio-tablet)
    qmp.py <socket> quit
    qmp.py <socket> drive <script> <outdir> [--size N]

`drive` runs a whole interaction on ONE connection, so the VM boots once and
the scenario costs a couple of seconds per step instead of a full boot. Its
script is one command per line, '#' comments and blank lines ignored, and
coordinates are PANEL PIXELS (0..size-1, default 720) rather than the
0..32767 absolute axis the raw `tap` takes -- the point of the mode is to write
"tap the gear at 47,676" straight off a screenshot.

    shot NAME              screendump to <outdir>/NAME.ppm
    tap X Y                press, hold past several LVGL poll periods, release
    swipe X1 Y1 X2 Y2 [MS] drag, interpolated so LVGL sees real motion
    fling X1 Y1 X2 Y2      swipe fast enough to leave momentum behind
    sleep SECONDS          let animations settle or timers fire
    echo TEXT              progress marker in the driver's own output

With `--ctl SOCK` (the unix socket run.sh --ctl creates; ui-drive.sh passes
it) the script can also ask the UI itself, through the debug channel in
warden_debug.c. The same channel is what tools/warden-ctl reaches over SSH on
a real panel, so these verbs mean the same thing on the rig and on hardware:

    nav MENU[/TAB]         jump to a page; fails if the page is unknown
    page | stats | hit X Y print the reply, judge nothing
    ctl WORDS...           raw passthrough for anything the channel grows
    assert_page MENU/TAB   the active page is exactly this
    assert_hit X Y CLASS [TEXT...]
                           a real tap at X,Y lands on an object of CLASS whose
                           caption contains TEXT: the pre-tap self-check that
                           reports a MOVED button as such, not as broken
                           behaviour

Every step is recorded to <outdir>/results.jsonl as ok / fail / fatal. An
assertion mismatch is a `fail` and the script continues, so one run reports
every broken expectation. With `--console LOG` the console is checked after
EVERY step for the stage-2 init's "warden-ui EXITED" line, so a crash is a
`fatal` pinned to the step that caused it rather than a stale picture found at
the end. The exit status is non-zero if anything failed.

A step that names an unknown command is a FATAL error rather than a skip: a
silently-ignored line in a scenario is a test that proves nothing.
"""
import json
import re
import socket
import sys
import time

AXIS_MAX = 32767


def rpc(sock, sock_file, obj):
    sock.sendall((json.dumps(obj) + "\n").encode())
    while True:
        line = sock_file.readline()
        if not line:
            raise RuntimeError("QMP connection closed")
        msg = json.loads(line)
        if "return" in msg:
            return msg["return"]
        if "error" in msg:
            raise RuntimeError(f"QMP error: {msg['error']}")
        # asynchronous events are interleaved; skip them


def send_events(s, f, events):
    rpc(s, f, {"execute": "input-send-event", "arguments": {"events": events}})


def abs_ev(axis, value):
    return {"type": "abs", "data": {"axis": axis, "value": value}}


def btn_ev(down):
    return {"type": "btn", "data": {"down": down, "button": "left"}}


def to_axis(px, size):
    """Panel pixels -> the tablet's absolute axis, clamped to the panel."""
    px = max(0, min(size - 1, int(px)))
    return int(px * AXIS_MAX / (size - 1))


def do_tap(s, f, ax, ay, hold=0.2):
    send_events(s, f, [abs_ev("x", ax), abs_ev("y", ay), btn_ev(True)])
    # Hold the press across several LVGL indev poll periods (33 ms each): an
    # instantaneous press+release lands inside one poll and no click is ever
    # registered.
    time.sleep(hold)
    send_events(s, f, [btn_ev(False)])


def do_swipe(s, f, x1, y1, x2, y2, size, ms=400, steps=None):
    """Drag with interpolated motion.

    LVGL decides a gesture from the movement BETWEEN indev polls, so a press at
    the start and a release at the end (with nothing in between) reads as a
    click on whatever was under the finger, not a scroll. The interpolation
    below is the whole reason a scroll can be tested at all; the step count is
    derived from the duration so the poll period always sees a few pixels of
    travel.
    """
    ax1, ay1 = to_axis(x1, size), to_axis(y1, size)
    ax2, ay2 = to_axis(x2, size), to_axis(y2, size)
    if steps is None:
        steps = max(6, int(ms / 25))
    send_events(s, f, [abs_ev("x", ax1), abs_ev("y", ay1), btn_ev(True)])
    time.sleep(0.05)
    for i in range(1, steps + 1):
        t = i / steps
        send_events(s, f, [
            abs_ev("x", int(ax1 + (ax2 - ax1) * t)),
            abs_ev("y", int(ay1 + (ay2 - ay1) * t)),
        ])
        time.sleep(ms / 1000.0 / steps)
    send_events(s, f, [btn_ev(False)])


class Ctl:
    """The UI's debug channel, reached through run.sh --ctl.

    Inside the VM that channel is a FIFO polled by warden-ui (warden_debug.c);
    rootfs/sbin/init bridges it to a pci-serial port that run.sh exposes as a
    unix socket. One command line in, the reply out, then a sentinel line. On
    real hardware flare-edge tools/warden-ctl reaches the identical FIFO over
    SSH, so the vocabulary here is the vocabulary there: a flow script that
    runs against this rig runs against a panel.

    A missing or silent channel is an infrastructure fault and raises; it is
    never reported as an assertion failure, because a rig that cannot ask is
    not evidence about the UI either way.
    """

    SENTINEL = "<<END>>"

    def __init__(self, path, timeout=15.0):
        self.sock = socket.socket(socket.AF_UNIX)
        self.sock.settimeout(timeout)
        self.sock.connect(path)
        self.buf = b""

    def send(self, cmd):
        self.sock.sendall((cmd + "\n").encode())
        lines = []
        while True:
            nl = self.buf.find(b"\n")
            if nl < 0:
                chunk = self.sock.recv(4096)
                if not chunk:
                    raise RuntimeError("control channel closed")
                self.buf += chunk
                continue
            line = self.buf[:nl].decode("utf-8", "replace").rstrip("\r")
            self.buf = self.buf[nl + 1:]
            if line == self.SENTINEL:
                break
            # The tty is in cooked mode, so the command comes back echoed.
            if line.strip() == cmd.strip():
                continue
            lines.append(line)
        return "\n".join(lines).strip()


HIT_RE = re.compile(r'^hit \d+,\d+: (\S+)(?: text="(.*)" box=(-?\d+),(-?\d+),(\d+)x(\d+))?$')


def parse_hit(reply):
    """-> (cls, text, box) with box as 'x1,y1,WxH', or None for 'nothing';
    raises on an unparseable reply."""
    m = HIT_RE.match(reply)
    if not m:
        raise RuntimeError(f"unparseable hit reply: {reply!r}")
    if m.group(1) == "nothing":
        return None
    box = f"{m.group(3)},{m.group(4)},{m.group(5)}x{m.group(6)}" if m.group(3) else ""
    return m.group(1), m.group(2) or "", box


def ui_exited(console_path):
    """The stage-2 init announces a dead UI on the console; the framebuffer
    does not, it just keeps the last frame. Checked after EVERY step so a crash
    is pinned to the step that caused it, not discovered at the end."""
    if not console_path:
        return None
    try:
        with open(console_path, "rb") as fh:
            data = fh.read()
    except OSError:
        return None
    i = data.find(b"warden-ui EXITED")
    if i < 0:
        return None
    return data[i:i + 80].split(b"\n", 1)[0].decode("utf-8", "replace")


def drive(s, f, script_path, outdir, size, ctl_path=None, console_path=None):
    import os
    os.makedirs(outdir, exist_ok=True)
    with open(script_path) as fh:
        lines = fh.readlines()

    ctl = Ctl(ctl_path) if ctl_path else None
    results = open(os.path.join(outdir, "results.jsonl"), "w")
    counts = {"ok": 0, "fail": 0, "fatal": 0}

    def record(lineno, cmd, status, detail=""):
        counts[status] += 1
        results.write(json.dumps({"line": lineno, "cmd": cmd, "status": status,
                                  "detail": detail}) + "\n")
        results.flush()
        tag = {"ok": "", "fail": "FAIL ", "fatal": "FATAL "}[status]
        print(f"  [{lineno}] {tag}{cmd}{(' -- ' + detail) if detail else ''}", flush=True)

    def need_ctl(lineno, cmd):
        if ctl is None:
            sys.exit(f"FATAL: {script_path}:{lineno}: '{cmd}' needs the control "
                     f"channel; run with --ctl (ui-drive.sh passes it)")

    for lineno, raw in enumerate(lines, 1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        parts = line.split()
        cmd, args = parts[0], parts[1:]

        if cmd == "shot":
            name = args[0]
            path = os.path.join(os.path.abspath(outdir), f"{name}.ppm")
            rpc(s, f, {"execute": "screendump", "arguments": {"filename": path}})
            record(lineno, line, "ok")
        elif cmd == "tap":
            x, y = int(args[0]), int(args[1])
            do_tap(s, f, to_axis(x, size), to_axis(y, size))
            record(lineno, line, "ok")
        elif cmd == "swipe":
            ms = int(args[4]) if len(args) > 4 else 400
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]), size, ms)
            record(lineno, line, "ok")
        elif cmd == "fling":
            do_swipe(s, f, int(args[0]), int(args[1]), int(args[2]), int(args[3]),
                     size, ms=120)
            record(lineno, line, "ok")
        elif cmd == "sleep":
            time.sleep(float(args[0]))
            record(lineno, line, "ok")
        elif cmd == "echo":
            print(f"  [{lineno}] {' '.join(args)}", flush=True)
        elif cmd == "nav":
            need_ctl(lineno, cmd)
            reply = ctl.send("nav " + " ".join(args))
            record(lineno, line, "ok" if reply.endswith(": ok") else "fail", reply)
        elif cmd in ("page", "hit", "stats", "ctl"):
            # Query verbs: print the reply, never judge it. `ctl` is a raw
            # passthrough for anything the channel grows later.
            need_ctl(lineno, cmd)
            reply = ctl.send(line if cmd != "ctl" else " ".join(args))
            record(lineno, line, "ok", reply)
        elif cmd == "assert_page":
            need_ctl(lineno, cmd)
            want = " ".join(args)
            got = ctl.send("page")
            record(lineno, line, "ok" if got == want else "fail",
                   f"page is {got!r}" if got != want else "")
        elif cmd == "assert_hit":
            # assert_hit X Y CLASS [box=X1,Y1,WxH] [TEXT...]: a real tap at X,Y
            # would land on an object of CLASS, with that exact bounding box if
            # one is given, whose caption contains TEXT. This is the pre-tap
            # self-check: "the button moved" fails HERE, by name, so it is
            # never mistaken for the behaviour behind the button.
            #
            # The box is the strong identity. Two rows of a list share a class,
            # and an icon's caption is a glyph nobody wants in a spec; the
            # geometry the UI itself reports is what tells one instance from
            # another.
            need_ctl(lineno, cmd)
            x, y, cls = args[0], args[1], args[2]
            rest = args[3:]
            want_box = None
            if rest and rest[0].startswith("box="):
                want_box = rest[0][4:]
                rest = rest[1:]
            want_text = " ".join(rest)
            got = parse_hit(ctl.send(f"hit {x} {y}"))
            if got is None:
                record(lineno, line, "fail", "nothing clickable there")
            elif got[0] != cls:
                record(lineno, line, "fail", f"{got[0]} text={got[1]!r} box={got[2]}")
            elif want_box and got[2] != want_box:
                record(lineno, line, "fail", f"{cls} moved: box={got[2]}")
            elif want_text and want_text not in got[1]:
                record(lineno, line, "fail", f"{cls} text={got[1]!r}")
            else:
                record(lineno, line, "ok")
        else:
            sys.exit(f"FATAL: {script_path}:{lineno}: unknown command '{cmd}'")

        crashed = ui_exited(console_path)
        if crashed:
            record(lineno, line, "fatal", crashed)
            break

    results.close()
    print(f"== {counts['ok']} ok, {counts['fail']} failed, {counts['fatal']} fatal "
          f"-> {os.path.join(outdir, 'results.jsonl')}", flush=True)
    if counts["fail"] or counts["fatal"]:
        sys.exit(1)


def main():
    if len(sys.argv) < 3:
        sys.exit(__doc__)
    path, cmd = sys.argv[1], sys.argv[2]
    need = {"screendump": 4, "tap": 5, "quit": 3, "drive": 5}
    if cmd not in need:
        sys.exit(f"unknown command {cmd}\n{__doc__}")
    if len(sys.argv) < need[cmd]:
        sys.exit(f"{cmd}: missing argument(s)\n{__doc__}")

    size = 720
    if "--size" in sys.argv:
        size = int(sys.argv[sys.argv.index("--size") + 1])
    ctl_path = sys.argv[sys.argv.index("--ctl") + 1] if "--ctl" in sys.argv else None
    console_path = sys.argv[sys.argv.index("--console") + 1] if "--console" in sys.argv else None

    s = socket.socket(socket.AF_UNIX)
    s.connect(path)
    f = s.makefile("r")
    f.readline()  # greeting banner
    rpc(s, f, {"execute": "qmp_capabilities"})

    if cmd == "screendump":
        rpc(s, f, {"execute": "screendump", "arguments": {"filename": sys.argv[3]}})
    elif cmd == "tap":
        do_tap(s, f, int(sys.argv[3]), int(sys.argv[4]))
    elif cmd == "drive":
        drive(s, f, sys.argv[3], sys.argv[4], size, ctl_path, console_path)
    elif cmd == "quit":
        s.sendall(b'{"execute":"quit"}\n')


if __name__ == "__main__":
    main()