···
+
from contextlib import contextmanager
+
from xml.sax.saxutils import XMLGenerator
+
def eprint(*args, **kwargs):
+
print(*args, file=sys.stderr, **kwargs)
+
def create_vlan(vlan_nr):
+
log.log("starting VDE switch for network {}".format(vlan_nr))
+
vde_socket = os.path.abspath("./vde{}.ctl".format(vlan_nr))
+
pty_master, pty_slave = pty.openpty()
+
vde_process = subprocess.Popen(
+
["vde_switch", "-s", vde_socket, "--dirmode", "0777"],
+
stdout=subprocess.PIPE,
+
stderr=subprocess.PIPE,
+
fd = os.fdopen(pty_master, "w")
+
# TODO: perl version checks if this can be read from
+
# an if not, dies. we could hang here forever. Fix it.
+
vde_process.stdout.readline()
+
if not os.path.exists(os.path.join(vde_socket, "ctl")):
+
raise Exception("cannot start vde_switch")
+
return (vlan_nr, vde_socket, vde_process, fd)
+
"""Call the given function repeatedly, with 1 second intervals,
+
until it returns True or a timeout is reached.
+
raise Exception("action timed out")
+
self.logfile = os.environ.get("LOGFILE", "/dev/null")
+
self.logfile_handle = open(self.logfile, "wb")
+
self.xml = XMLGenerator(self.logfile_handle, encoding="utf-8")
+
self.queue = queue.Queue(1000)
+
self.xml.startDocument()
+
self.xml.startElement("logfile", attrs={})
+
self.xml.endElement("logfile")
+
self.logfile_handle.close()
+
def sanitise(self, message):
+
return "".join(ch for ch in message if unicodedata.category(ch)[0] != "C")
+
def maybe_prefix(self, message, attributes):
+
if "machine" in attributes:
+
return "{}: {}".format(attributes["machine"], message)
+
def log_line(self, message, attributes):
+
self.xml.startElement("line", attributes)
+
self.xml.characters(message)
+
self.xml.endElement("line")
+
def log(self, message, attributes={}):
+
eprint(self.maybe_prefix(message, attributes))
+
self.log_line(message, attributes)
+
def enqueue(self, message):
+
self.queue.put(message)
+
def drain_log_queue(self):
+
item = self.queue.get_nowait()
+
attributes = {"machine": item["machine"], "type": "serial"}
+
self.log_line(self.sanitise(item["msg"]), attributes)
+
def nested(self, message, attributes={}):
+
eprint(self.maybe_prefix(message, attributes))
+
self.xml.startElement("nest", attrs={})
+
self.xml.startElement("head", attributes)
+
self.xml.characters(message)
+
self.xml.endElement("head")
+
self.log("({:.2f} seconds)".format(toc - tic))
+
self.xml.endElement("nest")
+
def __init__(self, args):
+
self.name = args["name"]
+
cmd = args["startCommand"]
+
self.name = re.search("run-(.+)-vm$", cmd).group(1)
+
self.script = args.get("startCommand", self.create_startcommand(args))
+
tmp_dir = os.environ.get("TMPDIR", tempfile.gettempdir())
+
path = os.path.join(tmp_dir, name)
+
os.makedirs(path, mode=0o700, exist_ok=True)
+
self.state_dir = create_dir("vm-state-{}".format(self.name))
+
self.shared_dir = create_dir("xchg-shared")
+
self.logger = args["log"]
+
self.allow_reboot = args.get("allowReboot", False)
+
def create_startcommand(args):
+
net_backend = "-netdev user,id=net0"
+
net_frontend = "-device virtio-net-pci,netdev=net0"
+
if "netBackendArgs" in args:
+
net_backend += "," + args["netBackendArgs"]
+
if "netFrontendArgs" in args:
+
net_frontend += "," + args["netFrontendArgs"]
+
"qemu-kvm -m 384 " + net_backend + " " + net_frontend + " $QEMU_OPTS "
+
hda_path = os.path.abspath(args["hda"])
+
if args.get("hdaInterface", "") == "scsi":
+
+ ",werror=report,if=none "
+
+ "-device scsi-hd,drive=hda "
+
start_command += "-cdrom " + args["cdrom"] + " "
+
"-device piix3-usb-uhci -drive "
+
+ "-device usb-storage,drive=usbdisk "
+
start_command += "-bios " + args["bios"] + " "
+
start_command += args.get("qemuFlags", "")
+
return self.booted and self.connected
+
self.logger.log(msg, {"machine": self.name})
+
def nested(self, msg, attrs={}):
+
my_attrs = {"machine": self.name}
+
return self.logger.nested(msg, my_attrs)
+
def wait_for_monitor_prompt(self):
+
answer = self.monitor.recv(1024).decode()
+
if answer.endswith("(qemu) "):
+
def send_monitor_command(self, command):
+
message = ("{}\n".format(command)).encode()
+
self.log("sending monitor command: {}".format(command))
+
self.monitor.send(message)
+
return self.wait_for_monitor_prompt()
+
def wait_for_unit(self, unit, user=None):
+
info = self.get_unit_info(unit, user)
+
state = info["ActiveState"]
+
raise Exception('unit "{}" reached state "{}"'.format(unit, state))
+
if state == "inactive":
+
status, jobs = self.systemctl("list-jobs --full 2>&1", user)
+
info = self.get_unit_info(unit)
+
if info["ActiveState"] == state:
+
'unit "{}" is inactive and there ' "are no pending jobs"
+
def get_unit_info(self, unit, user=None):
+
status, lines = self.systemctl('--no-pager show "{}"'.format(unit), user)
+
line_pattern = re.compile(r"^([^=]+)=(.*)$")
+
def tuple_from_line(line):
+
match = line_pattern.match(line)
+
return match[1], match[2]
+
for line in lines.split("\n")
+
if line_pattern.match(line)
+
def systemctl(self, q, user=None):
+
q = q.replace("'", "\\'")
+
"$'XDG_RUNTIME_DIR=/run/user/`id -u` "
+
return self.execute("systemctl {}".format(q))
+
def execute(self, command):
+
out_command = "( {} ); echo '|!EOF' $?\n".format(command)
+
self.shell.send(out_command.encode())
+
status_code_pattern = re.compile(r"(.*)\|\!EOF\s+(\d+)")
+
chunk = self.shell.recv(4096).decode()
+
match = status_code_pattern.match(chunk)
+
status_code = int(match[2])
+
return (status_code, output)
+
def succeed(self, *commands):
+
"""Execute each command and check that it succeeds."""
+
for command in commands:
+
with self.nested("must succeed: {}".format(command)):
+
status, output = self.execute(command)
+
self.log("output: {}".format(output))
+
"command `{}` failed (exit code {})".format(command, status)
+
def fail(self, *commands):
+
"""Execute each command and check that it fails."""
+
for command in commands:
+
with self.nested("must fail: {}".format(command)):
+
status, output = self.execute(command)
+
"command `{}` unexpectedly succeeded".format(command)
+
def wait_until_succeeds(self, command):
+
with self.nested("waiting for success: {}".format(command)):
+
status, output = self.execute(command)
+
def wait_until_fails(self, command):
+
with self.nested("waiting for failure: {}".format(command)):
+
status, output = self.execute(command)
+
def wait_for_shutdown(self):
+
with self.nested("waiting for the VM to power off"):
+
def get_tty_text(self, tty):
+
status, output = self.execute(
+
"fold -w$(stty -F /dev/tty{0} size | "
+
"awk '{{print $2}}') /dev/vcs{0}".format(tty)
+
def wait_until_tty_matches(self, tty, regexp):
+
matcher = re.compile(regexp)
+
with self.nested("waiting for {} to appear on tty {}".format(regexp, tty)):
+
text = self.get_tty_text(tty)
+
if len(matcher.findall(text)) > 0:
+
def send_chars(self, chars):
+
with self.nested("sending keys ‘{}‘".format(chars)):
+
def wait_for_file(self, filename):
+
with self.nested("waiting for file ‘{}‘".format(filename)):
+
status, _ = self.execute("test -e {}".format(filename))
+
def wait_for_open_port(self, port):
+
status, _ = self.execute("nc -z localhost {}".format(port))
+
with self.nested("waiting for TCP port {}".format(port)):
+
def wait_for_closed_port(self, port):
+
status, _ = self.execute("nc -z localhost {}".format(port))
+
def start_job(self, jobname, user=None):
+
return self.systemctl("start {}".format(jobname), user)
+
def stop_job(self, jobname, user=None):
+
return self.systemctl("stop {}".format(jobname), user)
+
def wait_for_job(self, jobname):
+
return self.wait_for_unit(jobname)
+
with self.nested("waiting for the VM to finish booting"):
+
self.log("connected to guest root shell")
+
self.log("(connecting took {:.2f} seconds)".format(toc - tic))
+
def screenshot(self, filename):
+
out_dir = os.environ.get("out", os.getcwd())
+
word_pattern = re.compile(r"^\w+$")
+
if word_pattern.match(filename):
+
filename = os.path.join(out_dir, "{}.png".format(filename))
+
tmp = "{}.ppm".format(filename)
+
"making screenshot {}".format(filename),
+
{"image": os.path.basename(filename)},
+
self.send_monitor_command("screendump {}".format(tmp))
+
ret = subprocess.run("pnmtopng {} > {}".format(tmp, filename), shell=True)
+
if ret.returncode != 0:
+
raise Exception("Cannot convert screenshot")
+
def get_screen_text(self):
+
if shutil.which("tesseract") is None:
+
raise Exception("get_screen_text used but enableOCR is false")
+
"-filter Catrom -density 72 -resample 300 "
+
+ "-contrast -normalize -despeckle -type grayscale "
+
+ "-sharpen 1 -posterize 3 -negate -gamma 100 "
+
tess_args = "-c debug_file=/dev/null --psm 11 --oem 2"
+
with self.nested("performing optical character recognition"):
+
with tempfile.NamedTemporaryFile() as tmpin:
+
self.send_monitor_command("screendump {}".format(tmpin.name))
+
cmd = "convert {} {} tiff:- | tesseract - - {}".format(
+
magick_args, tmpin.name, tess_args
+
ret = subprocess.run(cmd, shell=True, capture_output=True)
+
if ret.returncode != 0:
+
"OCR failed with exit code {}".format(ret.returncode)
+
return ret.stdout.decode("utf-8")
+
def wait_for_text(self, regex):
+
def screen_matches(last):
+
text = self.get_screen_text()
+
m = re.search(regex, text)
+
self.log("Last OCR attempt failed. Text was: {}".format(text))
+
with self.nested("waiting for {} to appear on screen".format(regex)):
+
def send_key(self, key):
+
key = CHAR_TO_KEY.get(key, key)
+
self.send_monitor_command("sendkey {}".format(key))
+
self.log("starting vm")
+
def create_socket(path):
+
if os.path.exists(path):
+
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
+
monitor_path = os.path.join(self.state_dir, "monitor")
+
self.monitor_socket = create_socket(monitor_path)
+
shell_path = os.path.join(self.state_dir, "shell")
+
self.shell_socket = create_socket(shell_path)
+
"" if self.allow_reboot else "-no-reboot",
+
"-monitor unix:{}".format(monitor_path),
+
"-chardev socket,id=shell,path={}".format(shell_path),
+
"-device virtio-serial",
+
"-device virtconsole,chardev=shell",
+
"-device virtio-rng-pci",
+
"-serial stdio" if "DISPLAY" in os.environ else "-nographic",
+
+ os.environ.get("QEMU_OPTS", "")
+
"QEMU_OPTS": qemu_options,
+
"SHARED_DIR": self.shared_dir,
+
environment.update(dict(os.environ))
+
self.process = subprocess.Popen(
+
stdin=subprocess.DEVNULL,
+
stdout=subprocess.PIPE,
+
stderr=subprocess.STDOUT,
+
self.monitor, _ = self.monitor_socket.accept()
+
self.shell, _ = self.shell_socket.accept()
+
def process_serial_output():
+
for line in self.process.stdout:
+
line = line.decode().replace("\r", "").rstrip()
+
eprint("{} # {}".format(self.name, line))
+
self.logger.enqueue({"msg": line, "machine": self.name})
+
_thread.start_new_thread(process_serial_output, ())
+
self.wait_for_monitor_prompt()
+
self.pid = self.process.pid
+
self.log("QEMU running (pid {})".format(self.pid))
+
self.shell.send("poweroff\n".encode())
+
self.wait_for_shutdown()
+
self.log("forced crash")
+
self.send_monitor_command("quit")
+
self.wait_for_shutdown()
+
"""Wait until it is possible to connect to the X server. Note that
+
testing the existence of /tmp/.X11-unix/X0 is insufficient.
+
with self.nested("waiting for the X11 server"):
+
"journalctl -b SYSLOG_IDENTIFIER=systemd | "
+
+ 'grep "Reached target Current graphical"'
+
status, _ = self.execute(cmd)
+
status, _ = self.execute("[ -e /tmp/.X11-unix/X0 ]")
+
"""Make the machine unreachable by shutting down eth1 (the multicast
+
interface used to talk to the other VMs). We keep eth0 up so that
+
the test driver can continue to talk to the machine.
+
self.send_monitor_command("set_link virtio-net-pci.1 off")
+
"""Make the machine reachable.
+
self.send_monitor_command("set_link virtio-net-pci.1 on")
+
def create_machine(args):
+
args["redirectSerial"] = os.environ.get("USE_SERIAL", "0") == "1"
+
with log.nested("starting all VMs"):
+
for machine in machines:
+
with log.nested("waiting for all VMs to finish"):
+
for machine in machines:
+
machine.wait_for_shutdown()
+
exec(os.environ["testScript"])
+
tests = os.environ.get("tests", None)
+
with log.nested("running the VM test script"):
+
eprint("error: {}".format(str(e)))
+
# TODO: Collect coverage data
+
for machine in machines:
+
machine.execute("sync")
+
log.log("{} out of {} tests succeeded".format(nr_succeeded, nr_tests))
+
log.log("error: {}".format(str(e)))
+
if __name__ == "__main__":
+
vlan_nrs = list(dict.fromkeys(os.environ["VLANS"].split()))
+
vde_sockets = [create_vlan(v) for v in vlan_nrs]
+
for nr, vde_socket, _, _ in vde_sockets:
+
os.environ["QEMU_VDE_SOCKET_{}".format(nr)] = vde_socket
+
vm_scripts = sys.argv[1:]
+
machines = [create_machine({"startCommand": s}) for s in vm_scripts]
+
"{0} = machines[{1}]".format(m.name, idx) for idx, m in enumerate(machines)
+
exec("\n".join(machine_eval))
+
with log.nested("cleaning up"):
+
for machine in machines:
+
if machine.pid is None:
+
log.log("killing {} (pid {})".format(machine.name, machine.pid))
+
for _, _, process, _ in vde_sockets:
+
print("test script finished in {:.2f}s".format(toc - tic))
pyrox.dev