Cross-architecture test suite
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Reproducible QEMU integration tests for danos.
|
||||
|
||||
For each test case it builds the kernel with `-Dtest-case=<name>`, boots it
|
||||
headless in QEMU with the serial port captured to a file, and asserts that the
|
||||
expected marker appears in that output before a timeout. The kernel's serial log
|
||||
is machine-readable (see src/tests.zig and src/arch/*/serial.zig), so no
|
||||
screen-scraping is involved.
|
||||
|
||||
Structured per-architecture so a second CPU (e.g. an AArch64 Raspberry Pi) is a
|
||||
matter of adding an ARCHES entry and its serial/boot support — the test cases and
|
||||
the runner stay the same.
|
||||
|
||||
Usage:
|
||||
python3 test/qemu_test.py # run all cases on the default arch
|
||||
python3 test/qemu_test.py --arch x86_64 # pick an architecture
|
||||
python3 test/qemu_test.py smoke fault-pf # run only named cases
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
WORK = os.path.join(REPO, "zig-out", "qemu-test")
|
||||
|
||||
# --- Architecture configurations -------------------------------------------
|
||||
# Each describes how to build and boot that architecture. Only the fields that
|
||||
# actually differ between architectures live here.
|
||||
ARCHES = {
|
||||
"x86_64": {
|
||||
"zig_flags": [], # build.zig currently pins the kernel to x86_64
|
||||
"qemu": "qemu-system-x86_64",
|
||||
"ovmf_code": "/usr/share/edk2/x64/OVMF_CODE.4m.fd",
|
||||
"ovmf_vars": "/usr/share/edk2/x64/OVMF_VARS.4m.fd",
|
||||
"efi_app": ("EFI/BOOT/BOOTX64.efi", "BOOTX64.efi"), # (dest in ESP, name in zig-out/bin)
|
||||
"kernel": ("danos", "danos"),
|
||||
# Built as a function so we can splice in per-run paths.
|
||||
"qemu_args": lambda a, esp, vars_fd, serial: [
|
||||
"-machine", "q35", "-m", "128M",
|
||||
"-drive", f"if=pflash,format=raw,readonly=on,file={a['ovmf_code']}",
|
||||
"-drive", f"if=pflash,format=raw,file={vars_fd}",
|
||||
"-drive", f"format=raw,file=fat:rw:{esp}",
|
||||
"-net", "none",
|
||||
"-vga", "none", "-device", "VGA,edid=on,xres=1280,yres=720",
|
||||
"-display", "none",
|
||||
"-serial", f"file:{serial}",
|
||||
"-no-reboot", # exit instead of rebooting on a triple fault
|
||||
],
|
||||
},
|
||||
# To add an architecture, e.g. "aarch64": provide its qemu binary, firmware,
|
||||
# boot method, and the AArch64 kernel/serial support in src/arch/aarch64/.
|
||||
}
|
||||
|
||||
# --- Test cases ------------------------------------------------------------
|
||||
# `expect`: a regex that must appear in serial output => pass.
|
||||
# `fail`: optional regex whose appearance => immediate fail.
|
||||
CASES = [
|
||||
{"name": "smoke",
|
||||
"expect": r"DANOS-TEST-RESULT: PASS",
|
||||
"fail": r"DANOS-TEST-RESULT: FAIL"},
|
||||
{"name": "fault-ud", "expect": r"invalid opcode \(vector 6\)"},
|
||||
{"name": "fault-pf", "expect": r"page fault \(vector 14\)"},
|
||||
{"name": "fault-df", "expect": r"double fault \(vector 8\)"},
|
||||
]
|
||||
|
||||
TIMEOUT = 30 # seconds per case
|
||||
|
||||
|
||||
def build(arch, case):
|
||||
cmd = ["zig", "build", f"-Dtest-case={case}"] + arch["zig_flags"]
|
||||
r = subprocess.run(cmd, cwd=REPO, capture_output=True, text=True)
|
||||
if r.returncode != 0:
|
||||
return r.stderr.strip() or r.stdout.strip()
|
||||
return None
|
||||
|
||||
|
||||
def make_esp(arch):
|
||||
"""Assemble a fresh EFI System Partition from the freshly built binaries."""
|
||||
esp = os.path.join(WORK, "esp")
|
||||
if os.path.exists(esp):
|
||||
shutil.rmtree(esp)
|
||||
efi_dest, efi_name = arch["efi_app"]
|
||||
kern_dest, kern_name = arch["kernel"]
|
||||
os.makedirs(os.path.join(esp, os.path.dirname(efi_dest)), exist_ok=True)
|
||||
shutil.copy(os.path.join(REPO, "zig-out", "bin", efi_name), os.path.join(esp, efi_dest))
|
||||
shutil.copy(os.path.join(REPO, "zig-out", "bin", kern_name), os.path.join(esp, kern_dest))
|
||||
return esp
|
||||
|
||||
|
||||
def run_case(arch, case):
|
||||
err = build(arch, case["name"])
|
||||
if err:
|
||||
return False, "build failed:\n" + err
|
||||
|
||||
esp = make_esp(arch)
|
||||
vars_fd = os.path.join(WORK, "vars.fd")
|
||||
shutil.copy(arch["ovmf_vars"], vars_fd)
|
||||
serial = os.path.join(WORK, "serial.log")
|
||||
if os.path.exists(serial):
|
||||
os.remove(serial)
|
||||
|
||||
expect = re.compile(case["expect"])
|
||||
fail = re.compile(case["fail"]) if case.get("fail") else None
|
||||
|
||||
cmd = [arch["qemu"]] + arch["qemu_args"](arch, esp, vars_fd, serial)
|
||||
qemu = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
try:
|
||||
deadline = time.monotonic() + TIMEOUT
|
||||
while time.monotonic() < deadline:
|
||||
time.sleep(0.2)
|
||||
text = ""
|
||||
if os.path.exists(serial):
|
||||
with open(serial, "r", errors="replace") as f:
|
||||
text = f.read()
|
||||
if fail and fail.search(text):
|
||||
return False, "hit failure marker"
|
||||
if expect.search(text):
|
||||
return True, "matched " + repr(case["expect"])
|
||||
if qemu.poll() is not None: # QEMU exited on its own
|
||||
if expect.search(text):
|
||||
return True, "matched " + repr(case["expect"])
|
||||
return False, "QEMU exited before matching (triple fault?)"
|
||||
return False, f"timed out after {TIMEOUT}s without matching {case['expect']!r}"
|
||||
finally:
|
||||
if qemu.poll() is None:
|
||||
qemu.terminate()
|
||||
try:
|
||||
qemu.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
qemu.kill()
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--arch", default="x86_64", choices=sorted(ARCHES))
|
||||
ap.add_argument("cases", nargs="*", help="case names to run (default: all)")
|
||||
args = ap.parse_args()
|
||||
|
||||
arch = ARCHES[args.arch]
|
||||
if not shutil.which(arch["qemu"]):
|
||||
print(f"error: {arch['qemu']} not found on PATH", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
selected = [c for c in CASES if not args.cases or c["name"] in args.cases]
|
||||
os.makedirs(WORK, exist_ok=True)
|
||||
|
||||
print(f"danos qemu tests arch={args.arch} cases={len(selected)}\n")
|
||||
failures = 0
|
||||
for case in selected:
|
||||
print(f" {case['name']:<12} ... ", end="", flush=True)
|
||||
ok, detail = run_case(arch, case)
|
||||
print(("PASS" if ok else "FAIL") + f" ({detail})")
|
||||
if not ok:
|
||||
failures += 1
|
||||
|
||||
print()
|
||||
total = len(selected)
|
||||
print(f"{total - failures}/{total} passed")
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user