feat(tools): commit the VM test harness + one-command installer regression
Promotes the throwaway harness that verified the installer into tools/:
- tools/vm/qmp.py QMP keystroke injection / typing / quit
- tools/vm/vncshot.py GL-safe screenshots via VNC readback (QMP
screendump shows "no surface" with virtio-vga-gl)
- tools/vm/gap-analysis.py drv-graph diff that converged the offline
pin set; run it when offline installs build
from source
- tools/test-install.sh the full offline regression: build ISO, boot
offline, unattended LUKS+swap install via a
config disk (typed long commands drop keys),
wait for poweroff, boot the installed disk,
screenshot the first boot for visual verdict
docs/TESTING.md §4 now points at the script.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Bernardo Magri
74
tools/vm/gap-analysis.py
Executable file
74
tools/vm/gap-analysis.py
Executable file
@@ -0,0 +1,74 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Which store paths would an offline nixos-install need that the ISO lacks?
|
||||
|
||||
Walk the target drv graph: an output that's PRESENT needs nothing; a
|
||||
missing output forces building its drv, which needs all its input
|
||||
sources present and all referenced input outputs (recursively).
|
||||
|
||||
This is the maintainer tool that converged the offline pin set in
|
||||
flake.nix (extraDependencies): when an offline install builds from
|
||||
source, run this instead of bisecting the VM. Healthy output is a few
|
||||
dozen per-machine config drvs (etc, initrd-*, unit-*, …) and zero
|
||||
[FETCH/NETWORK] leaves; any real package name here needs pinning.
|
||||
|
||||
Usage:
|
||||
# 1. the present set = everything the ISO carries:
|
||||
live=$(nix build --no-link --print-out-paths \
|
||||
.#nixosConfigurations.nomarchy-live.config.system.build.toplevel)
|
||||
nix path-info -r "$live" | sort -u > /tmp/present.txt
|
||||
# 2. instantiate an install-shaped toplevel for a FOREIGN
|
||||
# username/hostname/uuids (see docs/TESTING.md) and:
|
||||
gap-analysis.py <target.drv> /tmp/present.txt
|
||||
"""
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
target_drv, present_file = sys.argv[1], sys.argv[2]
|
||||
present = set(open(present_file).read().split())
|
||||
|
||||
graph = json.loads(subprocess.check_output(
|
||||
["nix", "derivation", "show", "-r", target_drv + "^*"]))
|
||||
|
||||
must_build = set()
|
||||
missing_srcs = set()
|
||||
visited = set()
|
||||
|
||||
|
||||
def outputs_of(drv_name):
|
||||
return [o["path"] for o in graph[drv_name]["outputs"].values()]
|
||||
|
||||
|
||||
def need_drv(drv_name):
|
||||
if drv_name in visited:
|
||||
return
|
||||
visited.add(drv_name)
|
||||
must_build.add(drv_name)
|
||||
node = graph[drv_name]
|
||||
for src in node.get("inputSrcs", []):
|
||||
if src not in present:
|
||||
missing_srcs.add(src)
|
||||
for dep_drv, want in node.get("inputDrvs", {}).items():
|
||||
dep_outputs = graph.get(dep_drv)
|
||||
if dep_outputs is None:
|
||||
continue
|
||||
wanted = want.get("outputs", []) if isinstance(want, dict) else want
|
||||
for out_name in wanted:
|
||||
path = graph[dep_drv]["outputs"][out_name]["path"]
|
||||
if path not in present:
|
||||
need_drv(dep_drv)
|
||||
break
|
||||
|
||||
|
||||
# Roots: the target's own outputs are by definition "to build".
|
||||
need_drv(target_drv)
|
||||
|
||||
print(f"drvs that must be built offline: {len(must_build)}")
|
||||
for d in sorted(must_build):
|
||||
name = d.split("-", 1)[1] if "-" in d else d
|
||||
builder = graph[d].get("builder", "")
|
||||
fetch = " [FETCH/NETWORK]" if graph[d].get("env", {}).get("urls") or graph[d].get("env", {}).get("url") else ""
|
||||
print(f" {name}{fetch}")
|
||||
print(f"missing input sources: {len(missing_srcs)}")
|
||||
for s in sorted(missing_srcs):
|
||||
print(f" SRC {s}")
|
||||
105
tools/vm/qmp.py
Executable file
105
tools/vm/qmp.py
Executable file
@@ -0,0 +1,105 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Minimal QMP driver for the Nomarchy test VMs.
|
||||
|
||||
Drives a QEMU instance through its QMP socket: inject keystrokes, type
|
||||
text, take screendumps, power off. Used by tools/test-install.sh; also
|
||||
handy interactively while debugging a VM that has no SSH.
|
||||
|
||||
Usage:
|
||||
qmp.py key <combo> # e.g. meta_l-ret, ctrl-alt-f2, ret
|
||||
qmp.py type <text...> # types literal text (US layout)
|
||||
qmp.py screenshot <out.png> # NOTE: with virtio-vga-gl this fails
|
||||
# ("no surface") once the guest uses GL —
|
||||
# use vncshot.py instead for desktops
|
||||
qmp.py quit # hard power-off
|
||||
|
||||
The socket path comes from $NOMARCHY_QMP_SOCK
|
||||
(default /tmp/nomarchy-vm/qmp.sock).
|
||||
|
||||
Keystroke pacing: the guest drops keys when busy (especially during
|
||||
boot). $NOMARCHY_TYPE_DELAY (seconds between keys, default 0.08) trades
|
||||
speed for reliability; verify long typed lines with a screenshot before
|
||||
pressing enter.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import sys
|
||||
import time
|
||||
|
||||
SOCK = os.environ.get("NOMARCHY_QMP_SOCK", "/tmp/nomarchy-vm/qmp.sock")
|
||||
TYPE_DELAY = float(os.environ.get("NOMARCHY_TYPE_DELAY", "0.08"))
|
||||
|
||||
CHAR_KEYS = {
|
||||
" ": "spc", "-": "minus", "=": "equal", "[": "bracket_left",
|
||||
"]": "bracket_right", ";": "semicolon", "'": "apostrophe",
|
||||
"`": "grave_accent", "\\": "backslash", ",": "comma", ".": "dot",
|
||||
"/": "slash", "\n": "ret", "\t": "tab",
|
||||
}
|
||||
SHIFT_CHARS = {
|
||||
"!": "1", "@": "2", "#": "3", "$": "4", "%": "5", "^": "6",
|
||||
"&": "7", "*": "8", "(": "9", ")": "0", "_": "minus", "+": "equal",
|
||||
"{": "bracket_left", "}": "bracket_right", ":": "semicolon",
|
||||
'"': "apostrophe", "~": "grave_accent", "|": "backslash",
|
||||
"<": "comma", ">": "dot", "?": "slash",
|
||||
}
|
||||
|
||||
|
||||
class QMP:
|
||||
def __init__(self):
|
||||
self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
self.s.connect(SOCK)
|
||||
self.f = self.s.makefile("rw")
|
||||
self._read() # greeting
|
||||
self.cmd("qmp_capabilities")
|
||||
|
||||
def _read(self):
|
||||
while True:
|
||||
msg = json.loads(self.f.readline())
|
||||
if "event" in msg:
|
||||
continue
|
||||
return msg
|
||||
|
||||
def cmd(self, name, **args):
|
||||
self.f.write(json.dumps({"execute": name, "arguments": args}) + "\n")
|
||||
self.f.flush()
|
||||
resp = self._read()
|
||||
if "error" in resp:
|
||||
raise RuntimeError(f"{name}: {resp['error']}")
|
||||
return resp
|
||||
|
||||
def send_keys(self, names, hold=40):
|
||||
keys = [{"type": "qcode", "data": n} for n in names]
|
||||
self.cmd("send-key", keys=keys, **{"hold-time": hold})
|
||||
|
||||
def type_text(self, text):
|
||||
for ch in text:
|
||||
if ch.isalpha() and ch.lower() != ch:
|
||||
self.send_keys(["shift", ch.lower()])
|
||||
elif ch in SHIFT_CHARS:
|
||||
self.send_keys(["shift", SHIFT_CHARS[ch]])
|
||||
elif ch in CHAR_KEYS:
|
||||
self.send_keys([CHAR_KEYS[ch]])
|
||||
else:
|
||||
self.send_keys([ch])
|
||||
time.sleep(TYPE_DELAY)
|
||||
|
||||
|
||||
def main():
|
||||
q = QMP()
|
||||
cmd = sys.argv[1]
|
||||
if cmd == "screenshot":
|
||||
q.cmd("screendump", filename=sys.argv[2], format="png")
|
||||
elif cmd == "key":
|
||||
q.send_keys(sys.argv[2].split("-"))
|
||||
elif cmd == "type":
|
||||
q.type_text(" ".join(sys.argv[2:]))
|
||||
elif cmd == "quit":
|
||||
q.f.write(json.dumps({"execute": "quit"}) + "\n")
|
||||
q.f.flush()
|
||||
else:
|
||||
sys.exit(f"unknown command {cmd}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
104
tools/vm/vncshot.py
Executable file
104
tools/vm/vncshot.py
Executable file
@@ -0,0 +1,104 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Grab one full framebuffer from a no-auth VNC server and save as PNG.
|
||||
|
||||
Why VNC and not QMP screendump: with virtio-vga-gl the guest renders
|
||||
into a GL scanout and screendump fails with "no surface"; QEMU's VNC
|
||||
server reads the GL framebuffer back, so this always works. Pair the VM
|
||||
with `-display egl-headless -vnc 127.0.0.1:17`.
|
||||
|
||||
Stdlib only (manual RFB handshake + raw-encoding framebuffer + PNG
|
||||
writer) so it runs anywhere python3 exists.
|
||||
|
||||
Usage: vncshot.py <out.png> [host] [port] # default 127.0.0.1:5917
|
||||
"""
|
||||
import socket, struct, sys, zlib
|
||||
|
||||
def recv_exact(s, n):
|
||||
buf = b""
|
||||
while len(buf) < n:
|
||||
chunk = s.recv(n - len(buf))
|
||||
if not chunk:
|
||||
raise EOFError("server closed")
|
||||
buf += chunk
|
||||
return buf
|
||||
|
||||
def png_write(path, w, h, rgb):
|
||||
def chunk(tag, data):
|
||||
c = struct.pack(">I", len(data)) + tag + data
|
||||
return c + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF)
|
||||
raw = b"".join(b"\x00" + rgb[y * w * 3:(y + 1) * w * 3] for y in range(h))
|
||||
out = (b"\x89PNG\r\n\x1a\n"
|
||||
+ chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0))
|
||||
+ chunk(b"IDAT", zlib.compress(raw, 6))
|
||||
+ chunk(b"IEND", b""))
|
||||
with open(path, "wb") as f:
|
||||
f.write(out)
|
||||
|
||||
def main():
|
||||
out = sys.argv[1]
|
||||
host = sys.argv[2] if len(sys.argv) > 2 else "127.0.0.1"
|
||||
port = int(sys.argv[3]) if len(sys.argv) > 3 else 5917
|
||||
s = socket.create_connection((host, port), timeout=30)
|
||||
s.settimeout(30)
|
||||
|
||||
ver = recv_exact(s, 12) # "RFB 003.008\n"
|
||||
s.sendall(b"RFB 003.008\n")
|
||||
ntypes = recv_exact(s, 1)[0]
|
||||
types = recv_exact(s, ntypes)
|
||||
if 1 not in types:
|
||||
sys.exit(f"no 'None' security type offered: {list(types)}")
|
||||
s.sendall(b"\x01")
|
||||
res = struct.unpack(">I", recv_exact(s, 4))[0]
|
||||
if res != 0:
|
||||
sys.exit("security handshake failed")
|
||||
s.sendall(b"\x01") # ClientInit: shared
|
||||
w, h = struct.unpack(">HH", recv_exact(s, 4))
|
||||
recv_exact(s, 16) # server pixel format (ignored)
|
||||
namelen = struct.unpack(">I", recv_exact(s, 4))[0]
|
||||
recv_exact(s, namelen)
|
||||
|
||||
# SetPixelFormat: 32bpp truecolor, shifts r=16 g=8 b=0 (little-endian BGRX)
|
||||
pf = struct.pack(">BBBBHHHBBBxxx", 32, 24, 0, 1, 255, 255, 255, 16, 8, 0)
|
||||
s.sendall(b"\x00\x00\x00\x00" + pf)
|
||||
s.sendall(struct.pack(">BxH i", 2, 1, 0)) # SetEncodings: Raw only
|
||||
s.sendall(struct.pack(">BBHHHH", 3, 0, 0, 0, w, h)) # full update request
|
||||
|
||||
fb = bytearray(w * h * 4)
|
||||
got = False
|
||||
while not got:
|
||||
mtype = recv_exact(s, 1)[0]
|
||||
if mtype == 0: # FramebufferUpdate
|
||||
recv_exact(s, 1)
|
||||
nrects = struct.unpack(">H", recv_exact(s, 2))[0]
|
||||
for _ in range(nrects):
|
||||
x, y, rw, rh = struct.unpack(">HHHH", recv_exact(s, 8))
|
||||
enc = struct.unpack(">i", recv_exact(s, 4))[0]
|
||||
if enc == 0:
|
||||
data = recv_exact(s, rw * rh * 4)
|
||||
for row in range(rh):
|
||||
off = ((y + row) * w + x) * 4
|
||||
fb[off:off + rw * 4] = data[row * rw * 4:(row + 1) * rw * 4]
|
||||
elif enc == -224: # LastRect
|
||||
break
|
||||
else:
|
||||
sys.exit(f"unsupported encoding {enc}")
|
||||
got = True
|
||||
elif mtype == 2: # Bell
|
||||
pass
|
||||
elif mtype == 3: # ServerCutText
|
||||
recv_exact(s, 3)
|
||||
ln = struct.unpack(">I", recv_exact(s, 4))[0]
|
||||
recv_exact(s, ln)
|
||||
else:
|
||||
sys.exit(f"unexpected server message {mtype}")
|
||||
|
||||
rgb = bytearray(w * h * 3)
|
||||
for i in range(w * h):
|
||||
rgb[i * 3 + 0] = fb[i * 4 + 2]
|
||||
rgb[i * 3 + 1] = fb[i * 4 + 1]
|
||||
rgb[i * 3 + 2] = fb[i * 4 + 0]
|
||||
png_write(out, w, h, bytes(rgb))
|
||||
print(f"saved {out} ({w}x{h})")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user