#!/usr/bin/env python3 """Talk to qemu's QMP socket: run a command, or wait for an event. `vm.py` drives the human monitor, which is the right interface for sendkey and screendump. This is the other one, and it exists for the thing HMP cannot do: say when something happened inside the guest. A scripted first-boot test has to know when the guest rebooted. HMP has no answer — `info status` reports running either side of a reset, and a script left watching the clock or the qcow2's size is guessing. QMP emits `RESET` as an event, so "the machine rebooted" becomes a thing to wait for rather than a thing to infer. When it does not arrive, the timeout is itself the finding. `set_link` is here for the same reason it is not in vm.py: taking the guest's route away from outside is the only way to do it that does not depend on the guest cooperating. Doing it from inside means `ip link set enp0s2 down` over the ssh session that command arrived on, which kills the session in the middle of the thing being measured — the hand run on 2026-08-25 needed a detached `setsid` script that brought the interface back up at the end, because there is no other way home. From out here the link is a property of the device. Use: qmp.py cmd query-status qmp.py cmd set_link '{"name": "n0", "up": false}' qmp.py wait RESET 900 qmp.py wait RESET 900 '{"guest": true}' qmp.py cmd quit `wait` exits 0 when the event arrives and 1 when it does not, so a scenario script reads the verdict off the exit status. """ import json import os import socket import sys import time SCRATCH = os.environ.get( "VM_STATE", os.path.join(os.path.dirname(os.path.abspath(__file__)), "state") ) SOCK = os.path.join(SCRATCH, "qmp.sock") class QMP: def __init__(self, path=SOCK, connect_timeout=30.0): # qemu creates the socket as it starts, so a scenario that launches # qemu and connects immediately is racing it. Retrying here rather than # sleeping at every call site. end = time.time() + connect_timeout while True: try: self.s = socket.socket(socket.AF_UNIX) self.s.connect(path) break except (FileNotFoundError, ConnectionRefusedError): if time.time() >= end: raise time.sleep(0.2) # A raw buffer rather than socket.makefile. A buffered reader that # raises socket.timeout can leave a partial line in a buffer nobody # can get at again, which turns one slow read into a stream that never # parses. Owning the buffer keeps a timeout a timeout. self.buf = b"" self.events = [] self._read(timeout=30.0) # the greeting self.execute("qmp_capabilities") def _line(self, timeout): """One newline-terminated message from the socket.""" while b"\n" not in self.buf: self.s.settimeout(timeout) chunk = self.s.recv(65536) if not chunk: raise EOFError("qemu closed the QMP socket") self.buf += chunk line, self.buf = self.buf.split(b"\n", 1) return line def _read(self, timeout=None): """One JSON reply, with events set aside rather than returned. Events interleave with replies, so a reader that returned the first message it saw would hand a caller someone else's RESET.""" while True: msg = json.loads(self._line(timeout)) if "event" in msg: self.events.append(msg) continue return msg def execute(self, command, arguments=None, timeout=30.0): payload = {"execute": command} if arguments: payload["arguments"] = arguments self.s.sendall((json.dumps(payload) + "\n").encode()) # `quit` is answered and then the socket goes away, sometimes in that # order and sometimes not, so losing the reply to it is not a failure. try: reply = self._read(timeout) except EOFError: if command == "quit": return {} raise if "error" in reply: raise RuntimeError("%s: %s" % (command, reply["error"].get("desc", reply))) return reply.get("return", {}) def wait_event(self, name, timeout, data=None): """The next `name` event, or None if the timeout runs out first. `data` narrows it to events whose payload carries those keys and values, which is how RESET becomes a question about the guest rather than about the machine: a reset the host asked for and a reboot the guest performed are the same event with a different `reason`, and only one of them is anything being measured. Events already buffered count: one that arrived while a command was in flight is the same event, and dropping it would make this depend on when it was called.""" end = time.time() + timeout def matches(e): if e.get("event") != name: return False payload = e.get("data", {}) return all(payload.get(k) == v for k, v in (data or {}).items()) while True: for i, e in enumerate(self.events): if matches(e): return self.events.pop(i) remaining = end - time.time() if remaining <= 0: return None try: # Nothing but events is expected here, and _read files those # away itself; a reply arriving would be someone else's. self._read(min(remaining, 1.0)) except socket.timeout: continue except EOFError: return None def close(self): try: self.s.close() except OSError: pass def main(): if len(sys.argv) < 2: raise SystemExit(__doc__) verb = sys.argv[1] q = QMP() if verb == "cmd": command = sys.argv[2] arguments = json.loads(sys.argv[3]) if len(sys.argv) > 3 else None print(json.dumps(q.execute(command, arguments))) return if verb == "wait": name = sys.argv[2] seconds = float(sys.argv[3]) if len(sys.argv) > 3 else 300.0 data = json.loads(sys.argv[4]) if len(sys.argv) > 4 else None event = q.wait_event(name, seconds, data) if event is None: print("no %s%s in %gs" % (name, " matching %s" % json.dumps(data) if data else "", seconds), file=sys.stderr) raise SystemExit(1) print(json.dumps(event)) return raise SystemExit(__doc__) if __name__ == "__main__": main()