diff --git a/image-create b/image-create
index b83c38ea5b..7051b364f3 100755
--- a/image-create
+++ b/image-create
@@ -1,215 +1,311 @@
#!/usr/bin/env python3
-# This file is part of Cockpit.
-#
-# Copyright (C) 2015 Red Hat, Inc.
-#
-# Cockpit is free software; you can redistribute it and/or modify it
-# under the terms of the GNU Lesser General Public License as published by
-# the Free Software Foundation; either version 2.1 of the License, or
-# (at your option) any later version.
-#
-# Cockpit is distributed in the hope that it will be useful, but
-# WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with Cockpit; If not, see .
-
-# image-create -- Make a root image suitable for use with vm-run.
-#
-# Installs the OS indicated by TEST_OS into the image
-# for test machine and tweaks it to be useable with
-# vm-run and testlib.py.
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""image-create - Make a root image suitable for use with vm-run.
+
+Installs the OS indicated by TEST_OS into the image for test machine and tweaks
+it to be useable with vm-run and testlib.py.
+"""
import argparse
+import asyncio
+import contextlib
+import hashlib
import os
-import shutil
-import subprocess
-import sys
import tempfile
+from collections.abc import Mapping
+from pathlib import Path
-from lib.constants import BOTS_DIR
-from machine import testvm
-
-parser = argparse.ArgumentParser(description='Create a virtual machine image')
-parser.add_argument('-v', '--verbose', action='store_true', help='Display verbose progress details')
-parser.add_argument('-s', '--sit', action='store_true', help='Sit and wait if setup script fails')
-parser.add_argument('-n', '--no-save', action='store_true', help='Don\'t save the new image')
-parser.add_argument('-u', '--upload', action='store_true', help='Upload the image after creation')
-parser.add_argument('--no-build', action='store_true', dest='no_build',
- help="Don't build packages and create the vm without build capabilities")
-parser.add_argument("--store", default=None, help="Where to send images")
-parser.add_argument('image', help='The image to create')
-args = parser.parse_args()
-
-# default to --no-build for some images
-if args.image in ["fedora-coreos", "services"]:
- if not args.no_build:
- if args.verbose:
- print("Creating machine without build capabilities based on the image type")
- args.no_build = True
-
-
-class MachineBuilder:
- def __init__(self, machine: testvm.VirtMachine):
- self.machine = machine
- self.iso = self.machine.image.endswith("boot")
- self.payload = self.machine.image.endswith("payload")
- if self.iso:
- self.suffix = ".iso"
- elif self.payload:
- self.suffix = ".tar.gz"
- else:
- self.suffix = ".qcow2"
+from lib import testthing
+from lib.aio.jsonutil import JsonObject
+from lib.constants import IMAGES_DIR, MACHINE_DIR, SCRIPTS_DIR
+from lib.directories import get_images_data_dir
- # Use /var/tmp/ as this is going to be a huge file; /tmp/ is commonly tmpfs
- self.target_file = self.machine.image_file
- fp, self.machine.image_file = tempfile.mkstemp(dir="/var/tmp/", prefix=self.machine.image, suffix=self.suffix)
- os.close(fp)
+WORKAROUNDS = Path(SCRIPTS_DIR) / 'test.thing-workarounds'
- def bootstrap_system(self) -> None:
- assert not self.machine._domain
- os.makedirs(self.machine.run_dir, 0o750, exist_ok=True)
+def format_journal_line(entry: Mapping[str, str]) -> str:
+ def one_of(*keys: str, default: str) -> str:
+ return next(filter(None, (entry.get(k) for k in keys)), default)
- bootstrap_script = os.path.join(testvm.SCRIPTS_DIR, f"{self.machine.image}.bootstrap")
+ stamp = one_of(
+ "_SOURCE_BOOTTIME_TIMESTAMP",
+ "_SOURCE_MONOTONIC_TIMESTAMP",
+ "__MONOTONIC_TIMESTAMP",
+ default="0",
+ )
+ if entry["_TRANSPORT"].lower() == "kernel":
+ ident = "kernel"
+ else:
+ name = one_of("SYSLOG_IDENTIIFIER", "_COMM", "_EXE", default="?")
+ pid = one_of("SYSLOG_PID", "_PID", default="?")
+ ident = f"{name}[{pid}]"
- if os.path.isfile(bootstrap_script):
- subprocess.check_call([bootstrap_script, self.machine.image_file])
- else:
- raise testvm.Failure(f"Unsupported OS {self.machine.image}: {bootstrap_script}")
-
- def run_setup_script(self, script: str) -> None:
- """Prepare a test image further by running some commands in it."""
- self.machine.start()
-
- try:
- self.machine.wait_boot(timeout_sec=300)
- self.machine.upload([os.path.join(testvm.SCRIPTS_DIR, "lib/")], "/var/lib/testvm")
- self.machine.upload([script], "/var/tmp/SETUP")
-
- env = {
- "TEST_OS": self.machine.image,
- "DO_BUILD": "0" if args.no_build else "1",
- "SERVER_REPO_URL": os.environ.get("SERVER_REPO_URL", ""),
- "EXTRAS_REPO_URL": os.environ.get("EXTRAS_REPO_URL", ""),
- "BASEOS_REPO_URL": os.environ.get("BASEOS_REPO_URL", ""),
- "APPSTREAM_REPO_URL": os.environ.get("APPSTREAM_REPO_URL", ""),
- }
- self.machine.message("run setup script on guest")
-
- try:
- self.machine.execute(f"/var/tmp/SETUP {self.machine.image}",
- environment=env, stdout=None, quiet=not self.machine.verbose, timeout=7200)
- self.machine.execute("rm -f /var/tmp/SETUP")
- except subprocess.CalledProcessError as exc:
- if args.sit:
- sys.stderr.write(self.machine.diagnose())
- input("Press RET to continue... ")
- raise testvm.Failure(f"setup failed with code {exc.returncode}\n") from exc
-
- finally:
- self.machine.stop(timeout_sec=500)
-
- def boot_system(self) -> None:
- """Start the system to make sure it can boot, then shutdown cleanly"""
-
- self.machine.start()
- # avoid too long boot times -- they cause painfully long tests, and are a bug
- try:
- timeout = 30
- if self.machine.image == "fedora-rawhide":
- # except with fedora-rawhide, which is slow by design:
- # https://lists.fedoraproject.org/archives/list/devel@lists.fedoraproject.org/thread/HRJT4UF35MRJ7C4ZXVHBFXB4BTW64YSO/
- timeout = 120
- self.machine.wait_boot(timeout_sec=timeout)
- finally:
- self.machine.stop(timeout_sec=30)
-
- def build(self) -> None:
- self.bootstrap_system()
-
- # gather the scripts, separated by reboots
- script = os.path.join(testvm.SCRIPTS_DIR, f"{self.machine.image}.setup")
-
- if not os.path.exists(script):
- return
+ return f"[{float(stamp) / 1_000_000:+.6f}] {ident}: {entry['MESSAGE']}"
- self.machine.message("Running setup script %s" % (script))
- self.run_setup_script(script)
- # make sure we can boot the system
- self.boot_system()
+def print_journal_line(entry: Mapping[str, str]) -> None:
+ if entry.get('PRIORITY', '9') in '012345678' or 'ssh' in entry.get('SYSLOG_IDENTIFIER', ""): # 4 = "Warning"
+ with contextlib.suppress(OSError):
+ print(format_journal_line(entry))
- def save(self) -> None:
- data_dir = testvm.get_images_data_dir()
- os.makedirs(data_dir, 0o750, exist_ok=True)
+def cloud_init_config(image: str) -> JsonObject | None:
+ def extra_file(dest: str) -> JsonObject:
+ source = WORKAROUNDS / Path(dest).name
+ return {
+ "path": dest,
+ "content": source.read_text(),
+ "permissions": "0755" if os.access(source, os.X_OK) else "0644",
+ }
- if not os.path.exists(self.machine.image_file):
- raise testvm.Failure("Nothing to save.")
+ # don't cloud-init images that we already used ignition for
+ if image in ['centos-9-bootc', 'fedora-coreos', 'services']:
+ return None
- if not self.iso and not self.payload:
- rebuild = os.path.join(data_dir, self.machine.image + ".rebuild")
+ ssh_authorized_keys = [Path(MACHINE_DIR, "identity.pub").read_text()]
+ host_key = Path(MACHINE_DIR, "host_key").read_text()
+ host_key_pub = Path(MACHINE_DIR, "host_key.pub").read_text()
+
+ bootcmd = []
+ write_files = {"/usr/bin/vsock-fling"} # weeeee
+ runcmd = []
+
+ # TODO: filter this?
+ runcmd.append("echo 100::55:4e4b:4e4f:574e UNKNOWN >> /etc/hosts")
+
+ if image == 'arch':
+ bootcmd.append("systemctl disable --now systemd-time-wait-sync.service")
+
+ openrc = image.startswith('alpine')
+ if openrc:
+ # ssh-vsock support for OpenRC
+ write_files.add("/usr/local/sbin/tt-vsock-sshd")
+ write_files.add("/etc/init.d/tt-vsock-ssh-service")
+ runcmd.append("rc-update add tt-vsock-ssh-service default")
+ runcmd.append("rc-service tt-vsock-ssh-service start")
+
+ # do this one last: it notifies that we're done
+ runcmd.append("vsock-fling 2 1111 X_SYSTEMD_UNIT_ACTIVE=multi-user.target")
- # Copy image via convert, to make it sparse again
- subprocess.check_call(["qemu-img", "convert", "-c", "-O", "qcow2", self.machine.image_file, rebuild])
-
- # Hash the image here
- sha, _, _rest = subprocess.check_output(
- ["sha256sum", self.machine.image_file if (self.iso or self.payload) else rebuild], text=True
- ).partition(" ")
- if not sha:
- raise testvm.Failure("sha256sum returned invalid output")
-
- name = self.machine.image + "-" + sha + self.suffix
- data_file = os.path.join(data_dir, name)
- if os.path.exists(data_file):
- # shutil.move has trouble when the destination exists but
- # has too restrictive permissions to be overwritten.
- os.unlink(data_file)
- shutil.move(self.machine.image_file if (self.iso or self.payload) else rebuild, data_file)
-
- if not self.iso and not self.payload:
- # Remove temp image file
- os.unlink(self.machine.image_file)
-
- # Update the images symlink
- if os.path.islink(self.target_file):
- os.unlink(self.target_file)
- os.symlink(name, self.target_file)
-
- # Handle alternate images data directory
- image_file = os.path.join(testvm.IMAGES_DIR, name)
- if not os.path.exists(image_file):
- os.symlink(os.path.abspath(data_file), image_file)
-
-
-try:
- if args.image == 'services':
- # deploying candlepin needs oodles of memory
- memory_mb = 3072
else:
- memory_mb = 2048
-
- machine = testvm.VirtMachine(verbose=args.verbose,
- image=args.image,
- memory_mb=memory_mb,
- maintain=True)
- builder = MachineBuilder(machine)
- builder.build()
- if not args.no_save:
- print("Saving...")
- builder.save()
- if args.upload:
- print("Uploading...")
- cmd = [os.path.join(BOTS_DIR, "image-upload"), '--prune-s3']
- if args.store:
- cmd += ["--store", args.store]
- cmd += [args.image]
- subprocess.check_call(cmd)
-
-except testvm.Failure as ex:
- sys.stderr.write("image-create: %s\n" % ex)
- sys.exit(1)
+ # systemd
+
+ if image.startswith(("centos-9-", "rhel-8-", "rhel-9-", "ubuntu-22", "ubuntu-24")):
+ # no `vmm.notify_socket`, `systemd-ssh-generator` `ssh-vsock.socket` support
+
+ if image.startswith(("centos-", "rhel-")):
+ # SELinux vsock fixes: https://issues.redhat.com/browse/RHEL-113647
+ # We need to apply these *before* we can start the socket unit...
+ print("Updating SELinux policy can take a long time... please be patient.")
+ write_files.add("/tmp/tt-ssh-vsock-selinux.sh")
+ runcmd.append("/tmp/tt-ssh-vsock-selinux.sh")
+
+ write_files.update({
+ "/etc/systemd/system/tt-sd_notify.service",
+ "/etc/systemd/system/tt-sshd-vsock.socket",
+ "/etc/systemd/system/tt-sshd-vsock@.service",
+ })
+ runcmd.extend([
+ r"systemctl daemon-reload",
+ r"systemctl enable --now tt-sshd-vsock.socket",
+ r"systemctl enable tt-sd_notify.service",
+ ])
+
+ # do this one last: it notifies that we're done
+ runcmd.append("vsock-fling 2 1111 X_SYSTEMD_UNIT_ACTIVE=multi-user.target")
+
+ return {
+ "users": [
+ "default",
+ {
+ "name": "root",
+ "lock_passwd": False,
+ "plain_text_passwd": "foobar",
+ "groups": "users,wheel",
+ "ssh_authorized_keys": ssh_authorized_keys,
+ },
+ {
+ "name": "admin",
+ "lock_passwd": False,
+ "plain_text_passwd": "foobar",
+ "gecos": "Administrator",
+ "primary_group": "admin",
+ "groups": "users,wheel",
+ "ssh_authorized_keys": ssh_authorized_keys,
+ },
+ ],
+ "ssh_pwauth": True,
+ "ssh_keys": {
+ "rsa_private": host_key,
+ "rsa_public": host_key_pub,
+ },
+ "bootcmd": [("sh", "-c", cmd) for cmd in bootcmd],
+ "write_files": [extra_file(file) for file in write_files],
+ "runcmd": [("sh", "-c", cmd) for cmd in runcmd],
+ "cloud_final_modules": [
+ "scripts-per-once",
+ "scripts-per-boot",
+ "scripts-per-instance",
+ ["scripts-user", "always"],
+ "final-message",
+ ],
+ }
+
+
+class ImageCreate:
+ def __init__(self, *, image: str, no_build: bool, sit: bool, verbose: bool) -> None:
+ self.ui = testthing.UI(status_messages=True, verbose=verbose)
+
+ if image in ["fedora-coreos", "services"] and not no_build:
+ self.ui.print_verbose("Creating machine without build capabilities based on the image type")
+ no_build = True
+
+ self.image = image
+ self.no_build = no_build
+ self.sit = sit
+
+ if image.endswith('boot'):
+ self.suffix = 'iso'
+ elif image.endswith('payload'):
+ self.suffix = 'tar.gz'
+ else:
+ self.suffix = 'qcow2'
+
+ def heading(self, text: str) -> None:
+ self.ui.print(f"\n\033[1m## {text}\033[0m\n")
+
+ async def bootstrap(self, tmpfile: Path) -> None:
+ self.heading("Run bootstrap script on host")
+ bootstrap_script = Path(SCRIPTS_DIR) / f"{self.image}.bootstrap"
+ await self.ui.run(bootstrap_script, tmpfile)
+
+ async def setup(self, tmpfile: Path) -> None:
+ scripts = Path(SCRIPTS_DIR)
+ setup_script = scripts / f"{self.image}.setup"
+ if not setup_script.exists():
+ self.heading("No additional setup required")
+ return
+
+ self.heading("Start guest")
+ cockpit_test_identity = Path(MACHINE_DIR) / "identity"
+ cockpit_test_identity.chmod(0o600)
+
+ cockpit_test_identity_pub = Path(MACHINE_DIR) / "identity.pub"
+
+ # deploying candlepin needs oodles of memory
+ memory = "3072M" if self.image in ["services"] else "2048M"
+
+ cloud_init_user_data = cloud_init_config(self.image)
+
+ with testthing.IpcDirectory() as ipc:
+ async with testthing.VirtualMachine(
+ # attach_console=True,
+ cloud_init_user_data=cloud_init_user_data,
+ boot="mbr" if self.image in ['alpine'] else "efi",
+ credentials={
+ # "ssh.authorized_keys.root": cockpit_test_identity_pub.read_text().strip(),
+ "passwd.plaintext-password.root": "foobar",
+ },
+ identity=(cockpit_test_identity, None),
+ image=tmpfile,
+ ipc=ipc,
+ journal=print_journal_line,
+ memory=memory,
+ networks=[testthing.Network("user")],
+ snapshot=False,
+ sit=self.sit,
+ target="multi-user.target",
+ timeout=120,
+ ui=self.ui,
+ ) as vm:
+ self.heading("Run setup script on guest")
+
+ await vm.scp("-r", scripts / "lib", "vm:/var/lib/testvm")
+ await vm.scp("-r", setup_script, "vm:/var/tmp/SETUP")
+ await vm.execute(
+ "/var/tmp/SETUP",
+ self.image,
+ environment={
+ "TEST_OS": self.image,
+ "DO_BUILD": "0" if self.no_build else "1",
+ "SERVER_REPO_URL": os.environ.get("SERVER_REPO_URL", ""),
+ "EXTRAS_REPO_URL": os.environ.get("EXTRAS_REPO_URL", ""),
+ "BASEOS_REPO_URL": os.environ.get("BASEOS_REPO_URL", ""),
+ "APPSTREAM_REPO_URL": os.environ.get("APPSTREAM_REPO_URL", ""),
+ },
+ stdout=None,
+ )
+
+ self.heading("Shutting down guest")
+
+ async def save(self, tmpfile: Path) -> None:
+ self.heading("Saving image")
+ data_dir = Path(get_images_data_dir())
+ data_dir.mkdir(exist_ok=True, mode=0o750, parents=True)
+
+ with tempfile.NamedTemporaryFile(
+ dir=data_dir,
+ prefix=f"{self.image}-",
+ suffix=".tmp.{self.suffix}",
+ delete_on_close=False,
+ ) as tmp:
+ # Copy image via convert, to make it sparse again
+ tmp_data_file = Path(tmp.name)
+
+ if self.suffix == 'qcow':
+ # re-sparseify
+ await self.ui.run(
+ ("qemu-img", "convert"),
+ ("-f", "qcow2", tmpfile),
+ ("-O", "qcow2", "-c", tmp_data_file),
+ )
+ else:
+ # copy from the tmpdir to the datadir
+ await self.ui.run('cp', tmpfile, tmp_data_file)
+
+ # Hash the image here
+ with tmp_data_file.open("rb") as fp:
+ sha = hashlib.file_digest(fp, "sha256").hexdigest()
+ name = f"{self.image}-{sha}.{self.suffix}"
+
+ # a simple rename always works because it's the same directory
+ final_data_file = data_dir / name
+ self.ui.print_verbose(f"~ mv \\\n {final_data_file} \\\n {name}\n")
+ tmp_data_file.rename(final_data_file)
+ tmp.close()
+
+ images_dir = Path(IMAGES_DIR)
+
+ self.ui.print_verbose(f"~ ln -sf \\\n {final_data_file} \\\n {images_dir}/\n")
+ (images_dir / name).unlink(missing_ok=True)
+ (images_dir / name).symlink_to(final_data_file)
+
+ self.ui.print_verbose(f"~ ln -sf \\\n {name} \\\n {images_dir}/{self.image}\n")
+ (images_dir / self.image).unlink(missing_ok=True)
+ (images_dir / self.image).symlink_to(name)
+
+ async def create(self) -> None:
+ with tempfile.NamedTemporaryFile(
+ dir="/var/tmp", prefix=f"cockpit-image-create-{self.image}-", suffix=f".tmp.{self.suffix}"
+ ) as tmp:
+ tmpfile = Path(tmp.name)
+ await self.bootstrap(tmpfile)
+ await self.setup(tmpfile)
+ await self.save(tmpfile)
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Create a virtual machine image")
+ parser.add_argument("-v", "--verbose", action="store_true", help="Display verbose progress details")
+ parser.add_argument("-s", "--sit", action="store_true", help="Sit and wait if setup script fails")
+ parser.add_argument("--no-build", action="store_true", help="Create the VM without build capabilities")
+ parser.add_argument("image", help="The image to create")
+ args = parser.parse_args()
+
+ with testthing.cli_helper():
+ asyncio.run(ImageCreate(**vars(args)).create())
+
+
+if __name__ == "__main__":
+ main()
diff --git a/image-customize b/image-customize
index 6f3be93924..3f5f87e131 100755
--- a/image-customize
+++ b/image-customize
@@ -17,17 +17,19 @@
# along with Cockpit; If not, see .
import argparse
+import asyncio
import os
import subprocess
-import sys
from collections.abc import Sequence
+from pathlib import Path
-from lib.constants import BOTS_DIR, TEST_DIR
-from machine import testvm
+from lib import testthing
+from lib.constants import BOTS_DIR, IMAGES_DIR, MACHINE_DIR, TEST_DIR
+from lib.testmap import get_test_image
opt_quick: bool = False
opt_verbose: bool = False
-opt_build_options: str = ''
+opt_build_options: str = ""
stdout_disposition: int | None = None
@@ -35,7 +37,7 @@ def prepare_install_image(base_image: str, install_image: str, resize: str | Non
"""Create the necessary layered image for the build/install"""
if "/" not in base_image:
- base_image = os.path.join(testvm.IMAGES_DIR, base_image)
+ base_image = os.path.join(IMAGES_DIR, base_image)
if "/" not in install_image:
install_image = os.path.join(os.path.join(TEST_DIR, "images"), os.path.basename(install_image))
@@ -54,8 +56,16 @@ def prepare_install_image(base_image: str, install_image: str, resize: str | Non
install_image_dir = os.path.dirname(install_image)
os.makedirs(install_image_dir, exist_ok=True)
base_image = os.path.realpath(base_image)
- subprocess.check_call(["qemu-img", "create", "-q", "-f", "qcow2",
- "-o", f"backing_file={base_image},backing_fmt=qcow2", qcow2_image])
+ subprocess.check_call([
+ "qemu-img",
+ "create",
+ "-q",
+ "-f",
+ "qcow2",
+ "-o",
+ f"backing_file={base_image},backing_fmt=qcow2",
+ qcow2_image,
+ ])
if os.path.lexists(install_image):
os.unlink(install_image)
os.symlink(os.path.basename(qcow2_image), install_image)
@@ -70,7 +80,7 @@ class ActionBase(argparse.Action):
"""Keep an ordered list of actions"""
@staticmethod
- def execute(machine_instance: testvm.Machine, argument: str) -> None:
+ async def execute(machine_instance: testthing.VirtualMachine, argument: str, /) -> None:
raise NotImplementedError
def __call__(
@@ -78,68 +88,76 @@ class ActionBase(argparse.Action):
parser: argparse.ArgumentParser,
namespace: argparse.Namespace,
value: str | Sequence[str] | None,
- option_string: object = None
+ option_string: object = None,
) -> None:
getattr(namespace, self.dest).append((self.execute, value))
class InstallAction(ActionBase):
"""Install local rpm or distro package"""
+
@staticmethod
- def execute(machine_instance: testvm.Machine, package: str) -> None:
+ async def execute(machine_instance: testthing.VirtualMachine, package: str) -> None:
# If we have a '/' in the package name, or if a file with that name
# exists in the current directory, then assume that this is a package
# we're uploading from the host.
- if '/' in package or os.path.isfile(package):
+ if "/" in package or os.path.isfile(package):
dest = "/var/tmp/" + os.path.basename(package)
- machine_instance.upload([os.path.abspath(package)], dest)
+ await machine_instance.scp(os.path.abspath(package), f"vm:{dest}")
package = dest
# requesting install of Python wheel?
- if package.endswith('.whl'):
- machine_instance.execute(f"python3 -m pip install --no-index --prefix=/usr/local {package}", timeout=120)
+ if package.endswith(".whl"):
+ await asyncio.wait_for(
+ machine_instance.execute(f"python3 -m pip install --no-index --prefix=/usr/local {package}"),
+ timeout=120,
+ )
return
# this will fail if neither is available -- exception is clear enough, this is a developer tool
- out = machine_instance.execute("which dnf || which yum || which apt-get")
- if 'dnf' in out:
+ out = await machine_instance.execute("which dnf || which yum || which apt-get || which pacman")
+ if "dnf" in out:
install_command = "dnf install -y"
- elif 'yum' in out:
+ elif "yum" in out:
install_command = "yum --setopt=skip_missing_names_on_install=False -y install"
- else:
+ elif "apt-get" in out:
install_command = "apt-get install -y"
+ elif "pacman" in out:
+ install_command = "pacman -S --noconfirm"
+ else:
+ raise NotImplementedError(f"unknown build platform: {out}")
- machine_instance.execute(f"{install_command} {package}", timeout=1800)
+ await machine_instance.execute(f"{install_command} {package}")
class BuildAction(ActionBase):
"""Build and install distribution package(s) from dist tarball or source RPM"""
@staticmethod
- def execute(machine_instance: testvm.Machine, source: str) -> None:
+ async def execute(machine_instance: testthing.VirtualMachine, source: str) -> None:
# upload the tarball or srpm
sourcename = os.path.basename(source)
vm_source = os.path.join("/var/tmp", sourcename)
- machine_instance.upload([source], vm_source, relative_dir=".")
+ await machine_instance.scp(source, f"vm:{vm_source}")
# this will fail if neither is available -- exception is clear enough, this is a developer tool
- out = machine_instance.execute("(which pbuilder || which mock || which pacman) 2>/dev/null")
- if 'pbuilder' in out:
- BuildAction.build_deb(machine_instance, vm_source)
- elif 'mock' in out:
- BuildAction.build_rpm(machine_instance, vm_source)
- elif 'pacman' in out:
- BuildAction.build_arch(machine_instance, vm_source)
+ out = await machine_instance.execute("(which pbuilder || which mock || which pacman) 2>/dev/null")
+ if "pbuilder" in out:
+ await BuildAction.build_deb(machine_instance, vm_source)
+ elif "mock" in out:
+ await BuildAction.build_rpm(machine_instance, vm_source)
+ elif "pacman" in out:
+ await BuildAction.build_arch(machine_instance, vm_source)
else:
raise NotImplementedError(f"unknown build platform: {out}")
@staticmethod
- def build_deb(machine: testvm.Machine, vm_source: str) -> None:
- build_opts = 'nocheck' if opt_quick else ''
+ async def build_deb(machine: testthing.VirtualMachine, vm_source: str) -> None:
+ build_opts = "nocheck" if opt_quick else ""
# build source packge
- machine.execute(f"""
+ await machine.execute(f"""
set -eu
rm -rf /var/tmp/build
mkdir -p /var/tmp/build
@@ -156,27 +174,30 @@ class BuildAction(ActionBase):
dpkg-buildpackage -S -us -uc -nc""")
# build binary packages
- machine.execute(f"cd /var/tmp/build; DEB_BUILD_OPTIONS='{build_opts}' pbuilder build --buildresult . "
- f"{opt_build_options} *.dsc", timeout=1800, stdout=stdout_disposition)
+ await machine.execute(
+ f"cd /var/tmp/build; DEB_BUILD_OPTIONS='{build_opts}' pbuilder build --buildresult . "
+ f"{opt_build_options} *.dsc",
+ stdout=stdout_disposition,
+ )
# install packages
- machine.execute("dpkg -i /var/tmp/build/*.deb")
+ await machine.execute("dpkg -i /var/tmp/build/*.deb")
@staticmethod
- def build_rpm(machine: testvm.Machine, vm_source: str) -> None:
- mock_opts = ''
+ async def build_rpm(machine: testthing.VirtualMachine, vm_source: str) -> None:
+ mock_opts = ""
if opt_verbose:
- mock_opts += ' --verbose'
+ mock_opts += " --verbose"
if opt_quick:
- mock_opts += ' --nocheck'
+ mock_opts += " --nocheck"
if opt_build_options:
- mock_opts += ' ' + opt_build_options
+ mock_opts += " " + opt_build_options
# build source package, unless this is running against an srpm already
if vm_source.endswith(".src.rpm"):
srpm = vm_source
else:
- machine.execute(f"""
+ await machine.execute(f"""
set -eu
rm -rf /var/tmp/build
su builder -c 'rpmbuild --define "_topdir /var/tmp/build" -ts "{vm_source}"'
@@ -184,7 +205,7 @@ class BuildAction(ActionBase):
srpm = "/var/tmp/build/SRPMS/*.src.rpm"
# HACK: mock in openSUSE must be called through sudo (but still originally as builder)
- if "opensuse" in machine.execute("cat /etc/os-release"):
+ if "opensuse" in await machine.execute("cat /etc/os-release"):
mock_sudo = "sudo"
else:
mock_sudo = ""
@@ -192,19 +213,25 @@ class BuildAction(ActionBase):
# build binary RPMs from srpm; disable all repositorys as mock insists on
# calling `dnf builddep`, which insists on a cache; our test VMs don't have a cache,
# as the mock is offline and pre-installed
- machine.execute(f"su builder -c '{mock_sudo} mock --no-clean --no-cleanup-after --disablerepo=* "
- f"--offline --resultdir /var/tmp/build {mock_opts} --rebuild {srpm}'",
- timeout=1800, stdout=stdout_disposition)
+ await machine.execute(
+ f"su builder -c '{mock_sudo} mock --no-clean --no-cleanup-after --disablerepo=* "
+ f"--offline --resultdir /var/tmp/build {mock_opts} --rebuild {srpm}'",
+ stdout=stdout_disposition,
+ )
# install RPMs
- machine.execute('packages=$(find /var/tmp/build -name "*.rpm" -not -name "*.src.rpm"); '
- f'rpm -U --force --verbose {"--nodigest --nosignature" if opt_quick else ""} $packages')
+ await machine.execute(
+ 'packages=$(find /var/tmp/build -name "*.rpm" -not -name "*.src.rpm"); '
+ f"rpm -U --force --verbose {'--nodigest --nosignature' if opt_quick else ''} $packages"
+ )
@staticmethod
- def build_arch(machine: testvm.Machine, vm_source: str) -> None:
+ async def build_arch(machine: testthing.VirtualMachine, vm_source: str) -> None:
# unpack source tree's arch packaging directory (PKGBUILD refers to some files)
# and set PKGBUILD variables
- machine.write("/var/tmp/mkbuild.sh", f"""#!/bin/sh
+ await machine.write(
+ "/var/tmp/mkbuild.sh",
+ f"""#!/bin/sh
set -eu
rm -rf /var/tmp/build
mkdir -p /var/tmp/build
@@ -215,52 +242,49 @@ class BuildAction(ActionBase):
cp "$archdir"/* .
# tarball must be in same directory as PKGBUILD
cp '{vm_source}' /var/tmp/build/
- """, perm="755")
- machine.execute(f"su builder {opt_build_options} /var/tmp/mkbuild.sh")
+ """,
+ perm="755",
+ )
+ await machine.execute(f"su builder {opt_build_options} /var/tmp/mkbuild.sh")
# build binaries
- machine.execute("cd /var/tmp/build; makechrootpkg -r /var/lib/archbuild/cockpit -U builder",
- timeout=1800, stdout=stdout_disposition)
+ await machine.execute(
+ "cd /var/tmp/build; makechrootpkg -r /var/lib/archbuild/cockpit -U builder",
+ stdout=stdout_disposition,
+ )
# install packages
- machine.execute("pacman -U --noconfirm /var/tmp/build/*.pkg.tar.zst")
+ await machine.execute("pacman -U --noconfirm /var/tmp/build/*.pkg.tar.zst")
class RunCommandAction(ActionBase):
@staticmethod
- def execute(machine_instance: testvm.Machine, command: str) -> None:
- try:
- machine_instance.execute(command, timeout=1800)
- except subprocess.CalledProcessError as e:
- sys.stderr.write("%s\n" % e)
- sys.exit(e.returncode)
+ async def execute(machine_instance: testthing.VirtualMachine, command: str) -> None:
+ await machine_instance.execute(command)
class ScriptAction(ActionBase):
@staticmethod
- def execute(machine_instance: testvm.Machine, script: str) -> None:
+ async def execute(machine_instance: testthing.VirtualMachine, script: str) -> None:
uploadpath = "/var/tmp/" + os.path.basename(script)
- machine_instance.upload([os.path.abspath(script)], uploadpath)
- machine_instance.execute("chmod a+x %s" % uploadpath)
- try:
- machine_instance.execute(uploadpath, timeout=1800)
- except subprocess.CalledProcessError as e:
- sys.stderr.write("%s\n" % e)
- sys.exit(e.returncode)
+ await machine_instance.scp(os.path.abspath(script), f"vm:{uploadpath}")
+ await machine_instance.execute("chmod a+x %s" % uploadpath)
+ await machine_instance.execute(uploadpath)
class UploadAction(ActionBase):
@staticmethod
- def execute(machine_instance: testvm.Machine, srcdest: str) -> None:
+ async def execute(machine_instance: testthing.VirtualMachine, srcdest: str) -> None:
src, dest = srcdest.split(":")
abssrc = os.path.abspath(src)
# preserve trailing / for rsync compatibility
- if src.endswith('/'):
- abssrc += '/'
- machine_instance.upload([abssrc], dest)
+ if src.endswith("/"):
+ abssrc += "/"
+ await machine_instance.scp("-r", abssrc, f"vm:{dest}")
-def main() -> None:
+async def main() -> None:
+ # fmt: off
parser = argparse.ArgumentParser(
description=('Run command inside or install packages into a Cockpit virtual machine. '
'All actions can be specified multiple times and run in the given order.'))
@@ -284,7 +308,7 @@ def main() -> None:
help="Additional options for mock/pbuilder/arch builder")
parser.add_argument('--resize', help="Resize the image. Size in bytes with using K, M, or G suffix.")
parser.add_argument('-n', '--no-network', action='store_true', help='Do not connect the machine to the Internet')
- parser.add_argument('--cpus', type=int, default=None,
+ parser.add_argument('--cpus', type=int, default=2,
help="Number of CPUs for the virtual machine")
parser.add_argument('--memory-mb', type=int, default=2048,
help="RAM size for the virtual machine")
@@ -294,6 +318,7 @@ def main() -> None:
help='Disable tests during package build with --build')
parser.add_argument('image', help='The image to use (destination name when using --base-image)')
parser.add_argument('--sit', action='store_true', help='Sit and wait if any VM action fails')
+ # fmt: on
args = parser.parse_args()
if not args.actions and not args.resize:
@@ -302,7 +327,7 @@ def main() -> None:
if not args.base_image:
args.base_image = os.path.basename(args.image)
- args.base_image = testvm.get_test_image(args.base_image)
+ args.base_image = get_test_image(args.base_image)
global opt_quick, opt_verbose, opt_build_options, stdout_disposition
opt_quick = args.quick
@@ -311,30 +336,28 @@ def main() -> None:
if not args.verbose:
stdout_disposition = subprocess.DEVNULL
- if '/' not in args.base_image:
+ if "/" not in args.base_image:
subprocess.check_call([os.path.join(BOTS_DIR, "image-download"), args.base_image])
- network = testvm.VirtNetwork(0, image=args.base_image)
- machine = testvm.VirtMachine(maintain=True,
- verbose=args.verbose,
- networking=network.host(restrict=args.no_network),
- image=prepare_install_image(args.base_image, args.image, args.resize, args.fresh),
- cpus=args.cpus,
- memory_mb=args.memory_mb)
- machine.start()
- machine.wait_boot()
- try:
- for (handler, arg) in args.actions:
- handler(machine, arg)
- except Exception as e:
- if args.sit:
- print(e, file=sys.stderr)
- print(machine.diagnose(), file=sys.stderr)
- print("Press RET to continue...")
- sys.stdin.readline()
- raise e
- finally:
- machine.stop()
-
-
-if __name__ == '__main__':
- main()
+
+ cockpit_test_identity = Path(MACHINE_DIR) / "identity"
+ cockpit_test_identity.chmod(0o600)
+
+ with testthing.cli_helper(), testthing.IpcDirectory() as ipc:
+ async with testthing.VirtualMachine(
+ prepare_install_image(args.base_image, args.image, args.resize, args.fresh),
+ boot="mbr",
+ cpus=args.cpus,
+ identity=(cockpit_test_identity, None),
+ ipc=ipc,
+ memory=f"{args.memory_mb}M",
+ networks=[] if args.no_network else [testthing.Network("user")],
+ sit=args.sit,
+ snapshot=False,
+ verbose=args.verbose,
+ ) as vm:
+ for handler, arg in args.actions:
+ await asyncio.wait_for(handler(vm, arg), timeout=1800)
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
diff --git a/image-diff b/image-diff
index 5a36076030..1b5a9e57b2 100755
--- a/image-diff
+++ b/image-diff
@@ -1,77 +1,70 @@
-#!/usr/bin/python3
-
-# This file is part of Cockpit.
-#
-# Copyright (C) 2020 Red Hat, Inc.
-#
-# Cockpit is free software; you can redistribute it and/or modify it
-# under the terms of the GNU Lesser General Public License as published by
-# the Free Software Foundation; either version 2.1 of the License, or
-# (at your option) any later version.
-#
-# Cockpit is distributed in the hope that it will be useful, but
-# WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
-# Lesser General Public License for more details.
-#
-# You should have received a copy of the GNU Lesser General Public License
-# along with Cockpit; If not, see .
+#!/usr/bin/env python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""image-diff - Compare package versions on VM images.
+
+Gives an overview of which packages have been added/removed/changed between two
+versions of a VM image.
+"""
import argparse
+import asyncio
from collections.abc import Mapping
+from pathlib import Path
+
+from lib import testthing
+from lib.constants import IMAGES_DIR
-from machine import testvm
+async def get_packages(image: str) -> Mapping[str, str]:
+ file = Path(image) if image.startswith("/") else Path(IMAGES_DIR, image)
-def get_packages(machine: testvm.VirtMachine) -> Mapping[str, str]:
- # List all packages, irrespective of package manager (rpm or dpkg or pacman).
- # If both are missing, then the command will fail.
- #
- # We'd ideally like to get source packages everywhere, but it's a
- # bit more difficult on RPM. (TODO)
- pkgcmd = """if type dpkg-query > /dev/null 2>&1; then
- dpkg-query -W 2>/dev/null;
- elif type rpm > /dev/null 2>&1; then
- rpm -qa --qf '%{NAME}\t%{EVR}\n' 2>/dev/null;
- else pacman -Q | sed 's/ /\t/' 2>/dev/null; fi"""
+ with testthing.IpcDirectory() as ipc:
+ async with testthing.VirtualMachine(file, boot="mbr", ipc=ipc, memory="1G") as vm:
+ # List all packages, irrespective of package manager (rpm or dpkg
+ # or pacman). If both are missing, then the command will fail.
+ #
+ # We'd ideally like to get source packages everywhere, but it's a
+ # bit more difficult on RPM. (TODO)
+ pkgcmd = """if type dpkg-query > /dev/null 2>&1; then
+ dpkg-query -W 2>/dev/null;
+ elif type rpm > /dev/null 2>&1; then
+ rpm -qa --qf '%{NAME}\t%{EVR}\n' 2>/dev/null;
+ else pacman -Q | sed 's/ /\t/' 2>/dev/null; fi"""
- output = machine.execute(pkgcmd).strip()
- return dict(line.split('\t') for line in output.splitlines())
+ output = (await vm.execute(pkgcmd)).strip()
+ return dict(line.split("\t") for line in output.splitlines())
-parser = argparse.ArgumentParser(description='Compare package versions on VM images')
-parser.add_argument('old', help='the "old" image to compare')
-parser.add_argument('new', help='the "new" image to compare')
-args = parser.parse_args()
+async def diff(old: str, new: str) -> None:
+ old_pkgs, new_pkgs = await asyncio.gather(get_packages(old), get_packages(new))
-# boot the machines in parallel
-old_vm = testvm.VirtMachine(image=args.old)
-new_vm = testvm.VirtMachine(image=args.new)
+ print("Removed:")
+ for name in sorted(set(old_pkgs) - set(new_pkgs)):
+ print(f" {name} ({old_pkgs[name]})")
+ print()
-old_vm.start()
-new_vm.start()
+ print("Added:")
+ for name in sorted(set(new_pkgs) - set(old_pkgs)):
+ print(f" {name} ({new_pkgs[name]})")
+ print()
-old_vm.wait_boot()
-new_vm.wait_boot()
+ print("Changed:")
+ for name in sorted(set.intersection(set(new_pkgs), set(old_pkgs))):
+ if new_pkgs[name] != old_pkgs[name]:
+ print(f" {name} ({old_pkgs[name]} -> {new_pkgs[name]})")
+ print()
-old_pkgs = get_packages(old_vm)
-new_pkgs = get_packages(new_vm)
-old_vm.kill()
-new_vm.kill()
+def main() -> None:
+ parser = argparse.ArgumentParser(description="Compare package versions on VM images")
+ parser.add_argument("old", help='the "old" image to compare')
+ parser.add_argument("new", help='the "new" image to compare')
+ args = parser.parse_args()
-print('Removed:')
-for name in sorted(set(old_pkgs) - set(new_pkgs)):
- print(f' {name} ({old_pkgs[name]})')
-print()
+ with testthing.cli_helper():
+ asyncio.run(diff(args.old, args.new))
-print('Added:')
-for name in sorted(set(new_pkgs) - set(old_pkgs)):
- print(f' {name} ({new_pkgs[name]})')
-print()
-print('Changed:')
-for name in sorted(set.intersection(set(new_pkgs), set(old_pkgs))):
- if new_pkgs[name] != old_pkgs[name]:
- print(f' {name} ({old_pkgs[name]} -> {new_pkgs[name]})')
-print()
+if __name__ == "__main__":
+ main()
diff --git a/images/scripts/arch.setup b/images/scripts/arch.setup
index 8b8813f911..987477d8ce 100755
--- a/images/scripts/arch.setup
+++ b/images/scripts/arch.setup
@@ -184,11 +184,6 @@ pacman -R --noconfirm cloud-init
# https://github.com/cockpit-project/bots/issues/3901#issuecomment-1260579703
systemctl disable systemd-time-wait-sync.service
-# https://gitlab.archlinux.org/archlinux/packaging/packages/openssh/-/issues/16
-# https://github.com/openssh/openssh-portable/pull/388
-# https://github.com/openssh/openssh-portable/pull/593
-echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts
-
# Reduce image size, clear package cache (/var/cache/pacman/pkg)
rm -f /var/cache/pacman/pkg/*
diff --git a/images/scripts/bootc.setup b/images/scripts/bootc.setup
index a551d8b016..b5f50f5149 100755
--- a/images/scripts/bootc.setup
+++ b/images/scripts/bootc.setup
@@ -27,20 +27,6 @@ podman rmi localhost:5000/bootc:latest localhost/bootc:latest
podman rm -f -t0 ostree-registry
rm /var/cache/bootc.oci.tar
-# Older RHEL releases need vsock support for test.thing
-if [ "${IMAGE#rhel-9}" != "$IMAGE" ] || [ "${IMAGE#centos-9}" != "$IMAGE" ]; then
- cp -va /var/lib/testvm/test.thing-workarounds/*.{socket,service} /etc/systemd/system
- systemctl daemon-reload
- systemctl enable tt-sshd-vsock.socket tt-sd_notify.service
- checkmodule -M -m -o /tmp/sshd_vsock.mod /var/lib/testvm/test.thing-workarounds/sshd_vsock.te
- semodule_package -m /tmp/sshd_vsock.mod -o /tmp/sshd_vsock.pp
- semodule -i /tmp/sshd_vsock.pp
-fi
-
-# https://github.com/openssh/openssh-portable/pull/388
-# https://github.com/openssh/openssh-portable/pull/593
-echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts
-
# disable various maintenance tasks which interfere with tests and don't make sense for our tests
systemctl disable bootc-fetch-apply-updates.timer fstrim.timer logrotate.timer raid-check.timer
diff --git a/images/scripts/lib/test.thing-workarounds/sshd_vsock.te b/images/scripts/lib/test.thing-workarounds/sshd_vsock.te
deleted file mode 100644
index d05cceab2b..0000000000
--- a/images/scripts/lib/test.thing-workarounds/sshd_vsock.te
+++ /dev/null
@@ -1,17 +0,0 @@
-# test.thing workaround for SELinux issues with vsock on older RHEL versions
-# https://issues.redhat.com/browse/RHEL-113647
-
-module sshd_vsock 1.0;
-
-require {
- type init_t;
- type sshd_t;
- type sshd_net_t;
- class vsock_socket {
- create bind listen accept getattr read write getopt setopt ioctl name_bind
- };
-}
-
-allow init_t sshd_t:vsock_socket { create bind listen accept getattr setopt name_bind };
-allow sshd_t self:vsock_socket { read write getattr getopt setopt ioctl accept };
-allow sshd_net_t sshd_t:vsock_socket { read write getattr };
diff --git a/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service b/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service
deleted file mode 100644
index cc7a4d9c84..0000000000
--- a/images/scripts/lib/test.thing-workarounds/tt-sd_notify.service
+++ /dev/null
@@ -1,19 +0,0 @@
-# test.thing polyfill for systems that lack support for vmm.notify_socket
-#
-# This works also on systemd versions that don't know about credentials from
-# SMBIOS OEM strings by using dmidecode directly. socat is also required.
-
-[Unit]
-After=sockets.target
-Wants=sockets.target
-
-[Service]
-Type=oneshot
-ExecStart=/bin/bash -ec '\
- if addr="$$(dmidecode -qt11 | grep -om1 "vsock-stream:2:[0-9]*")"; then \
- printf X_SYSTEMD_UNIT_ACTIVE=sockets.target | socat - "$${addr/stream/connect}"; \
- fi \
-'
-
-[Install]
-WantedBy=sockets.target
diff --git a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket b/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket
deleted file mode 100644
index 2923be62dd..0000000000
--- a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock.socket
+++ /dev/null
@@ -1,10 +0,0 @@
-# test.thing polyfill for systems that lack support for sshd-vsock.socket
-#
-# If you use this, you also need to use tt-sshd-vsock@.service
-
-[Socket]
-ListenStream=vsock::22
-Accept=yes
-
-[Install]
-WantedBy=sockets.target
diff --git a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service b/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service
deleted file mode 100644
index ba1a80c883..0000000000
--- a/images/scripts/lib/test.thing-workarounds/tt-sshd-vsock@.service
+++ /dev/null
@@ -1,13 +0,0 @@
-# test.thing polyfill for systems that lack support for sshd-vsock.socket
-#
-# If you use this, you also need to use tt-sshd-vsock.socket
-
-[Unit]
-Wants=sshd-keygen.target
-After=sshd-keygen.target
-
-[Service]
-EnvironmentFile=-/etc/sysconfig/sshd
-ExecStart=-/usr/sbin/sshd -i $OPTIONS -o "AuthorizedKeysFile ${CREDENTIALS_DIRECTORY}/ssh.ephemeral-authorized_keys-all .ssh/authorized_keys"
-StandardInput=socket
-LoadCredential=ssh.ephemeral-authorized_keys-all
diff --git a/images/scripts/opensuse-tumbleweed.setup b/images/scripts/opensuse-tumbleweed.setup
index 077f31fd56..0b75841cfd 100755
--- a/images/scripts/opensuse-tumbleweed.setup
+++ b/images/scripts/opensuse-tumbleweed.setup
@@ -171,10 +171,6 @@ echo 'GRUB_CMDLINE_LINUX_DEFAULT=""' >> /etc/default/grub
echo 'GRUB_CMDLINE_LINUX="no_timer_check net.ifnames=0 console=tty1 console=ttyS0,115200n8"' >> /etc/default/grub
grub2-mkconfig -o /boot/grub2/grub.cfg
-# https://github.com/openssh/openssh-portable/pull/388
-# https://github.com/openssh/openssh-portable/pull/593
-echo 100::55:4e4b:4e4f:574e UNKNOWN | tee -a /etc/hosts
-
# reduce image size
zypper clean
diff --git a/images/scripts/rhel.setup b/images/scripts/rhel.setup
index d4430d0361..8b34855343 100755
--- a/images/scripts/rhel.setup
+++ b/images/scripts/rhel.setup
@@ -291,16 +291,6 @@ if [ "${IMAGE#rhel-[89]}" != "$IMAGE" ] || [ "${IMAGE#centos-[89]}" != "$IMAGE"
TEST_PACKAGES="$TEST_PACKAGES python3-tracer"
fi
-# Older RHEL releases need vsock support for test.thing
-if [ "${IMAGE#rhel-[89]}" != "$IMAGE" ] || [ "${IMAGE#centos-9}" != "$IMAGE" ]; then
- cp -va /var/lib/testvm/test.thing-workarounds/*.{socket,service} /etc/systemd/system
- systemctl daemon-reload
- systemctl enable tt-sshd-vsock.socket tt-sd_notify.service
- checkmodule -M -m -o /tmp/sshd_vsock.mod /var/lib/testvm/test.thing-workarounds/sshd_vsock.te
- semodule_package -m /tmp/sshd_vsock.mod -o /tmp/sshd_vsock.pp
- semodule -i /tmp/sshd_vsock.pp
-fi
-
# These packages are downloaded to the image so that the tests can
# install them on-demand.
diff --git a/lib/testthing.py b/lib/testthing.py
new file mode 100755
index 0000000000..02a2ba6812
--- /dev/null
+++ b/lib/testthing.py
@@ -0,0 +1,1549 @@
+#!/usr/bin/python3
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+"""test.thing - A simple modern VM runner.
+
+https://codeberg.org/lis/test.thing
+
+A simple VM runner script exposing an API useful for use as a pytest fixture.
+Can also be used to run a VM and login via the console.
+
+Each VM is allocated an identifier: 'tt.0', 'tt.1', etc.
+
+For each VM, an ephemeral ssh key is created and used to connect to the VM via
+vsock with systemd-ssh-proxy, which works even if the guest doesn't have
+networking configured. The ephemeral key means that access is limited to the
+current user (since vsock connections are otherwise available to all users on
+the host system). The guest needs to have systemd 256 for this to work.
+
+An ssh control socket is created for sending commands and can also be used
+externally, avoiding the need to authenticate. A suggested ssh config:
+
+```
+Host tt.*
+ ControlPath ${XDG_RUNTIME_DIR}/test.thing/%h/ssh
+```
+
+And then you can say `ssh tt.0` or `scp file tt.0:/tmp`.
+"""
+
+# When copying test.thing into your own project, try to use a tagged version.
+# If you need to use a version between tags or have made your own
+# modifications, please make note if it by modifying the version number.
+__version__ = "0.4.0"
+
+import argparse
+import asyncio
+import contextlib
+import contextvars
+import ctypes
+import dataclasses
+import functools
+import itertools
+import json
+import logging
+import os
+import pathlib
+import re
+import shlex
+import shutil
+import signal
+import sys
+import tempfile
+import traceback
+import weakref
+from collections.abc import (
+ AsyncGenerator,
+ AsyncIterator,
+ Callable,
+ Coroutine,
+ Iterable,
+ Iterator,
+ Mapping,
+ Sequence,
+)
+from pathlib import Path
+from typing import Any, Literal, Never, Self
+from types import TracebackType
+
+logger = logging.getLogger(__name__)
+
+COCKPIT_TEST_IDENTITY = """
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpQIBAAKCAQEA1DrTSXQRF8isQQfPfK3U+eFC4zBrjur+Iy15kbHUYUeSHf5S
+jXPYbHYqD1lHj4GJajC9okle9rykKFYZMmJKXLI6987wZ8vfucXo9/kwS6BDAJto
+ZpZSj5sWCQ1PI0Ce8CbkazlTp5NIkjRfhXGP8mkNKMEhdNjaYceO49ilnNCIxhpb
+eH5dH5hybmQQNmnzf+CGCCLBFmc4g3sFbWhI1ldyJzES5ZX3ahjJZYRUfnndoUM/
+TzdkHGqZhL1EeFAsv5iV65HuYbchch4vBAn8jDMmHh8G1ixUCL3uAlosfarZLLyo
+3HrZ8U/llq7rXa93PXHyI/3NL/2YP3OMxE8baQIDAQABAoIBAQCxuOUwkKqzsQ9W
+kdTWArfj3RhnKigYEX9qM+2m7TT9lbKtvUiiPc2R3k4QdmIvsXlCXLigyzJkCsqp
+IJiPEbJV98bbuAan1Rlv92TFK36fBgC15G5D4kQXD/ce828/BSFT2C3WALamEPdn
+v8Xx+Ixjokcrxrdeoy4VTcjB0q21J4C2wKP1wEPeMJnuTcySiWQBdAECCbeZ4Vsj
+cmRdcvL6z8fedRPtDW7oec+IPkYoyXPktVt8WsQPYkwEVN4hZVBneJPCcuhikYkp
+T3WGmPV0MxhUvCZ6hSG8D2mscZXRq3itXVlKJsUWfIHaAIgGomWrPuqC23rOYCdT
+5oSZmTvFAoGBAPs1FbbxDDd1fx1hisfXHFasV/sycT6ggP/eUXpBYCqVdxPQvqcA
+ktplm5j04dnaQJdHZ8TPlwtL+xlWhmhFhlCFPtVpU1HzIBkp6DkSmmu0gvA/i07Z
+pzo5Z+HRZFzruTQx6NjDtvWwiXVLwmZn2oiLeM9xSqPu55OpITifEWNjAoGBANhH
+XwV6IvnbUWojs7uiSGsXuJOdB1YCJ+UF6xu8CqdbimaVakemVO02+cgbE6jzpUpo
+krbDKOle4fIbUYHPeyB0NMidpDxTAPCGmiJz7BCS1fCxkzRgC+TICjmk5zpaD2md
+HCrtzIeHNVpTE26BAjOIbo4QqOHBXk/WPen1iC3DAoGBALsD3DSj46puCMJA2ebI
+2EoWaDGUbgZny2GxiwrvHL7XIx1XbHg7zxhUSLBorrNW7nsxJ6m3ugUo/bjxV4LN
+L59Gc27ByMvbqmvRbRcAKIJCkrB1Pirnkr2f+xx8nLEotGqNNYIawlzKnqr6SbGf
+Y2wAGWKmPyEoPLMLWLYkhfdtAoGANsFa/Tf+wuMTqZuAVXCwhOxsfnKy+MNy9jiZ
+XVwuFlDGqVIKpjkmJyhT9KVmRM/qePwgqMSgBvVOnszrxcGRmpXRBzlh6yPYiQyK
+2U4f5dJG97j9W7U1TaaXcCCfqdZDMKnmB7hMn8NLbqK5uLBQrltMIgt1tjIOfofv
+BNx0raECgYEApAvjwDJ75otKz/mvL3rUf/SNpieODBOLHFQqJmF+4hrSOniHC5jf
+f5GS5IuYtBQ1gudBYlSs9fX6T39d2avPsZjfvvSbULXi3OlzWD8sbTtvQPuCaZGI
+Df9PUWMYZ3HRwwdsYovSOkT53fG6guy+vElUEDkrpZYczROZ6GUcx70=
+-----END RSA PRIVATE KEY-----
+"""
+"""A copy of `bots/machine/identity` from the cockpit project. Many existing
+VM images have the public half of this key inside of them, so it's useful for
+gaining access to those if they lack support for ephemeral ssh keys."""
+
+
+# This is basically tempfile.TemporaryDirectory but sequentially-allocated.
+# We do that so we can easily interact with the VMs from outside (with ssh).
+class IpcDirectory:
+ """A context manager for the VM IPC directory.
+
+ This is very similar to tempfile.TemporaryDirectory() except that the
+ allocation is predictable (sequential): the created directory will be
+ `/run/user/$uid/test.thing/tt.n` for the smallest `n` that we find.
+
+ It works the same way:
+
+ with IpcDirectory() as path:
+ ...use path...
+
+ The directory gets tagged with a `pid` file containing the pid and pidfd
+ inode of the current process. This could be helpful for pruning dead
+ directories, but is currently unused.
+ """
+
+ finalizer: Callable[[], None] | None = None
+
+ @staticmethod
+ def _find_dir() -> Path:
+ try:
+ xdg_rundir = Path(os.environ["XDG_RUNTIME_DIR"])
+ except KeyError:
+ # No XDG_RUNTIME_DIR? Somewhere in /tmp will have to do
+ return Path(tempfile.mkdtemp())
+
+ for n in range(10000):
+ tmpdir = xdg_rundir / "test.thing" / f"tt.{n}"
+
+ try:
+ tmpdir.mkdir(exist_ok=False, parents=True, mode=0o700)
+ except FileExistsError:
+ continue
+
+ return tmpdir
+
+ raise FileExistsError
+
+ def __enter__(self) -> Path:
+ """Create a unique directory.
+
+ This will sequentially allocate the first available 'tt.0', 'tt.1',
+ etc. directory and return it as a `Path`.
+ """
+ pid = os.getpid()
+ pidfd = os.pidfd_open(pid)
+ try:
+ buf = os.fstat(pidfd)
+ unique_id = f"{pid} {buf.st_ino}\n"
+ finally:
+ os.close(pidfd)
+
+ tmpdir = self._find_dir()
+ self.finalizer = weakref.finalize(self, shutil.rmtree, tmpdir)
+ (tmpdir / "pid").write_text(unique_id)
+ return tmpdir
+
+ def __exit__(self, *args: object) -> None:
+ """Delete the IPC directory and its contents."""
+ del args
+ if self.finalizer:
+ self.finalizer()
+
+
+def _normalize_args(
+ *args: str | pathlib.PurePath | tuple[str | pathlib.PurePath, ...],
+) -> Iterable[Iterable[str]]:
+ for chunk in args:
+ if not isinstance(chunk, tuple):
+ yield (str(chunk),)
+ elif len(chunk) != 0:
+ yield map(str, chunk)
+
+
+def _pretty_print_args(
+ *args: str | pathlib.PurePath | tuple[str | pathlib.PurePath, ...],
+) -> str:
+ """Pretty-print a nested argument list.
+
+ This takes the argument list format used by test.thing and turns it into a
+ format that looks like a nicer version of `set -x` from POSIX shell.
+ """
+ if not any(isinstance(arg, tuple) for arg in args):
+ # No tuples: use the boring format
+ return shlex.join(map(str, args))
+
+ # There are tuples: use the fancy format
+ return " \\\n ".join(map(shlex.join, _normalize_args(*args)))
+
+
+def _find_qemu() -> Path:
+ for candidate in ("qemu-kvm", "kvm"):
+ if cmd := shutil.which(candidate):
+ return Path(cmd)
+
+ raise FileNotFoundError("Unable to find qemu-kvm")
+
+
+def _find_ovmf() -> Path:
+ candidates = [
+ # path for Fedora/RHEL (our tasks container)
+ "/usr/share/OVMF/OVMF_CODE.fd",
+ # path for Ubuntu (GitHub Actions runners)
+ "/usr/share/ovmf/OVMF.fd",
+ # path for Arch
+ "/usr/share/edk2/x64/OVMF.4m.fd",
+ ]
+
+ for path in map(Path, candidates):
+ if path.exists():
+ return path
+
+ raise FileNotFoundError("Unable to find OVMF UEFI BIOS")
+
+
+async def _qmp_command(ipc: Path, command: str) -> object:
+ reader, writer = await asyncio.open_unix_connection(ipc / "qmp")
+
+ async def execute(command: str) -> object:
+ writer.write((json.dumps({"execute": command}) + "\n").encode())
+ await writer.drain()
+ while True:
+ response = json.loads(await reader.readline())
+ if "event" in response:
+ continue
+ if "return" in response:
+ return response["return"]
+ raise RuntimeError(f"Got error response from qmp: {response!r}")
+
+ # Trivial handshake (ignore them, send nothing)
+ _ = json.loads(await reader.readline())
+ await execute("qmp_capabilities")
+
+ response = await execute(command)
+
+ writer.close()
+ await writer.wait_closed()
+
+ return response
+
+
+def _ssh_direct_args(
+ identities: Sequence[Path], vsock: Path
+) -> tuple[tuple[str, str], ...]:
+ options = {
+ # Fake that we know the host key
+ "KnownHostsCommand": "/bin/echo %H %t %K",
+ # Use systemd-ssh-proxy to connect via vsock
+ "ProxyCommand": f"/usr/lib/systemd/systemd-ssh-proxy vsock-mux/{vsock} 22",
+ "ProxyUseFdpass": "yes",
+ # Try to prevent interactive prompting and/or updating known_hosts
+ # files or otherwise interacting with the environment
+ "BatchMode": "yes",
+ "IdentitiesOnly": "yes",
+ "PKCS11Provider": "none",
+ "PasswordAuthentication": "no",
+ "StrictHostKeyChecking": "yes",
+ "User": "root",
+ "UserKnownHostsFile": "/dev/null",
+ }
+
+ return (
+ ("-F", "none"), # don't use the user's config
+ *(("-o", f"{k}={v}") for k, v in options.items()),
+ *(("-i", f"{path}") for path in identities),
+ )
+
+
+@functools.cache
+def _stderr_is_tty() -> bool:
+ return os.isatty(2)
+
+
+class UI:
+ """A helper for printing messages and launching subprocesses."""
+
+ def __init__(self, *, status_messages: bool, verbose: bool) -> None:
+ """Create a UI helper.
+
+ This controls the stderr output of test.thing. The test.thing library
+ never writes to stdout.
+
+ - status_messages: if intermediary messages about the state of the
+ machine should be printed or not
+ - verbose: extra output is printed (like all executed commands)
+
+ If both are false then nothing will be printed.
+ """
+ self._status_messages = status_messages
+ self._verbose = verbose
+
+ def clear_status_message(self) -> None:
+ """Clear any displayed status message."""
+ if _stderr_is_tty():
+ sys.stderr.write("\r\033[2K")
+
+ def print_status(self, line: str) -> None:
+ """Print a status line message.
+
+ This is only printed if status_messages=True. A status message is a
+ transient message that will be erased at the next output (unless stderr
+ is not a TTY).
+ """
+ if self._status_messages:
+ if os.isatty(2):
+ sys.stderr.write("\r\033[2K " + line + "\r")
+ else:
+ sys.stderr.write(line + "\n")
+
+ def print_verbose(self, line: str) -> None:
+ """Print a verbose message, if verbose=True."""
+ if self._verbose:
+ self.clear_status_message()
+ sys.stderr.write(line + "\n")
+
+ def print(self, line: str) -> None:
+ """Print a message, unconditionally."""
+ self.clear_status_message()
+ sys.stderr.write(line + "\n")
+
+ async def _wait_stdin(self, msg: str) -> None:
+ r"""Wait until stdin sees \n or EOF.
+
+ This prints the given message to stdout without adding an extra newline.
+
+ The input is consumed (and discarded) up to the \n and maybe more...
+ """
+ done = asyncio.Event()
+
+ def stdin_ready() -> None:
+ data = os.read(0, 4096)
+ if not data:
+ sys.stdout.write("\n")
+ if not data or b"\n" in data:
+ done.set()
+
+ loop = asyncio.get_running_loop()
+ loop.add_reader(0, stdin_ready)
+ sys.stdout.write(msg)
+ sys.stdout.flush()
+ try:
+ await done.wait()
+ finally:
+ loop.remove_reader(0)
+
+ async def sit(
+ self,
+ msg: str | None = None,
+ vm_id: str | None = None,
+ exc: BaseException | None = None,
+ ) -> None:
+ """Wait for the user to press Enter."""
+ # lis: the ages old design question: does the button show
+ # the current state or the future one when you press it :)
+ # pitti: that's exactly my question. by taking the one with
+ # both then i don't have to choose! :D
+ # although to be honest, i find your argument convincing.
+ # i'll put the ⏸️ back
+ # lis: it was more of a joke -- I actually agree that a
+ # play/pause button is nicer
+ # too late lol
+
+ self.clear_status_message()
+
+ if exc is not None:
+ self.print(f"\n🤦 {''.join(traceback.format_exception(exc))}")
+ if msg is not None:
+ self.print(f"\n{msg}")
+ if vm_id is not None:
+ self.print(
+ f"Guest is still running. Connect with: \033[1mssh {vm_id}\033[0m"
+ )
+
+ await self._wait_stdin("\nEnter or EOF to exit ⏸️ ")
+
+ async def spawn(
+ self,
+ *args: str | Path | tuple[str | Path, ...],
+ stdin: int | None = asyncio.subprocess.DEVNULL,
+ stdout: int | None = None,
+ stderr: int | None = None,
+ ) -> asyncio.subprocess.Process:
+ """Spawn a process.
+
+ This has a couple of extra niceties: the args list is flattened, Path is
+ converted to str, the spawned process is logged to stderr for debugging,
+ and we call PR_SET_PDEATHSIG with SIGTERM after forking to make sure the
+ process exits with us.
+
+ The flattening allows grouping logically-connected arguments together,
+ producing nicer verbose output, allowing for adding groups of arguments
+ from helper functions or comprehensions, and works nicely with code
+ formatters:
+
+ For example:
+
+ private = Path(...)
+ options = { ... }
+
+ ssh = await spawn(
+ "ssh",
+ ("-i", private),
+ *(("-o", f"{k} {v}") for k, v in options.items()),
+ ("-l", "root", "x"),
+ ...
+ )
+
+ The type of the groups is `tuple`. It could be `Sequence` but this would
+ also allow using bare strings, which would be split into their individual
+ characters. Using `tuple` prevents this from happening.
+ """
+ # This might be complicated: do it before the fork
+ prctl = ctypes.CDLL(None, use_errno=True).prctl
+
+ def pr_set_pdeathsig() -> None:
+ PR_SET_PDEATHSIG = 1 # noqa: N806
+ if prctl(PR_SET_PDEATHSIG, signal.SIGTERM):
+ os._exit(1) # should never happen
+
+ self.print_verbose(f"+ {_pretty_print_args(*args)}\n")
+
+ return await asyncio.subprocess.create_subprocess_exec(
+ *itertools.chain(*_normalize_args(*args)),
+ stdin=stdin,
+ stdout=stdout,
+ stderr=stderr,
+ preexec_fn=pr_set_pdeathsig,
+ start_new_session=True,
+ )
+
+ async def run(
+ self,
+ *args: str | Path | tuple[str | Path, ...],
+ stdin: int | None = asyncio.subprocess.DEVNULL,
+ check: bool = True,
+ ) -> int:
+ """Run a process, waiting for it to exit.
+
+ This takes the same arguments as spawn, plus a "check" argument (True by
+ default) which works in the usual way.
+ """
+ process = await self.spawn(*args, stdin=stdin)
+ returncode = await process.wait()
+ if check and returncode != 0:
+ raise SubprocessError(args, returncode=returncode)
+ return returncode
+
+
+class GuestPath(pathlib.PurePosixPath):
+ """A path on the virtual machine guest.
+
+ This aims to support similar operations to pathlib.Path (with similar
+ APIs), but most operations are async and many have slightly different
+ feature sets.
+ """
+
+ __slots__ = ("_vm",)
+
+ def __init__(self, *args: str | os.PathLike[str], vm: "VirtualMachine") -> None:
+ """Create a GuestPath for a path on a guest."""
+ super().__init__(*args)
+ self._vm = vm
+
+ def with_segments(self, *pathsegments: str | os.PathLike[str]) -> Self:
+ """Create a new path by combining the given pathsegments."""
+ return type(self)(*pathsegments, vm=self._vm)
+
+ async def mkdir(self, *, mode: int | None = None, parents: bool = False) -> None:
+ """Create a directory."""
+ await self._vm.execute(
+ "mkdir",
+ "-p" if parents else (),
+ ("-m", f"{mode:0o}") if mode is not None else (),
+ self,
+ )
+
+ async def chmod(self, mode: int | str, *, follow_symlinks: bool = True) -> None:
+ """Change a file mode."""
+ await self._vm.execute(
+ "chmod",
+ "-h" if not follow_symlinks else (),
+ f"{mode:0o}" if isinstance(mode, int) else mode,
+ self,
+ )
+
+ async def chown(
+ self,
+ owner: str | tuple[str | int | None, str | int | None],
+ *,
+ follow_symlinks: bool = True,
+ ) -> None:
+ """Change the owner of a file.
+
+ The owner can be a string like 'user:group' or a pair of (user, group)
+ where each can be a string, int, or None (to make no change).
+ """
+ if isinstance(owner, tuple):
+ user, group = owner
+ owner = f"{user or ''}:{group or ''}"
+
+ await self._vm.execute(
+ "chown", "-h" if not follow_symlinks else (), owner, self
+ )
+
+ async def write_bytes(self, data: bytes, *, append: bool = False) -> None:
+ """Write or append to to a binary file."""
+ await self._vm.execute(
+ "dd",
+ "status=none",
+ ("conv=notrunc", "oflag=append") if append else (),
+ f"of={self}",
+ input=data,
+ )
+
+ async def read_text(self) -> str:
+ """Read a text file."""
+ return await self._vm.execute("cat", self)
+
+ async def write_text(self, data: str, *, append: bool = False) -> None:
+ """Write or append to a text file."""
+ await self.write_bytes(data.encode(), append=append)
+
+ async def unlink(
+ self, *, missing_ok: bool = False, recursive: bool = False
+ ) -> None:
+ """Unlink the given file."""
+ await self._vm.execute(
+ "rm", "-f" if missing_ok else (), "-r" if recursive else (), self
+ )
+
+ async def rmdir(self) -> None:
+ """Remove a directory."""
+ await self._vm.execute("rmdir", self)
+
+
+@dataclasses.dataclass
+class Network:
+ """A virtual machine network."""
+
+ id: int | Literal["user"]
+ """The network identifier. If this is an integer then it specifies a
+ multicast network on which other virtual machines can communicate. If it's
+ the literal value "user" then this sets up usermode networking."""
+
+ @classmethod
+ def user(cls) -> Self:
+ """Create a user-mode network."""
+ return cls(id="user")
+
+ @classmethod
+ def multicast(cls, netnr: int) -> Self:
+ """Create a multicast network for talking to other machines."""
+ return cls(id=netnr)
+
+ def to_qemu(self) -> str:
+ """Describe the network in a way that qemu understands."""
+ if self.id == "user":
+ return "user"
+
+ # Same as Cockpit
+ return f"socket,mcast=230.0.0.1:{self.id},localaddr=127.0.0.1"
+
+
+class _ServiceGroup(asyncio.TaskGroup):
+ """A special kind of TaskGroup to support 'background services'.
+
+ Tasks normally run 'forever', and end only via cancellation. There's also a
+ "ready" notification mechanism via contextvars, which allows combining
+ "setup" and "running" phases into a single task, allowing parallel startup.
+ """
+
+ ready_var = contextvars.ContextVar[asyncio.Event]("Service task ready event")
+
+ def __init__(self) -> None:
+ super().__init__()
+ self.__ready: list[asyncio.Event] = []
+ self.__tasks: list[asyncio.Task[None]] = []
+
+ def add_service(self, coro: Coroutine[None, None, None]) -> None:
+ event = asyncio.Event()
+ context = contextvars.copy_context()
+ context.run(self.ready_var.set, event)
+ self.__tasks.append(self.create_task(coro, context=context))
+ self.__ready.append(event)
+
+ @classmethod
+ def notify_ready(cls) -> None:
+ cls.ready_var.get().set()
+
+ async def wait_all_ready(self) -> None:
+ for event in self.__ready:
+ await event.wait()
+
+ async def cancel_all(self) -> None:
+ for task in self.__tasks:
+ task.cancel() # already checks task.done()
+
+ async def __aexit__(
+ self,
+ et: type[BaseException] | None,
+ exc: BaseException | None,
+ tb: TracebackType | None,
+ ) -> None:
+ try:
+ return await super().__aexit__(et, exc, tb)
+ except BaseExceptionGroup as eg:
+ if len(eg.exceptions) == 1:
+ raise eg.exceptions[0]
+ raise
+
+class VirtualMachine(contextlib.AsyncExitStack):
+ """A handle to a running virtual machine.
+
+ This is meant to be used as an async context manager like so:
+
+ with IpcDirectory() as ipc:
+ image = Path("...")
+ async with VirtualMachine(image, ipc=ipc) as vm:
+ await vm.execute("cat", "/usr/lib/os-release")
+
+ The user of the context manager runs in context of an asyncio.TaskGroup and
+ will be cancelled if anything unexpected happens (ssh connection lost, VM
+ exiting, etc).
+
+ When the context manager is exited the machine is taken down.
+
+ When the machine is running it is also possible to access it from outside.
+ See the documentation for the module.
+ """
+
+ ssh_args: tuple[str | Path | tuple[str | Path, ...], ...]
+ """ssh command-line arguments for executing commands"""
+
+ ssh_direct_args: tuple[str | Path | tuple[str | Path, ...], ...] | None
+ """ssh command-line arguments to be used to connect directly to the vsock"""
+
+ journal: list[dict[str, str]]
+ """A list of journal entries from the guest.
+
+ Each entry is a dictionary. Multiple values per key are not supported, but
+ binary data is: the strings are encoded with errors='surrogateescape' so
+ it's possible to get the original binary back if that's what you're
+ expecting.
+
+ See the journal= kwarg to VirtualMachine().
+ """
+
+ _ssh_control_task: asyncio.Task[None] | None = None
+
+ def __init__(
+ self,
+ image: Path | str,
+ *,
+ ipc: Path,
+ attach_console: bool = False,
+ boot: Literal["efi", "mbr"] = "efi",
+ cloud_init_user_data: Mapping[str, object] | None = None,
+ credentials: Mapping[str, str] = {},
+ cpus: int = 4,
+ identity: tuple[Path, str | None] | None = None,
+ identities: Sequence[str | Path] = (),
+ journal: bool | Callable[[dict[str, str]], bool | None] = False,
+ memory: int | str = "4G",
+ networks: Sequence[Network] = (),
+ provision_ssh_key: bool = False,
+ sit: bool = False,
+ snapshot: bool = True,
+ status_messages: bool = False,
+ target: str = "sockets.target",
+ timeout: float = 30.0,
+ ui: UI | None = None,
+ verbose: bool = False,
+ ) -> None:
+ """Construct a VM.
+
+ The kwargs allow customizing the behaviour:
+ - attach_console: if qemu should connect the console to stdio
+ - boot: if we should boot with EFI or via the MBR
+ - cloud_init_user_data: JSON user-data for cloud-init
+ - credentials: extra system credentials
+ - cpus: the number of CPUs
+ - identity: a path to an ssh private key and the public key as a string.
+ If the public key is specified as None then it won't be configured on
+ the guest. The default (None) is to generate an ephemeral keypair.
+ - identities: extra private keys (either as paths on the disk or
+ directly as strings) to pass to ssh. Useful if you're not sure
+ which key the image will accept and want to try multiple.
+ - journal: False (default) to disable journal handling, True to
+ record all entries, and a callable to decide on a per-entry basis.
+ - memory: how much memory the guest gets in MiB, or a string like "4G"
+ - networks: a list of Network objects (or empty to disable networking)
+ - provision_ssh_key: if we should attempt to install the public side
+ of the ssh key as ~root/.ssh/authorized_keys in the guest
+ - sit: if we should "sit" when an exception occurs: print the exception
+ and wait for input (to allow inspecting the running VM)
+ - snapshot: if the 'snapshot' option is used on the disk (changes are
+ transient)
+ - status_messages: if we should do output of status messages (stderr)
+ - target: the name of the systemd target to wait for
+ - timeout: how long to wait for the VM to start, or 'inf'
+ - ui: a custom instance of UI (otherwise a new one is constructed per
+ status_messages= and verbose=)
+ - verbose: if we should do output of verbose messages (stderr)
+ """
+ super().__init__()
+
+ self.image = image
+ self._ipc = ipc
+ self._attach_console = attach_console
+ self._boot = boot
+ self._cloud_init_user_data = cloud_init_user_data
+ self._cpus = cpus
+ self._credentials = credentials
+ self._identity = identity
+ self._identities = identities
+ self._journal = journal
+ self._memory = memory
+ self._networks = networks
+ self._provision_ssh_key = provision_ssh_key
+ self._sit = sit
+ self._snapshot = snapshot
+ self._target = target
+ self._timeout = timeout
+ self._ui = ui or UI(status_messages=status_messages, verbose=verbose)
+
+ self._tasks = _ServiceGroup()
+ self._ssh_control_ready = asyncio.Event()
+ self._qemu_exited = asyncio.Event()
+ self._shutdown_ok = False
+
+ self.root = GuestPath("/", vm=self)
+ self.home = GuestPath(".", vm=self)
+ self.journal = []
+
+ async def _run_helper(
+ self, *args: str | Path | tuple[str | Path, ...], ready_when: Path
+ ) -> None:
+ """Run a helper process in the background.
+
+ This is designed to be used from _ServiceGroup and will notify when the
+ process is properly running. On cancellation, the process is
+ terminated.
+ """
+ process = await self._ui.spawn(*args)
+
+ # block until we see the expected socket or an unexpected exit
+ while not ready_when.exists():
+ with contextlib.suppress(TimeoutError):
+ returncode = await asyncio.wait_for(process.wait(), 0.01)
+ raise SubprocessError(args, returncode=returncode)
+
+ _ServiceGroup.notify_ready()
+
+ try:
+ returncode = await process.wait() # this should never return
+ raise SubprocessError(args, returncode=returncode)
+ except asyncio.CancelledError:
+ with contextlib.suppress(ProcessLookupError):
+ process.kill()
+ await process.wait()
+
+ async def _ssh_keygen_service(self) -> None:
+ """Create the ephemeral ssh key.
+
+ This is designed to be used from _ServiceGroup and will notify when the
+ key has been generated and stored on self._identity.
+ """
+ assert self._identity is None
+ private_key = self._ipc / "id"
+
+ await self._ui.run(
+ "ssh-keygen",
+ "-q", # quiet
+ ("-t", "ed25519"),
+ ("-N", ""), # no passphrase
+ ("-C", ""), # no comment
+ ("-f", f"{private_key}"),
+ )
+
+ self._identity = private_key, (self._ipc / "id.pub").read_text().strip()
+ _ServiceGroup.notify_ready()
+
+ def _sd_notify(self, line: str) -> None:
+ logger.debug("sd_notify:%s", line)
+
+ # Only print target updates when ssh is offline
+ if self._ssh_control_task is not None:
+ return
+
+ key, _, value = line.partition("=")
+ if key == "X_SYSTEMD_UNIT_ACTIVE":
+ self._ui.print_status(f"Reached target: {value}")
+ if value == self._target:
+ self._ssh_control_task = self._tasks.create_task(self._ssh_control())
+
+ elif key == "X_SYSTEMD_UNIT_INACTIVE":
+ self._ui.print_status(f"Unit inactive: {value}")
+ elif key == "X_SYSTEMD_SHUTDOWN":
+ self._ui.print_status(f"Shutdown: {value}")
+
+ async def _sd_notify_service(self, path: Path) -> None:
+ async def connection(
+ reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+ ) -> None:
+ logger.debug("sd_notify connection")
+ try:
+ # Actually we should read until EOF but see
+ # https://github.com/rust-vmm/vhost-device/issues/874
+ message = await reader.read(65536)
+ self._sd_notify(message.decode())
+ finally:
+ writer.close()
+ await writer.wait_closed()
+
+ async with await asyncio.start_unix_server(connection, path) as srv:
+ _ServiceGroup.notify_ready()
+ await srv.serve_forever()
+
+ async def _journal_service(self, path: Path) -> None:
+ intern_table: dict[str, str] = {}
+
+ def intern(bval: bytes) -> str:
+ # The spec says that binary is "rare", so let's do strings, but use
+ # surrogateescape to leave the door open to having a way back.
+ sval = bval.decode(errors="surrogateescape")
+ return intern_table.setdefault(sval, sval)
+
+ async def connection(
+ reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+ ) -> None:
+ assert self._journal is not False
+
+ print("JOURNAL CON")
+ print("JOURNAL CON")
+ print("JOURNAL CON")
+
+ # https://systemd.io/JOURNAL_EXPORT_FORMATS/#journal-export-format
+ try:
+ entry: dict[str, str] = {}
+
+ while line := await reader.readline():
+ if line := line.rstrip(b"\n"):
+ key, eq, value = line.partition(b"=")
+
+ if not eq: # no equal sign?
+ # 64bit le size, followed by data, followed by newline
+ size = await reader.readexactly(8)
+ value = await reader.readexactly(
+ int.from_bytes(size, "little")
+ )
+ _ = await reader.readexactly(1)
+
+ # NB: no multiple-entries supported :(
+ entry[intern(key)] = intern(value)
+
+ else:
+ if self._journal is True or self._journal(entry):
+ self.journal.append(entry)
+ entry = {}
+ finally:
+ writer.close()
+
+ async with await asyncio.start_unix_server(connection, path) as srv:
+ _ServiceGroup.notify_ready()
+ await srv.serve_forever()
+
+ async def _qemu(self, creds: Mapping[str, str]) -> None:
+ snap = "on" if self._snapshot else "off"
+ drives = [f"file={self.image},format=qcow2,discard=unmap,snapshot={snap}"]
+
+ if self._cloud_init_user_data:
+ cloud_init = self._ipc / "cloud-init"
+ cloud_init.mkdir()
+ (cloud_init / "meta-data").touch()
+ (cloud_init / "user-data").write_text(
+ "#cloud-config\n" + json.dumps(self._cloud_init_user_data) + "\n"
+ )
+ drives.append(f"driver=vvfat,dir={cloud_init},readonly=on,label=CIDATA")
+
+ args = (
+ _find_qemu(),
+ "-nodefaults",
+ ("-object", f"memory-backend-memfd,share=on,id=mem0,size={self._memory}"),
+ ("-bios", _find_ovmf()) if self._boot == "efi" else (),
+ ("-boot", "menu=on"),
+ ("-machine", "q35,accel=kvm,memory-backend=mem0"),
+ ("-cpu", "host"),
+ ("-smp", f"{self._cpus}"),
+ ("-m", f"{self._memory}"),
+ ("-display", "none"),
+ ("-qmp", f"unix:{self._ipc}/qmp,server,wait=off"),
+ ("-chardev", f"socket,id=vsock,reconnect=0,path={self._ipc}/vsock-device"),
+ ("-device", "vhost-user-vsock-pci,chardev=vsock"),
+ # Console stuff...
+ ("-device", "virtio-serial-pci"),
+ (
+ "-chardev",
+ f"socket,path={self._ipc}/vsock_1111,id=tt-notify,reconnect-ms=1",
+ ),
+ ("-device", "virtserialport,chardev=tt-notify,name=tt-notify"),
+ ("-serial", "chardev:console"),
+ *(
+ (
+ ("-chardev", "stdio,mux=on,signal=off,id=console"),
+ ("-mon", "chardev=console,mode=readline"),
+ )
+ if self._attach_console
+ else (
+ # In the cases that the console isn't directed to stdio
+ # then we write it to a log file instead. Unfortunately,
+ # we also get a getty in our log file:
+ # https://github.com/systemd/systemd/issues/37928
+ ("-chardev", f"file,path={self._ipc}/console,id=console"),
+ )
+ ),
+ *(("-drive", f"{drive},if=virtio,media=disk") for drive in drives),
+ *(
+ ("-nic", net.to_qemu() + ",model=virtio-net-pci")
+ for net in self._networks
+ ),
+ # Credentials
+ *(
+ ("-smbios", f"type=11,value=io.systemd.credential:{k}={v}")
+ for k, v in creds.items()
+ ),
+ )
+
+ qemu = None
+ try:
+ self._ui.print_status("Waiting for guest")
+ qemu = await self._ui.spawn(*args, stdin=None)
+ returncode = await qemu.wait()
+ if not self._shutdown_ok:
+ raise SubprocessError(args, returncode)
+ except asyncio.CancelledError:
+ logger.debug("qemu task cancelled")
+ if qemu is not None:
+ logger.debug("Terminating qemu")
+ qemu.terminate()
+ try:
+ logger.debug("Waiting for qemu to quit")
+ await asyncio.shield(asyncio.wait_for(qemu.wait(), 5))
+ except TimeoutError:
+ logger.debug("Timed out -- killing qemu")
+ qemu.kill()
+ await asyncio.shield(qemu.wait())
+ finally:
+ logger.debug("qemu exited")
+ self._qemu_exited.set()
+
+ async def _ssh_control(self) -> None:
+ ssh = None
+ try:
+ assert self.ssh_direct_args is not None
+
+ self._ui.print_status("ssh control socket: connecting via vsock")
+
+ control_socket = self._ipc / "ssh"
+
+ args = (
+ "ssh",
+ *self.ssh_direct_args,
+ ("-N", "-n"), # no command, stdin disconnected
+ ("-M", "-S", control_socket), # listen on the control socket
+ self.get_id(), # unused, but shows up in messages
+ )
+ ssh = await self._ui.spawn(
+ *args,
+ stdin=asyncio.subprocess.DEVNULL,
+ stdout=asyncio.subprocess.PIPE,
+ )
+
+ # ssh sends EOF after the connection succeeds
+ assert ssh.stdout
+ await ssh.stdout.read()
+
+ # ..but that might have been because of an error, so check if the
+ # control socket actually exists
+ if not control_socket.exists():
+ raise SubprocessError(args, await ssh.wait())
+
+ # we're online!
+ self._ssh_control_ready.set()
+ self._ui.print_status("ssh control socket: connected.")
+
+ returncode = await ssh.wait()
+ if not self._shutdown_ok:
+ raise SubprocessError(args, returncode)
+ except asyncio.CancelledError:
+ if ssh is not None:
+ ssh.terminate()
+ await asyncio.shield(ssh.wait())
+ finally:
+ # We try to reset our state best as possible here to deal with
+ # reboots in the shutdown_ok case: we want the control socket
+ # reestablished when the machine comes back.
+ self._ssh_control_ready.clear()
+ self._ssh_control_task = None
+
+ def _get_console_log(self) -> Sequence[str]:
+ try:
+ log = (self._ipc / "console").read_text(errors="replace")
+ except FileNotFoundError:
+ log = ""
+
+ # Remove ANSI escapes, control characters, extra newlines
+ log_lines = re.sub(
+ r"\x1b\[[ -?]*[@-~]|" # CSI: ESC [ + params/interms + final
+ r"\x1b\][^\a\x1b]*|" # OSC: ESC ] + everything to \a or ESC
+ r"\x1b[ ()O].|\x1b.|" # two- and one-character escapes
+ r"[\x00-\b\v-\x1f\x7f]|" # all control chars but [\t\n]
+ r"(<=\n)\n", # extra newlines
+ "",
+ log,
+ ).splitlines()
+
+ return ["Console log:" if log_lines else "Console log unavailable.", *log_lines]
+
+ async def _start_services(
+ self, services: _ServiceGroup, creds: dict[str, str]
+ ) -> None:
+ if self._identity is None:
+ services.add_service(self._ssh_keygen_service())
+
+ services.add_service(self._sd_notify_service(self._ipc / "vsock_1111"))
+ creds["vmm.notify_socket"] = "vsock-stream:2:1111"
+
+ if self._journal:
+ services.add_service(self._journal_service(self._ipc / "vsock_1112"))
+ creds["journal.forward_to_socket"] = "vsock-stream:2:1112"
+
+ services.add_service(
+ self._run_helper(
+ "vhost-device-vsock",
+ ("--socket", self._ipc / "vsock-device"),
+ ("--uds-path", self._ipc / "vsock"),
+ ready_when=self._ipc / "vsock-device",
+ )
+ )
+
+ def _setup_identities(self, creds: dict[str, str]) -> Iterable[Path]:
+ assert self._identity is not None
+ private, public = self._identity
+ yield private
+
+ if public is not None:
+ creds["ssh.ephemeral-authorized_keys-all"] = public
+ if self._provision_ssh_key:
+ creds["ssh.authorized_keys.root"] = public
+
+ for nr, identity in enumerate(self._identities):
+ if isinstance(identity, str):
+ path = self._ipc / f"id.{nr}"
+ with path.open("x") as extra:
+ extra.write(identity)
+ path.chmod(0o400)
+ yield path
+ else:
+ yield identity
+
+ @contextlib.asynccontextmanager
+ async def _run(self) -> AsyncIterator[Self]:
+ # It goes like this:
+ # - we start listening on the sd-notify socket
+ # - we start qemu
+ # - at some point the guest will notify that ssh is ready
+ # - this causes us to spawn the ssh control task
+ # - once connected, _ssh_control_ready gets set
+ # - we wait for that, so when it's done, we're done
+ creds = {**self._credentials}
+
+ self.ssh_args = (
+ ("-F", "none"), # don't use the user's config
+ ("-o", f"ControlPath={self._ipc}/ssh"), # connect via the control socket
+ )
+
+ async with _ServiceGroup() as services:
+ # Start all of the background services in parallel
+ await self._start_services(services, creds)
+
+ # ...and wait for them to be ready
+ await services.wait_all_ready()
+
+ # we should have our ssh key by now, so deal with that
+ identities = tuple(self._setup_identities(creds))
+ self.ssh_direct_args = _ssh_direct_args(identities, self._ipc / "vsock")
+
+ async with self._tasks:
+ # start QEMU
+ self._tasks.create_task(self._qemu(creds))
+
+ # the notify socket server will create the ssh control socket task
+ # which, in turn, sets this ready once it's online.
+ try:
+ await asyncio.wait_for(
+ self._ssh_control_ready.wait(), self._timeout
+ )
+ except TimeoutError as exc:
+ lines = [
+ "Timed out waiting for the VM to start"
+ f" (after {self._timeout}s).",
+ *self._get_console_log(),
+ ]
+ raise TimeoutError("\n".join(lines) + "\n") from exc
+
+ # we're online
+ try:
+ yield self
+ except asyncio.CancelledError:
+ raise
+ except Exception as exc:
+ if self._sit:
+ await self._ui.sit(vm_id=self.get_id(), exc=exc)
+ raise
+
+ # time to shutdown.
+ self._shutdown_ok = True
+
+ # start with the control channel.
+ if self._ssh_control_task is not None:
+ task = self._ssh_control_task
+ task.cancel()
+ await task # Cancellation is async, so wait
+
+ # next, qemu: if we're in snapshot mode then we don't have to do a
+ # clean shutdown
+ with contextlib.suppress(FileNotFoundError):
+ await self.qmp("quit" if self._snapshot else "system_powerdown")
+
+ # finally, all background services
+ await services.cancel_all()
+
+ async def __aenter__(self) -> Self:
+ """Start the virtual machine."""
+ await super().__aenter__()
+ return await self.enter_async_context(self._run())
+
+ def get_id(self) -> str:
+ """Get the machine identifier like `tt.0`, `tt.1`, etc."""
+ return self._ipc.name
+
+ async def wait_exit(self) -> None:
+ """Wait for the VM to exit."""
+ self._shutdown_ok = True
+ await self._qemu_exited.wait()
+
+ async def _ssh_cmd(self, *args: tuple[str | Path, ...]) -> None:
+ await self._ui.run("ssh", *self.ssh_args, *args, self.get_id())
+
+ async def forward_port(self, *args: tuple[str, ...]) -> None:
+ """Set up a port forward.
+
+ The `spec` is the format used by `ssh -L`, and looks something like
+ `2222:127.0.0.1:22`.
+ """
+ return await self._ssh_cmd(("-O", "forward"), *args)
+
+ async def cancel_port(self, *args: tuple[str, ...]) -> None:
+ """Cancel a previous forward."""
+ return await self._ssh_cmd(("-O", "cancel"), *args)
+
+ async def wait_boot(self) -> None:
+ """Wait for the machine to be fully-booted."""
+ await self.execute("systemctl", "is-system-running", "--wait")
+
+ async def execute(
+ self,
+ cmd: str,
+ *args: str | GuestPath | tuple[str | GuestPath, ...],
+ check: bool = True,
+ direct: bool = False,
+ input: bytes | str | None = b"", # noqa:A002 # shadows `input()` but so does subprocess module
+ environment: Mapping[str, str] = {},
+ stdin: int | None = asyncio.subprocess.PIPE,
+ stdout: int | None = asyncio.subprocess.PIPE,
+ ) -> str:
+ """Execute a command on the guest.
+
+ If a single argument is given, it is expected to be a valid shell
+ script. If multiple arguments are given, they will interpreted as an
+ argument vector and will be properly quoted before being sent to the guest.
+ """
+ if args:
+ cmd = shlex.join(itertools.chain(*_normalize_args(cmd, *args)))
+
+ assert self.ssh_direct_args is not None
+ full_command = (
+ "ssh",
+ *(self.ssh_direct_args if direct else self.ssh_args),
+ self.get_id(), # unused, but shows up in messages
+ ("--", "set -eu;"),
+ *(f"export {k}={v and shlex.quote(v)};" for k, v in environment.items()),
+ cmd,
+ )
+
+ ssh = await self._ui.spawn(*full_command, stdin=stdin, stdout=stdout)
+ input_bytes = input.encode() if isinstance(input, str) else input
+ output, _ = await ssh.communicate(input_bytes)
+ returncode = await ssh.wait()
+ if check and returncode != 0:
+ raise SubprocessError(full_command, returncode, output)
+ return output.decode() if output is not None else ""
+
+ async def write(
+ self,
+ dest: str | GuestPath,
+ content: str | bytes,
+ *,
+ mkdir: bool = True,
+ owner: str | tuple[str | int | None, str | int | None] | None = None,
+ perm: str | int | None = None,
+ ) -> None:
+ """Write a file into the test machine.
+
+ Arguments:
+ dest: The file name in the machine to write to
+ content: Raw data to write to file
+ append: If True, append to existing file instead of replacing it
+ mkdir: if the parent directory should be created
+ owner: If set, call chown on the file with the given owner string
+ perm: Optional file permission as chmod shell string or integer
+
+ """
+ dest = GuestPath(dest, vm=self)
+
+ if mkdir:
+ await dest.parent.mkdir(parents=True)
+
+ if isinstance(content, str):
+ await dest.write_text(content)
+ else:
+ await dest.write_bytes(content)
+
+ if owner is not None:
+ await dest.chown(owner)
+
+ if perm:
+ await dest.chmod(perm)
+
+ async def scp(self, *args: str | pathlib.Path, direct: bool = False) -> None:
+ """Do a file transfer with scp.
+
+ The hostname is ignored, so for paths on the guest use something like
+ `vm:/tmp/path`.
+
+ All arguments that are given in the form of `pathlib.Path` are passed to
+ `scp` in absolute form, avoiding worries about `:` characters, but also
+ making it impossible to use certain scp features (such as the special
+ treatment of `.` — which pathlib collapses anyway). Use string form if
+ you need this, and worry about the escaping yourself.
+ """
+ assert self.ssh_direct_args is not None
+ await self._ui.run(
+ "scp",
+ *(self.ssh_direct_args if direct else self.ssh_args),
+ tuple(p.absolute() if isinstance(p, pathlib.Path) else p for p in args),
+ )
+
+ async def upload(
+ self, *args: str | pathlib.Path, target_directory: str | GuestPath | None = None
+ ) -> None:
+ """Upload files to the guest.
+
+ This works similarly to `cp --target-directory` (`-t`).
+
+ All arguments are interpreted as local paths. The target directory is
+ a directory on the remote system (defaulting to root's home directory)
+ which will be created (`mkdir -p`) if it doesn't exist. Each argument
+ is recursively copied into that directory by its basename.
+ """
+ if target_directory is not None:
+ await self.execute("mkdir", "-p", target_directory)
+ await self.scp(
+ "-r", *map(pathlib.Path, args), f"{self.get_id()}:{target_directory or ''}"
+ )
+
+ @contextlib.asynccontextmanager
+ async def disconnected(self) -> AsyncGenerator[None]:
+ """Temporarily disconnect the control socket.
+
+ On enter, disconnect the ssh control socket from the guest system.
+ Inside of the block it's possible to perform commands that would
+ otherwise result in the control socket being destroyed (which would be
+ a hard error).
+
+ On exit from the block, the connection is reestablished.
+
+ The most obvious use-case for this is rebooting.
+ """
+ assert self._ssh_control_task is not None
+ assert self._ssh_control_ready.is_set()
+ self._ssh_control_ready.clear()
+ self._ssh_control_task.cancel()
+ self._ssh_control_task = None
+
+ try:
+ yield
+ finally:
+ await self._ssh_control_ready.wait()
+
+ async def reboot(self) -> None:
+ """Reboot the guest, waiting until it's back online."""
+ async with self.disconnected():
+ await self.qmp("system_reset")
+
+ async def qmp(self, command: str) -> object:
+ """Send a QMP command to the hypervisor.
+
+ This can be used for things like modifying the hardware configuration.
+ Don't power it off this way: the correct way to stop the VM is to exit
+ the context manager.
+ """
+ return await _qmp_command(self._ipc, command)
+
+
+class SubprocessError(Exception):
+ """An exception thrown when a subprocess failed unexpectedly."""
+
+ def __init__(
+ self,
+ args: tuple[str | Path | tuple[str | Path, ...], ...],
+ returncode: int,
+ output: bytes | None = None,
+ ) -> None:
+ """Create a SubprocessError instance.
+
+ - args: the arguments to the command that failed
+ - returncode: the non-zero return code
+ """
+ self.args = args
+ self.returncode = returncode
+ self.output = output
+
+ if returncode < 0:
+ msg = f"Subprocess terminated by {signal.Signals(-returncode).name}\n"
+ else:
+ msg = f"Subprocess exited unexpectedly with return code {returncode}:\n"
+
+ if self.output:
+ out = "\n🗯️ Output:\n\n" + self.output.decode(errors="replace")
+ else:
+ out = ""
+
+ super().__init__(f"{msg}\n{_pretty_print_args(*args)}\n{out}\n")
+
+
+def cleanup_on_signal() -> None:
+ """Register SIGHUP and SIGTERM signal handlers to cleanly exit.
+
+ This raises an exception, cleaning up running subprocesses and the IPC
+ directory, in contrast to the default interpreter behaviour of a direct
+ exit.
+ """
+
+ def _term(*args: object) -> Never:
+ del args
+ # This raises SystemExit which will bubble out of the handler
+ sys.exit("I don't blame you.")
+
+ signal.signal(signal.SIGHUP, _term)
+ signal.signal(signal.SIGTERM, _term)
+
+
+async def _ssh_properly_configured() -> bool:
+ proc = None
+ try:
+ proc = await asyncio.subprocess.create_subprocess_exec(
+ *("ssh", "-G", "tt.n"),
+ stdout=asyncio.subprocess.PIPE,
+ )
+ stdout, _stderr = await proc.communicate()
+ except OSError:
+ return False
+ else:
+ return b"/test.thing/tt.n/ssh\n" in stdout
+ finally:
+ if proc is not None:
+ await proc.wait()
+
+
+async def _show_ssh_hints(ui: UI, vm_id: str) -> None:
+ if not _stderr_is_tty():
+ return
+
+ if await _ssh_properly_configured():
+ ui.print(f"\n🍓 VM running. Connect with: \033[1mssh {vm_id}\033[0m\n")
+ else:
+ ui.print(f"""
+Please consider adding this stanza to your SSH config:
+
+Host tt.*
+ ControlPath ${{XDG_RUNTIME_DIR}}/test.thing/%h/ssh
+
+At which point you can connect to the VM using \033[1mssh {vm_id}\033[0m\n""")
+
+
+@contextlib.contextmanager
+def cli_helper() -> Iterator[None]:
+ """Help use test.thing from CLI tools.
+
+ This installs a signal handler for clean exit on SIGHUP and SIGTERM and
+ catches SubprocessError, TimeoutError, and KeyboardInterrupt, printing a
+ message to stderr and calling sys.exit().
+
+ Because of how cancellation is used internally to handle KeyboardInterrupt,
+ you should use this *outside* of asyncio.run().
+ """
+ cleanup_on_signal()
+ try:
+ yield
+ except* (SubprocessError, TimeoutError, KeyboardInterrupt) as eg:
+ for exc in eg.exceptions:
+ sys.stderr.write(f"\n🤦 [{exc.__class__.__name__}] {exc}\n\n")
+ sys.exit("I'm sorry it didn't work out.")
+
+
+def _main() -> None:
+ class AppendTuple(argparse.Action):
+ def __call__(
+ self,
+ parser: argparse.ArgumentParser,
+ namespace: argparse.Namespace,
+ values: str | Sequence[Any] | None,
+ option_string: str | None = None,
+ ) -> None:
+ del parser
+ fwds = getattr(namespace, self.dest) or ()
+ fwds = (*fwds, (option_string, values))
+ setattr(namespace, self.dest, fwds)
+
+ parser = argparse.ArgumentParser(
+ description="test.thing - a simple modern VM runner"
+ )
+ parser.add_argument(
+ "--maintain", "-m", action="store_true", help="Changes are permanent"
+ )
+ parser.add_argument(
+ "--attach", "-a", action="store_true", help="Attach to the VM console"
+ )
+ parser.add_argument(
+ "--sit", action="store_true", help="Wait for enter key on exceptions"
+ )
+ parser.add_argument(
+ "--debug", "-d", action="store_true", help="Enable debug output"
+ )
+ parser.add_argument("--root-pw", help="Set root's password")
+ parser.add_argument(
+ "--boot",
+ choices=("efi", "mbr"),
+ default="efi",
+ help="How to boot the image (default: efi)",
+ )
+ parser.add_argument(
+ "--verbose", "-v", action="store_true", help="Print verbose output"
+ )
+ parser.add_argument(
+ "--ssh-key",
+ "-i",
+ type=Path,
+ action="append",
+ help="Path to SSH private key (default: generate)",
+ )
+ parser.add_argument(
+ "--timeout", type=float, help="For startup, in seconds, or 'inf' (default: 30)"
+ )
+ parser.add_argument(
+ "--no-network", action="store_true", help="Isolate the VM from the Internet"
+ )
+ parser.add_argument(
+ "-L",
+ "-R",
+ "-D",
+ default=[],
+ dest="fwd_spec",
+ action=AppendTuple,
+ help="Setup an SSH-style port forward",
+ )
+ parser.add_argument(
+ "--script",
+ "-c",
+ metavar="COMMAND",
+ action="append",
+ help="Execute this (shell-interpreted) command",
+ )
+ parser.add_argument(
+ "--start-unit",
+ "-s",
+ metavar="UNIT",
+ action="append",
+ dest="script",
+ type=(lambda s: f"systemctl enable --now {shlex.quote(s)}"),
+ help="Start this systemd unit",
+ )
+
+ parser.add_argument("image", type=Path, help="The path to a qcow2 VM image to run")
+ parser.add_argument("cmd", nargs="*")
+ args = parser.parse_intermixed_args()
+
+ async def _async_main() -> None:
+ with cli_helper(), IpcDirectory() as ipc:
+ ui = UI(status_messages=not args.attach, verbose=args.verbose)
+
+ async with VirtualMachine(
+ args.image,
+ ipc=ipc,
+ attach_console=args.attach,
+ boot=args.boot,
+ cloud_init_user_data={
+ "chpasswd": {"list": "root:foobar", "expire": False},
+ "ssh_pwauth": True,
+ },
+ identities=args.ssh_key or (COCKPIT_TEST_IDENTITY,),
+ journal=print,
+ networks=(() if args.no_network else (Network.user(),)),
+ provision_ssh_key=not args.maintain,
+ sit=args.sit,
+ snapshot=not args.maintain,
+ timeout=args.timeout,
+ ui=ui,
+ ) as vm:
+ for spec in args.fwd_spec:
+ await vm.forward_port(spec)
+
+ for cmd in args.script or ():
+ await vm.execute(cmd, stdout=None)
+
+ if args.attach:
+ await vm.wait_exit()
+ elif args.cmd:
+ await vm.execute(*args.cmd, stdin=None, stdout=None)
+ else:
+ await _show_ssh_hints(ui, vm.get_id())
+ await ui.run("ssh", *vm.ssh_args, vm.get_id(), stdin=None)
+
+ if args.debug:
+ logging.basicConfig(level=logging.DEBUG)
+
+ asyncio.run(_async_main(), debug=args.debug)
+
+
+if __name__ == "__main__":
+ _main()
diff --git a/pyproject.toml b/pyproject.toml
index de28854bcf..9e7e509824 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -38,6 +38,7 @@ module = [
[tool.ruff]
exclude = [
+ "lib/testthing.py",
".git/",
]
line-length = 118
diff --git a/test/run b/test/run
index 97fc396462..edf87f0a32 100755
--- a/test/run
+++ b/test/run
@@ -17,6 +17,6 @@ find_python_files() {
find_scripts 'python3' '*.py'
}
-find_python_files | xargs -0 ruff check --quiet
+find_python_files | xargs -0 ruff check --quiet --force-exclude
find_python_files | xargs -0 mypy --no-error-summary
pytest -vv