rajames/LithosAnanake
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
954b4d5f70894e1d503c29e4f574227bc1a8e007
LithosAnanake/scripts/qemu_session.py
T
rajamesandClaude Opus 5.5 954b4d5f70 feat(kernel): screen=boot keeps the console on the screen until login -- for watching a PC boot 
KernelArgs gains screen_boot; cmdline.c parses screen=boot (with or without --; any other value:
off). With it, the v4 boot keeps the console mirrored to the framebuffer after POST, with the TTF
font, and says 'Screen: the console stays on until login (screen=boot)'; login and logout work as
without it, and nothing can be typed without Zuse. Without it, unchanged. For the SER5, which has
no known serial port (ruled 2026-10-28). BRINGUP.md: the smoke test's stick and starforth.cfg.

Tests: screen_boot (new) failed on amd64 and aarch64 before, passes on all three ISAs after;
screen_dark_until_login, screen_dark_after_logout, zuse pass. Acceptance: amd64, aarch64, riscv64 to
[zuse@Hera] ok>, POST 538/538, dict_hash 0xfec8c47cfc6f437f; v4 suite and test-32 0 failures.

Plan: docs/v4.0.0/plans/2026-10-27-stick-4-one-disk.md, Task 5.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-28 17:50:10 -04:00

646 lines
30 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""qemu_session.py -- one typed QEMU session against a kernel boot, and its verdict.
scripts/qemu_session.py MODE ARCH [--label L] [--make "VAR=val ..."] [--no-zuse]
Builds and boots `make ARCH=<arch> clean qemu` (one QEMU, in the foreground of
this script), with copies of the minted Zuse pair (disk/zuse-scratch/) made
under build/qemu-session/<label>/ so the committed images are never booted;
Zuse's drive is that pair laid out as a stick (zusestick.img).
It reads the serial socket continuously, drives the boot over the serial and
QMP sockets the `qemu` goal opens, prints PASS:/FAIL: lines and
RESULT: PASS|FAIL, and quits QEMU over QMP. The serial log is the usual
logs/<TS>/<arch>/qemu-<arch>-<TS>.log.
Modes (the identity gate's, plan 2026-10-26-engine-step4-identity-gate.md, and
the stick's, plan 2026-10-26-stick-1-screen-and-keyboard.md):
headless no Zuse drive: POST: PASSED, then no `ok> ` for 60 s
zuse the minted pair: identity confirmed, [zuse@Hera] ok>, `3 4 + .` -> 7 ok
hotplug no drive; Zuse added over QMP -> prompt; pulled -> session ended,
headless; added again -> prompt
midpull pulled while a line is in hand: what is typed after does not run
waitpull pulled while a line WAITS: the wait ends at logout; re-attach -> prompt
typeahead serial bytes typed while headless do not run once Zuse attaches
emergency an EMERGENCY_CONSOLE_ENABLED=1 build: ok> with no drive
v3_keys system=v3: `3 4 + .` typed on QEMU's keyboard (send-key) -> 7 ok
screen_dark_until_login no drive: 30 s after POST the screen is all black
screen_console_after_login Zuse added: the screen shows the console; keys typed
on QEMU's keyboard answer 7 ok and change the screen
screen_dark_after_logout ... and when her drive is pulled, the screen is black again
kbd_typeahead keys pressed while headless do not run once Zuse attaches
usb_kbd the only keyboard a USB one (--make "QEMU_EXTRA='-machine
i8042=off -device usb-kbd,bus=xhci0.0,port=3'"): it types
no_framebuffer no display (-vga none): serial alone, 7 ok
alttab_after_logout after logout, Alt+TAB twice: the screen stays black
alttab_before_pull Alt+TAB, then the drive pulled: black; re-attach draws the console
emergency_pull EMERGENCY_CONSOLE_ENABLED=1: Zuse added and pulled, the screen
still shows the running console, and it types
gpt_zuse a GPT drive, Zuse in its StarshipOS identity partition: recognised; the partition
is attached (run with --make KERNEL_ARGS=--log-level=info)
gpt_hotplug ... pulled: session ended; added again: prompt
gpt_foreign a GPT drive with only a Linux partition: no identity, not attached (info level)
gpt_bad_crc the gpt_zuse image with its header CRC broken: not used, said so, not attached
gpt_bad_entries ... its entry array changed (the header intact): the same
gpt_no_header ... its "EFI PART" signature gone, the protective MBR kept: the same
gpt_read_error the raw Zuse drive with reads of its sector 0 failing (blkdebug): not used,
said so, not attached
virtio_kbd aarch64/riscv64: QEMU's virtio keyboard types 7 ok
stick_boot --target qemu-stick (amd64): the PC stick booted as a USB stick through OVMF; its
blank identity partition attached (500 MiB, not the 2 GiB disk); headless
stick_artemis ARTDISK= (no virtio disk): Artemis and Zuse from the Zuse stick alone -- the
stick's Artemis partition attached first (785938 user blocks), then its identity
(48724); [zuse@Hera] ok> (plan 2026-10-27-stick-4-one-disk.md, Task 4)
stick_zuse --target qemu-stick, the PC stick with Zuse's pair in it: a stick, Artemis attached,
[zuse@Hera] ok>
two_sticks --make "DATASTICK=1 ...": Zuse's stick on port 1, a blank data stick on port 2;
Zuse logs in; the data stick is "not this system's stick -- not used" (replaces
datastick)
artemis_unsigned the Zuse stick with its Artemis signature zeroed: "signature is bad -- not
used", nothing attached, headless
raw_ident Zuse's raw identity image, no GPT: "not this system's stick -- not used", headless
screen_boot KERNEL_ARGS=screen=boot, no drive: "Screen: the console stays on until login
(screen=boot)"; 30 s after POST the screen is not black; no ok> (plan
2026-10-27-stick-4-one-disk.md, Task 5)
stick_v3 --target qemu-stick, system=v3 in the stick's starforth.cfg: "Boot disk: a StarshipOS
stick", the refusal, then v4's POST; no v3 (plan 2026-10-27-stick-4-one-disk.md, Task 3)
entropy after "System: StarForth v4", the entropy backend -- rdrand on amd64 (qemu and
--target qemu-stick), virtio-rng on aarch64 and riscv64 -- then "entropy: ready";
on the qemu target, [zuse@Hera] ok> (plan 2026-10-27-stick-4-one-disk.md, Task 2)
Needs only Python 3. One QEMU at a time (.claude/CLAUDE.md).
"""
import argparse, glob, json, os, re, shlex, shutil, socket, subprocess, sys, threading, time
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
PROMPT = r"\[zuse@Hera\][^\n]*ok> "
ANYPROMPT = r"ok> "
ap = argparse.ArgumentParser()
ap.add_argument("mode")
ap.add_argument("arch")
ap.add_argument("--label", default=None)
ap.add_argument("--make", default="")
ap.add_argument("--no-zuse", action="store_true")
ap.add_argument("--target", default="qemu") # qemu-stick: boot the PC stick (amd64)
a = ap.parse_args()
NO_ZUSE_MODES = {"gpt_zuse", "gpt_hotplug", "gpt_foreign", "gpt_bad_crc", "gpt_bad_entries", "gpt_no_header",
"gpt_read_error", "headless", "hotplug", "midpull", "waitpull", "typeahead", "emergency",
"alttab_after_logout", "alttab_before_pull", "emergency_pull",
"screen_dark_until_login", "screen_console_after_login", "screen_dark_after_logout",
"kbd_typeahead", "stick_boot", "stick_v3", "stick_zuse", "screen_boot"}
label = a.label or f"{a.mode}-{a.arch}"
work = os.path.join(ROOT, "build", "qemu-session", label)
shutil.rmtree(work, ignore_errors=True)
os.makedirs(work)
for f in ("artemis.img", "zuse-thumb-ident.img"):
subprocess.run(["cp", "--sparse=always", os.path.join(ROOT, "disk/zuse-scratch", f), work], check=True)
art = os.path.join(work, "artemis.img")
ident = os.path.join(work, "zuse-thumb-ident.img") # the raw identity image (GPT-mode payloads)
# Zuse's drive is her pair as a stick (plan 2026-10-27-stick-4-one-disk.md, Task 1): identity and
# Artemis partitions on one disk, built from this session's copies
zimg = os.path.join(work, "zusestick.img")
subprocess.run(["bash", os.path.join(ROOT, "scripts/mkdiskimage.sh"), "--out", zimg, "--table", "gpt",
"--label", "ZUSESTICK", "--esp-mib", "0", "--starship-from", ident, art],
check=True, stdout=subprocess.DEVNULL)
if a.mode == "artemis_unsigned":
# the Zuse stick with the 4 KiB artemis_sig_t zeroed: it is devblock total - 1 - 64
# (artemis_sig.c, ARTEMIS_SIG_DEVBLOCK_FROM_TOP), i.e. 65 4 KiB devblocks below the top of the
# Artemis partition, which starts 1 MiB + |ident| in
unsigned = os.path.join(work, "unsigned.img")
subprocess.run(["cp", "--sparse=always", zimg, unsigned], check=True)
sig_at = 1048576 + os.path.getsize(ident) + os.path.getsize(art) - 65 * 4096
with open(unsigned, "r+b") as f:
f.seek(sig_at)
f.write(bytes(4096))
attach_zuse = not (a.no_zuse or a.mode in NO_ZUSE_MODES)
# v3 reads identity drives raw, never through a GPT (only the v4 path reads one: sk_v4_identity.c),
# so a system=v3 boot gets Zuse's raw identity image, not her stick (plan 2026-10-27-stick-4, R2)
zdrive = {"v3_keys": ident, "raw_ident": ident}.get(a.mode, zimg)
if a.mode == "artemis_unsigned":
zdrive = unsigned
make_args = ["make", "-f", "kernel/Makefile", f"ARCH={a.arch}", "clean", a.target, f"ARTDISK={art}",
f"ZUSEDISK={zdrive if attach_zuse else ''}"] + shlex.split(a.make)
if a.mode in ("v3_keys", "stick_v3"):
make_args.append("KERNEL_ARGS=system=v3")
if a.mode == "screen_boot":
make_args.append("KERNEL_ARGS=screen=boot")
if a.mode == "stick_zuse":
make_args.append(f"STICK_PAIR={work}") # the PC stick with this session's copy of the pair
start = time.time()
make_log = open(os.path.join(work, "make.txt"), "w")
mk = subprocess.Popen(make_args, cwd=ROOT, stdout=make_log, stderr=subprocess.STDOUT)
deadline = time.time() + 1800
ok = True
def newest(pattern):
c = [p for p in glob.glob(pattern) if os.path.getmtime(p) >= start]
return max(c, key=os.path.getmtime) if c else None
log = ser = qmp = None
while time.time() < deadline and not (log and ser and qmp):
if mk.poll() is not None:
print(f"MAKE FAILED (exit {mk.returncode}); see {make_log.name}", flush=True)
sys.exit(1)
log = log or newest(f"{ROOT}/logs/*/{a.arch}/qemu-{a.arch}-*.log")
ser = ser or newest(f"{ROOT}/build/{a.arch}/kernel/qemu-serial-{a.arch}-*.sock")
qmp = qmp or newest(f"{ROOT}/build/{a.arch}/kernel/qemu-qmp-{a.arch}-*.sock")
time.sleep(1)
print(f"LOG: {os.path.relpath(log, ROOT)}", flush=True)
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
for _ in range(60):
try:
s.connect(ser)
break
except OSError:
time.sleep(1)
def drain():
while True:
try:
if not s.recv(4096):
return
except OSError:
return
threading.Thread(target=drain, daemon=True).start()
q = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
q.connect(qmp)
qf = q.makefile("r")
qf.readline()
def qcmd(cmd, **args):
msg = {"execute": cmd}
if args:
msg["arguments"] = args
q.send(json.dumps(msg).encode())
while True:
r = json.loads(qf.readline())
if "return" in r or "error" in r:
if "error" in r:
print(f"QMP {cmd}: {json.dumps(r)}", flush=True)
return r
qcmd("qmp_capabilities")
def text():
with open(log, errors="replace") as f:
return f.read()
def count(pat):
return len(re.findall(pat, text()))
def wait_for(pat, secs, after=0):
t = time.time()
while time.time() - t < secs:
if count(pat) > after:
return True
if "POST: FAILED" in text() or "PARITY:FAIL" in text():
return False
time.sleep(0.5)
return False
def check(name, cond):
global ok
print(f"{'PASS' if cond else 'FAIL'}: {name}", flush=True)
ok = ok and cond
def type_line(line):
"""Over the serial line, a character at a time."""
for ch in line:
s.send(ch.encode())
time.sleep(0.05)
s.send(b"\r")
# QEMU qcodes for what the tests type on a keyboard
QCODE = {" ": ["spc"], ".": ["dot"], "+": ["shift", "equal"], "-": ["minus"], "\n": ["ret"]}
def key_line(line):
"""On QEMU's keyboard (send-key), a key at a time."""
for ch in line + "\n":
keys = QCODE.get(ch, [ch.lower()])
qcmd("send-key", keys=[{"type": "qcode", "data": k} for k in keys])
time.sleep(0.1)
# The two StarshipOS partition type GUIDs (docs/v4.0.0/plans/2026-10-26-stick-2-gpt-reader.md)
GPT_IDENTITY = "23BEBFE1-4817-4CF3-A537-9069BF159788"
GPT_LINUX = "0FC63DAF-8483-4772-8E79-3D69D8477DE4"
def make_gpt_image(path, ptype, payload, mib=72, break_crc=False, break_entries=False, break_sig=False):
"""A GPT disk image: one partition at LBA 2048, 131072 sectors (64 MiB), of type
`ptype`, holding `payload` from its start; optionally the primary header's CRC broken,
a byte of the entry array changed (the header left valid), or the header's "EFI PART"
signature zeroed (the protective MBR left in place)."""
subprocess.run(["truncate", "-s", f"{mib}M", path], check=True)
sf = shutil.which("sfdisk") or "/usr/sbin/sfdisk"
subprocess.run([sf, "--quiet", path], input=f"label: gpt\nstart=2048, size=131072, type={ptype}\n",
text=True, check=True)
subprocess.run(["dd", f"if={payload}", f"of={path}", "bs=512", "seek=2048", "conv=notrunc,sparse",
"status=none"], check=True)
if break_crc:
with open(path, "r+b") as f:
f.seek(512 + 16) # the primary header's HeaderCRC32
b = f.read(1)
f.seek(512 + 16)
f.write(bytes([b[0] ^ 0xFF]))
if break_entries:
with open(path, "r+b") as f:
f.seek(1024 + 56) # entry 0's name, inside the CRC'd entry array
b = f.read(1)
f.seek(1024 + 56)
f.write(bytes([b[0] ^ 0xFF]))
if break_sig:
with open(path, "r+b") as f:
f.seek(512) # the primary header's Signature
f.write(bytes(8))
def blk_lines():
return count(r"blk: ")
def alt_tab():
qcmd("send-key", keys=[{"type": "qcode", "data": "alt"}, {"type": "qcode", "data": "tab"}])
time.sleep(1)
SHOTS = 0
def screen():
"""A screendump of QEMU's display: (width, height, number of non-black pixels, bytes)."""
global SHOTS
SHOTS += 1
path = os.path.join(work, f"screen-{SHOTS}.ppm")
qcmd("screendump", filename=path)
for _ in range(20):
if os.path.exists(path) and os.path.getsize(path) > 0:
break
time.sleep(0.5)
with open(path, "rb") as f:
data = f.read()
parts = data.split(b"\n", 3) # P6, "w h", maxval, pixels
w, h = (int(x) for x in parts[1].split())
px = parts[3]
lit = sum(1 for i in range(0, len(px) - 2, 3) if px[i] or px[i + 1] or px[i + 2])
return w, h, lit, px
def add_zuse(n, img=None, fail_sector0=False):
f = {"driver": "file", "filename": img or zimg}
if fail_sector0: # every read touching sector 0 fails, EIO
f = {"driver": "blkdebug", "image": f,
"inject-error": [{"event": "none", "iotype": "read", "sector": 0, "errno": 5}]}
qcmd("blockdev-add", driver="raw", **{"node-name": f"zuse{n}"}, file=f)
qcmd("device_add", driver="usb-storage", bus="xhci0.0", port="1",
drive=f"zuse{n}", id=f"zusedev{n}")
def del_zuse(n):
qcmd("device_del", id=f"zusedev{n}")
time.sleep(3)
qcmd("blockdev-del", **{"node-name": f"zuse{n}"})
check("POST: PASSED", wait_for(r"POST: PASSED", 900))
m = a.mode
if m == "headless":
check("no ok> for 60 s after POST", not wait_for(ANYPROMPT, 60))
elif m == "zuse":
check("Boot disk: not a StarshipOS stick (an ISO or ESP boot)",
re.search(r"Boot disk: not a StarshipOS stick", text()) is not None)
check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
n = count(PROMPT)
type_line("3 4 + .")
check("7 ok, and a prompt", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m in ("stick_boot", "two_sticks"):
def disks():
# "us", not "user": the kernel cuts log messages at 63 characters (LOG_LINE_MAX 64,
# v3/include/vm.h), and with seven-digit LBNs the attach line ends "(383542 us"
return [int(u) for u in re.findall(r"blk: disk .*?\((\d+) us", text())]
if m == "stick_boot":
# the PC stick: Artemis is the rest of 2 GiB (well over 1 million user blocks), its blank
# identity 500 MiB (about 380000)
check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", text()) is not None)
check("System: StarForth v4", re.search(r"System: StarForth v4", text()) is not None)
check("Artemis: the stick's partition attached",
wait_for(r"Artemis: the stick's partition attached", 120))
check("Zuse: no genesis on this disk", wait_for(r"Zuse: no genesis on this disk", 60))
time.sleep(5)
d = disks()
check(f"Artemis attached first, then the identity partition (user blocks {d})",
len(d) >= 2 and d[0] > 1000000 and 300000 < d[1] < 512000)
check("no ok> for 30 s (a blank identity: headless)", not wait_for(ANYPROMPT, 30))
else:
check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
n = count(PROMPT)
type_line("3 4 + .")
check("7 ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
check(f"two USB drives found ({count(r'blkio_usb: MSC device ready')})",
count(r"blkio_usb: MSC device ready") >= 2)
time.sleep(5)
nu = count(r"USB: a drive that is not this system's stick -- not used")
check(f"the data stick: not this system's stick -- not used (said {nu} times)", nu == 1)
d = disks()
check(f"nothing of the data stick attached (user blocks {d})", not any(300000 < u < 512000 for u in d))
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m in ("stick_artemis", "stick_zuse"):
if m == "stick_zuse":
check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", text()) is not None)
check("Artemis: the stick's partition attached", wait_for(r"Artemis: the stick's partition attached", 120))
check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
check("[zuse@Hera] ok>", wait_for(PROMPT, 60))
if m == "stick_artemis":
d = [int(u) for u in re.findall(r"blk: disk .*?\((\d+) us", text())]
check(f"Artemis first (785938 user blocks), then the identity (48724): {d}", d[:2] == [785938, 48724])
n = count(PROMPT)
type_line("3 4 + .")
check("7 ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()[-400:]) is not None)
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m in ("artemis_unsigned", "raw_ident"):
if m == "artemis_unsigned":
check("Artemis: the stick's signature is bad -- not used",
wait_for(r"Artemis: the stick's signature is bad -- not used", 120))
else:
check("USB: a drive that is not this system's stick -- not used",
wait_for(r"USB: a drive that is not this system's stick -- not used", 120))
check("no ok> for 60 s", not wait_for(ANYPROMPT, 60))
check("no identity", count(r"Zuse: identity confirmed") == 0)
check("no blk: disk line (nothing attached)", count(r"blk: disk ") == 0)
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m == "stick_v3":
t = text()
check("Boot disk: a StarshipOS stick", re.search(r"Boot disk: a StarshipOS stick", t) is not None)
check("System: v3 is not on this stick -- StarForth v4",
re.search(r"System: v3 is not on this stick -- StarForth v4", t) is not None)
check("v4's POST (PARITY:V4_POST)", re.search(r"PARITY:V4_POST tests=\d+ pass=\d+ fail=0", t) is not None)
check("no System: StarForth v3", re.search(r"System: StarForth v3", t) is None)
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m == "entropy":
want = "rdrand" if a.arch == "amd64" else "virtio-rng"
wait_for(r"entropy: (ready|not available)|Zuse:|ok> ", 120) # the block chain runs after POST
t = text()
v4 = re.search(r"System: StarForth v4", t)
after = t[v4.end():] if v4 else ""
check("System: StarForth v4", v4 is not None)
be = re.search(r"rng: backend = (\S+)", after)
check(f"after it, rng: backend = {want} (got {be.group(1) if be else None})",
be is not None and be.group(1) == want)
check("entropy: ready", re.search(r"entropy: ready", after) is not None)
if a.target == "qemu":
check("[zuse@Hera] ok>", wait_for(PROMPT, 120))
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m == "hotplug":
check("no ok> for 20 s with no drive", not wait_for(ANYPROMPT, 20))
add_zuse(1)
check("identity confirmed on attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 60))
n = count(PROMPT)
type_line("3 4 + .")
check("7 ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
del_zuse(1)
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
n = count(ANYPROMPT)
check("no new ok> for 30 s after detach", not wait_for(ANYPROMPT, 30, n))
c = count(r"Zuse: identity confirmed from attached thumbdrive")
p = count(PROMPT)
add_zuse(2)
check("identity confirmed on re-attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120, c))
check("[zuse@Hera] ok> again", wait_for(PROMPT, 60, p))
elif m == "midpull":
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
qcmd("device_del", id="zusedev1")
type_line("3 4 + .")
type_line("5 6 + .")
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
time.sleep(10)
t = text()
check("the line typed after the pull did not run (no 11)", re.search(r"\b11\s+ok", t) is None)
check("no ok> after the session ended", "ok> " not in t[t.rfind("Zuse: session ended"):])
# the replug: identity and Artemis were both let go; Artemis is attached first again
qcmd("blockdev-del", **{"node-name": "zuse1"})
p = count(PROMPT)
add_zuse(2)
check("[zuse@Hera] ok> again after the replug", wait_for(PROMPT, 120, p))
na = count(r"Artemis: the stick's partition attached")
check(f"Artemis attached again: two Artemis lines in all ({na})", na == 2)
elif m == "waitpull":
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
type_line("5 AWAIT")
time.sleep(5)
del_zuse(1)
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
check("the waiting line ended at logout (Interrupted)", wait_for(r"Interrupted", 30))
p = count(PROMPT)
add_zuse(2)
check("identity confirmed on re-attach", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120, 1))
check("[zuse@Hera] ok> again", wait_for(PROMPT, 60, p))
n = count(PROMPT)
type_line("3 4 + .")
check("7 ok after re-attach", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m == "typeahead":
check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
type_line("3 4 + .")
time.sleep(5)
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
time.sleep(5)
check("what was typed while headless did not run (no 7)", re.search(r"\b7\s+ok", text()) is None)
n = count(PROMPT)
type_line("1 2 + .")
check("3 ok typed at the prompt", wait_for(PROMPT, 60, n) and re.search(r"\b3\s+ok", text()) is not None)
elif m == "emergency":
check("ok> with no drive", wait_for(ANYPROMPT, 60))
n = count(ANYPROMPT)
type_line("3 4 + .")
check("7 ok", wait_for(ANYPROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m == "v3_keys":
check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
time.sleep(2)
n = count(PROMPT)
key_line("3 4 + .")
check("7 ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m == "screen_dark_until_login":
time.sleep(30)
w, h, lit, _ = screen()
check(f"the screen ({w}x{h}) is all black before login ({lit} lit)", lit == 0)
elif m == "screen_boot":
check("Screen: the console stays on until login (screen=boot)",
wait_for(r"Screen: the console stays on until login \(screen=boot\)", 60))
time.sleep(30)
w, h, lit, _ = screen()
check(f"the screen ({w}x{h}) shows the console before login ({lit} lit)", lit >= 500)
check("no ok> (no one is logged in)", not wait_for(ANYPROMPT, 5))
check("no fault", count(r"EXCEPTION|PANIC|panic") == 0)
elif m in ("screen_console_after_login", "screen_dark_after_logout"):
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
time.sleep(3)
w, h, lit1, px1 = screen()
check(f"the screen shows the console ({lit1} lit pixels)", lit1 >= 500)
n = count(PROMPT)
key_line("3 4 + .")
check("7 ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
time.sleep(3)
_, _, lit2, px2 = screen()
check("the screen changed with what was typed", px2 != px1)
if m == "screen_dark_after_logout":
del_zuse(1)
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
time.sleep(5)
_, _, lit3, _ = screen()
check(f"the screen is black after logout ({lit3} lit)", lit3 == 0)
elif m == "kbd_typeahead":
check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
key_line("3 4 + .")
time.sleep(5)
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
time.sleep(5)
check("keys pressed while headless did not run (no 7)", re.search(r"\b7\s+ok", text()) is None)
n = count(PROMPT)
key_line("1 2 + .")
check("3 ok typed on the keyboard", wait_for(PROMPT, 60, n) and re.search(r"\b3\s+ok", text()) is not None)
elif m == "usb_kbd":
check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
time.sleep(2)
n = count(PROMPT)
key_line("3 4 + .") # the USB keyboard is the only one (-machine i8042=off)
check("7 ok typed on the USB keyboard", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m in ("no_framebuffer", "virtio_kbd"):
check("[zuse@Hera] ok>", wait_for(PROMPT, 300))
time.sleep(2)
n = count(PROMPT)
if m == "no_framebuffer":
type_line("3 4 + .")
else:
key_line("3 4 + .")
check("7 ok", wait_for(PROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
elif m in ("alttab_after_logout", "alttab_before_pull"):
add_zuse(1)
check("[zuse@Hera] ok> after attach", wait_for(PROMPT, 120))
n = count(PROMPT)
key_line("3 4 + .")
check("7 ok typed", wait_for(PROMPT, 60, n))
if m == "alttab_before_pull":
alt_tab() # to the graphics view, her session on screen
del_zuse(1)
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
time.sleep(5)
if m == "alttab_after_logout":
alt_tab()
alt_tab()
time.sleep(3)
_, _, lit, _ = screen()
check(f"the screen is black after logout ({lit} lit)", lit == 0)
if m == "alttab_before_pull":
p = count(PROMPT)
add_zuse(2)
check("[zuse@Hera] ok> again", wait_for(PROMPT, 120, p))
time.sleep(3)
_, _, lit1, px1 = screen()
check(f"the console is drawn again ({lit1} lit)", lit1 >= 500)
n = count(PROMPT)
key_line("1 2 + .")
check("3 ok typed", wait_for(PROMPT, 60, n))
time.sleep(3)
_, _, _, px2 = screen()
check("the screen changed with what was typed", px2 != px1)
elif m == "emergency_pull":
check("ok> with no drive", wait_for(ANYPROMPT, 60))
add_zuse(1)
check("Zuse: identity confirmed", wait_for(r"Zuse: identity confirmed from attached thumbdrive", 120))
time.sleep(3)
del_zuse(1)
check("session ended on detach", wait_for(r"Zuse: session ended -- reattach to re-authenticate", 60))
time.sleep(5)
_, _, lit1, px1 = screen()
check(f"the screen still shows the console ({lit1} lit)", lit1 >= 500)
n = count(ANYPROMPT)
key_line("3 4 + .")
check("7 ok typed after the pull", wait_for(ANYPROMPT, 60, n) and re.search(r"7\s+ok", text()) is not None)
time.sleep(3)
_, _, _, px2 = screen()
check("the screen changed with what was typed", px2 != px1)
elif m in ("gpt_zuse", "gpt_hotplug", "gpt_foreign", "gpt_bad_crc", "gpt_bad_entries", "gpt_no_header",
"gpt_read_error"):
gimg = os.path.join(work, "gpt.img")
make_gpt_image(gimg, GPT_LINUX if m == "gpt_foreign" else GPT_IDENTITY, ident,
break_crc=(m == "gpt_bad_crc"), break_entries=(m == "gpt_bad_entries"),
break_sig=(m == "gpt_no_header"))
check("no ok> for 10 s with no drive", not wait_for(ANYPROMPT, 10))
b0 = blk_lines()
if m == "gpt_read_error":
add_zuse(1, ident, fail_sector0=True) # the raw Zuse drive: no GPT, but block 0 unreadable
else:
add_zuse(1, gimg)
if m in ("gpt_zuse", "gpt_hotplug"):
# a GPT disk with only an identity partition is not a stick: on v4 it is not used
# (plan 2026-10-27-stick-4-one-disk.md, R2; Zuse's own case is the stick now)
check("not this system's stick -- not used",
wait_for(r"USB: a drive that is not this system's stick -- not used", 60))
check("no ok> for 30 s", not wait_for(ANYPROMPT, 30))
check(f"not attached (blk: lines {b0} -> {blk_lines()})", blk_lines() == b0)
else:
if m in ("gpt_bad_crc", "gpt_bad_entries", "gpt_no_header"):
check("the damaged table is said so", wait_for(r"GPT: a USB drive has a damaged partition table -- not used", 60))
if m == "gpt_read_error":
check("the unreadable drive is said so", wait_for(r"GPT: a USB drive could not be read -- not used", 60))
check("no ok> for 30 s", not wait_for(ANYPROMPT, 30))
check("no identity", count(r"Zuse: identity confirmed") == 0)
check(f"not attached (blk: lines {b0} -> {blk_lines()})", blk_lines() == b0)
else:
check(f"known mode {m}", False)
print(f"RESULT: {'PASS' if ok else 'FAIL'}", flush=True)
q.send(json.dumps({"execute": "quit"}).encode())
mk.wait()
for f in ("v3/include/version.h", "capsules/BLOCK_MAP.md"):
subprocess.run(["git", "checkout", "-q", "--", f], cwd=ROOT)
sys.exit(0 if ok else 2)
Reference in New Issue View Git Blame Copy Permalink