utils/vmbox/lib/qmp.py
Krzysztof kuhy Rudnicki a2d672f7fd vmbox: disposable Arch VM sandbox for destructive system tests
Much of testsAndMisc/linux_configuration cannot be tested on the host: 15
files use `chattr +i` (so a bad test is not simply revertable), the shutdown
installers write units that power the machine off, and boot_recovery installs
pacman hooks. The repo's own CI says as much -- its Ubuntu job runs only the
side-effect-free tests, noting the rest "drive pacman, /etc, systemd or adb".

vmbox runs them in a throwaway Arch guest and verifies the result from the
HOST, which is the part that is not obvious: a successful shutdown kills the
connection carrying the report, and ssh returns 255 for a clean poweroff, a
crash and a dropped link alike. So the verdict comes from two artefacts that
outlive the VM -- a QMP event log written by a recorder attached at launch
(attach later and qemu has already exited), and the serial console log.

Five outcomes are distinguished: clean poweroff (SHUTDOWN guest:true plus a
serial tail reaching "reboot: Power down"), dirty stop, guest-reset (rebooted
instead of powering off), GUEST_PANICKED, and hung. A POWERDOWN event proves
nothing -- it means the request was delivered, not that anything acted on it.

Reset is a qcow2 backing-file trick: the golden image is built once and sealed
(chmod 444 plus a sha256 sidecar verified on every launch, since booting the
base would silently invalidate every overlay), and each sandbox is a thin
overlay, so reset is rm + recreate. No root anywhere -- user-mode networking
with per-VM forwarded ssh ports, and a socket multicast segment for VM-to-VM.

The guest clock is pinned and re-applied on every launch, because the shutdown
installers gate on a 21:00-05:00 window; NTP is masked in the image or it
silently undoes the pin. Repos are cloned from a read-only share scoped to an
explicit directory, never $HOME, and credentials are sentinel values.

`vm build` ends by booting a throwaway sandbox off the sealed image and
refusing to ship if ssh does not answer. Several distinct faults produced an
identical "boots fine, connection reset" symptom; checks inside the build VM
only prove state at seal time, so the gate has to be an end-to-end boot.

Tests: 8 bats cases over the verdict table and name/meta/serial handling;
shellcheck and ruff clean; every file within the 250-line cap.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdyMBGCrGVGDWtJ9xe48f3
2026-08-22 20:47:56 +02:00

85 lines
2.5 KiB
Python
Executable File

#!/usr/bin/env python3
"""One-shot QMP client for vmbox: issue a command and print the reply.
Used for actions that need a round trip rather than passive event recording --
notably `screendump`, which is how locker/X11 tests get a screenshot without
any VNC or SPICE viewer installed on the host.
Usage:
qmp.py <qmp.sock> screendump <out.png>
qmp.py <qmp.sock> status
"""
from __future__ import annotations
import json
import os
import socket
import sys
from typing import Any, TextIO
def _open(path: str) -> tuple[socket.socket, TextIO]:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(20)
sock.connect(path)
stream = sock.makefile("rw", encoding="utf-8", newline="\n")
stream.readline() # greeting
_command(stream, "qmp_capabilities")
return sock, stream
def _command(stream: TextIO, name: str, **args: Any) -> dict[str, Any]:
"""Send one command and return its reply, skipping any events in between."""
payload: dict[str, Any] = {"execute": name}
if args:
payload["arguments"] = args
stream.write(json.dumps(payload) + "\n")
stream.flush()
while True:
line = stream.readline()
if not line:
raise SystemExit(f"qmp: connection closed during '{name}'")
msg = json.loads(line)
if "return" in msg or "error" in msg:
return msg
def main() -> int:
if len(sys.argv) < 3:
print(__doc__, file=sys.stderr)
return 2
path, action = sys.argv[1], sys.argv[2]
if not os.path.exists(path):
print(f"qmp: no such socket: {path} (is the sandbox running?)", file=sys.stderr)
return 1
sock, stream = _open(path)
try:
if action == "screendump":
if len(sys.argv) < 4:
print("qmp: screendump needs an output path", file=sys.stderr)
return 2
out = os.path.abspath(sys.argv[3])
# PNG is native to qemu's screendump; no ImageMagick needed.
reply = _command(stream, "screendump", filename=out, format="png")
elif action == "status":
reply = _command(stream, "query-status")
else:
print(f"qmp: unknown action '{action}'", file=sys.stderr)
return 2
if "error" in reply:
print(f"qmp: {reply['error'].get('desc', reply['error'])}", file=sys.stderr)
return 1
print(json.dumps(reply.get("return", {})))
return 0
finally:
stream.close()
sock.close()
if __name__ == "__main__":
sys.exit(main())