test: Add a boot test for bootc pxe-tar-xz

This uses QEMU from vmtest, provided by the osbuild/images repository.
It grabs the ostree= kernel cmdline from the grub.cfg in the tarball,
boots qemu with the artifacts and tests that ssh login is successful.

NOTE: Running bootc status from the booted system currently does not
work so that has been disabled until it has been fixed.

The tests include a cmdline option, --keep-tmpdir, that will keep the
pxe-tar-xz tmpdir around after the test runs in order to help with
debugging problems.

Author: bcl

test/bib/conftest.py

@@ -10,12 +10,20 @@ def pytest_addoption(parser):
                      help=("Force AWS upload when building AMI, failing if credentials are not set. "
                            "If not set, the upload will be performed only when credentials are available."))
 
+    parser.addoption("--keep-tmpdir", action="store_true", default=False,
+                     help="Keep the pxe tar tmpdir.")
+
 
 @pytest.fixture(name="force_aws_upload", scope="session")
 def force_aws_upload_fixture(request):
     return request.config.getoption("--force-aws-upload")
 
 
+@pytest.fixture(name="keep_tmpdir", scope="session")
+def keep_tmpdir_fixture(request):
+    return request.config.getoption("--keep-tmpdir")
+
+
 # see https://hackebrot.github.io/pytest-tricks/param_id_func/ and
 # https://docs.pytest.org/en/7.1.x/reference/reference.html#pytest.hookspec.pytest_make_parametrize_id
 def pytest_make_parametrize_id(config, val):  # pylint: disable=W0613

test/bib/test_build_pxe.py

@@ -6,7 +6,11 @@
 import string
 import subprocess
 import textwrap
+import threading
+
 from contextlib import ExitStack
+from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
+from tempfile import TemporaryDirectory
 
 import pytest
 # local test utils
@@ -24,6 +28,83 @@
     shared_tmpdir_fixture,
 )
 from vmtest.vm import QEMU
+from vmtest.util import get_free_port
+
+
+def get_ostree_path(grub_path):
+    """Get the ostree= boot path from a grub.cfg file"""
+    with open(grub_path, encoding="utf-8") as f:
+        for line in f.readlines():
+            if "ostree=" not in line:
+                continue
+            args=line.split()
+            for a in args:
+                if a.startswith("ostree="):
+                    return a
+    return ""
+
+
+class DirServer(ThreadingHTTPServer):
+    def __init__(self, address, directory):
+        super().__init__(address, SimpleHTTPRequestHandler)
+        self.directory = directory
+
+    def finish_request(self, request, client_address):
+        SimpleHTTPRequestHandler(request, client_address, self, directory=self.directory)
+
+
+def boot_qemu_pxe(arch, pxe_tar_path, container_ref, username, password, ssh_key_path, keep=False):
+    with ExitStack() as cm:
+        # unpack the tar and create a combined image
+        tmpdir = cm.enter_context(TemporaryDirectory(dir="/var/tmp", prefix="qemu-pxe-", delete=keep))
+        subprocess.check_call(
+            ["tar", "-C", tmpdir, "-x", "-f", pxe_tar_path])
+        subprocess.check_call(
+            "echo rootfs.img | cpio -c --quiet -L -o > rootfs.cpio", shell=True, cwd=tmpdir)
+        subprocess.check_call(
+            "cat initrd.img rootfs.cpio > combined.img", shell=True, cwd=tmpdir)
+
+        # Get the ostree= kernel cmdline argument from grub.cfg
+        ostree_path = get_ostree_path(pathlib.Path(tmpdir) / "grub.cfg")
+        assert ostree_path.startswith("ostree=")
+
+        # Setup http server thread for the rootfs.img
+        http_server = DirServer(('127.0.0.1', 0), tmpdir)
+        http_port = http_server.server_port
+        threading.Thread(target=http_server.serve_forever).start()
+        try:
+            # test disk is unused for live OS
+            test_disk_path = pathlib.Path(tmpdir) / "disk.img"
+            with open(test_disk_path, "w", encoding="utf-8") as fp:
+                fp.truncate(0)
+
+            # test both the combined and HTTP rootfs variants
+            for use_ovmf in [False, True]:
+                for root_arg, initrd_file in [
+                    ("live:/rootfs.img", "combined.img"),
+                    (f"live:http://10.0.2.2:{http_port}/rootfs.img", "initrd.img")
+                ]:
+                    append_arg = (
+                        f"rd.live.image root={root_arg} rw console=ttyS0 "
+                        f"systemd.debug-shell=ttyS0 "
+                        f"{ostree_path}"
+                    )
+                    extra_args = [
+                        "-kernel", str(pathlib.Path(tmpdir) / "vmlinuz"),
+                        "-initrd", str(pathlib.Path(tmpdir) / initrd_file),
+                        "-append", append_arg
+                    ]
+
+                    with QEMU(test_disk_path, memory="4096", arch=arch, extra_args=extra_args) as vm:
+                        vm.start(use_ovmf=use_ovmf)
+                        vm.run("true", user=username, password=password)
+                        ret = vm.run("mount", user="root", keyfile=ssh_key_path)
+##                        ret = vm.run(["bootc", "status"], user="root", keyfile=ssh_key_path)
+##                        assert f"image: {container_ref}" in ret.stdout
+        finally:
+            http_server.shutdown()
+
+        return True
 
 
 @pytest.mark.skipif(platform.system() != "Linux", reason="boot test only runs on linux right now")
@@ -33,7 +114,7 @@
     "quay.io/centos-bootc/centos-bootc:stream9",
 ])
 # pylint: disable=too-many-locals
-def test_bootc_pxe_tar_xz(tmp_path, build_container, container_ref):
+def test_bootc_pxe_tar_xz(keep_tmpdir, tmp_path, build_container, container_ref):
     # XXX: duplicated from test_build_disk.py
     username = "test"
     password = "".join(
@@ -101,4 +182,7 @@ def test_bootc_pxe_tar_xz(tmp_path, build_container, container_ref):
         ]
         subprocess.check_call(cmd)
         assert os.path.exists(output_path / "xz" / "pxe.tar.xz")
-        # TODO use lzap's pxe boot code from images to boot the result
+        boot_qemu_pxe(platform.machine(), output_path / "xz" / "pxe.tar.xz",
+                      container_ref,
+                      username, password,
+                      ssh_keyfile_private_path)