# SPDX-FileCopyrightText: Copyright (C) 2026 Bruce Ashfield # # SPDX-License-Identifier: MIT """ Libvirt recipe and runtime tests. Tier 1: Static assertions on recipe files (no build required) Tier 2: Build verification (requires bitbake) Tier 3: Boot tests on kvm-image-minimal (requires QEMU with KVM) The tests automatically build kvm-image-minimal with the kvm DISTRO_FEATURE before booting. No local.conf changes needed. Run: # Static tests only pytest tests/test_libvirt.py -v -m "not slow and not boot" --poky-dir /opt/bruce/poky # All tests including build and boot pytest tests/test_libvirt.py -v --poky-dir /opt/bruce/poky Options: --boot-timeout QEMU boot timeout (default: 120s) --no-kvm Disable KVM acceleration """ import os import re import subprocess import tempfile import time import pytest from pathlib import Path try: import pexpect PEXPECT_AVAILABLE = True except ImportError: PEXPECT_AVAILABLE = False def _run_bitbake(build_dir, recipe, extra_vars=None, timeout=3600): """Run bitbake with optional variable overrides via -R conf file.""" bb_cmd = "bitbake" conf_file = None if extra_vars: conf_file = tempfile.NamedTemporaryFile( mode='w', suffix='.conf', prefix='pytest-libvirt-', dir=str(build_dir / "conf"), delete=False) for var, val in extra_vars.items(): conf_file.write(f'{var} = "{val}"\n') conf_file.close() bb_cmd += f" -R {conf_file.name}" bb_cmd += f" {recipe}" poky_dir = build_dir.parent full_cmd = f"bash -c 'cd {poky_dir} && source oe-init-build-env {build_dir} >/dev/null 2>&1 && {bb_cmd}'" try: return subprocess.run(full_cmd, shell=True, cwd=build_dir, timeout=timeout, capture_output=True, text=True) finally: if conf_file: os.unlink(conf_file.name) # ============================================================================ # Fixtures # ============================================================================ @pytest.fixture(scope="module") def poky_dir(request): path = Path(request.config.getoption("--poky-dir")) if not path.exists(): pytest.skip(f"Poky directory not found: {path}") return path @pytest.fixture(scope="module") def build_dir(request, poky_dir): bd = request.config.getoption("--build-dir") path = Path(bd) if bd else poky_dir / "build" if not path.exists(): pytest.skip(f"Build directory not found: {path}") return path @pytest.fixture(scope="module") def meta_virt_dir(poky_dir): path = poky_dir / "meta-virtualization" if not path.exists(): pytest.skip(f"meta-virtualization not found: {path}") return path @pytest.fixture(scope="module") def libvirt_recipe(meta_virt_dir): path = meta_virt_dir / "recipes-extended" / "libvirt" / "libvirt_git.bb" if not path.exists(): pytest.skip(f"libvirt recipe not found: {path}") return path @pytest.fixture(scope="module") def libvirt_files_dir(meta_virt_dir): path = meta_virt_dir / "recipes-extended" / "libvirt" / "libvirt" if not path.exists(): pytest.skip(f"libvirt files dir not found: {path}") return path @pytest.fixture(scope="module") def kvm_image(build_dir): """Build kvm-image-minimal with kvm DISTRO_FEATURE.""" result = _run_bitbake( build_dir, "kvm-image-minimal", extra_vars={ "DISTRO_FEATURES:append": " kvm", }, ) if result.returncode != 0: pytest.fail(f"KVM image build failed: {result.stderr}") @pytest.fixture(scope="module") def libvirt_session(request, poky_dir, build_dir, kvm_image): """Boot kvm-image-minimal and provide a pexpect session.""" if not PEXPECT_AVAILABLE: pytest.skip("pexpect not installed") machine = request.config.getoption("--machine", default="qemux86-64") timeout = int(request.config.getoption("--boot-timeout", default="120")) no_kvm = request.config.getoption("--no-kvm", default=False) kvm_opt = "" if no_kvm else "kvm" cmd = ( f"bash -c 'cd {poky_dir} && " f"source oe-init-build-env {build_dir} >/dev/null 2>&1 && " f"runqemu {machine} kvm-image-minimal ext4 nographic slirp " f"{kvm_opt} qemuparams=\"-m 2048\"'" ) child = pexpect.spawn(cmd, encoding='utf-8', timeout=timeout) child.logfile_read = open('/tmp/runqemu-libvirt-test.log', 'w') try: index = child.expect([r'login:', r'root@', pexpect.TIMEOUT, pexpect.EOF], timeout=timeout) if index == 0: child.sendline('root') child.expect([r'root@', r'#'], timeout=30) elif index >= 2: raise RuntimeError("Boot failed") child.sendline('export TERM=dumb') child.expect(r'root@[^:]+:[^#]+#', timeout=10) yield child except RuntimeError as e: pytest.skip(f"Failed to boot: {e}") finally: child.sendline('poweroff') try: child.expect(pexpect.EOF, timeout=30) except pexpect.TIMEOUT: child.terminate(force=True) def run_cmd(child, cmd, timeout=60): """Run a command and return (output, returncode).""" marker = f"__MARKER_{time.monotonic_ns()}__" child.sendline(f"{cmd}; echo {marker} $?") child.expect(marker + r" (\d+)", timeout=timeout) raw = child.before raw = re.sub(r'\x1b\]3008;[^\x07\x1b]*(?:\x07|\x1b\\)', '', raw) raw = re.sub(r'\x1b\[[0-9;]*[A-Za-z]', '', raw) output = raw.strip() rc = int(child.match.group(1)) child.expect(r'root@[^:]+:[^#]+#', timeout=10) return output, rc # ============================================================================ # Tier 1: Static assertions (no build required) # ============================================================================ class TestLibvirtRecipeStatic: """Static checks on the libvirt recipe file.""" def test_recipe_exists(self, libvirt_recipe): assert libvirt_recipe.exists() def test_has_meson_inherit(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "inherit meson" in content def test_has_systemd_support(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "inherit" in content and "systemd" in content assert "SYSTEMD_SERVICE" in content def test_systemd_services_defined(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "libvirtd.service" in content assert "virtlockd.service" in content def test_has_useradd(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "inherit" in content and "useradd" in content assert "qemu" in content assert "libvirt" in content def test_qemu_packageconfig(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "PACKAGECONFIG[qemu]" in content assert "driver_qemu" in content def test_lxc_packageconfig(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "PACKAGECONFIG[lxc]" in content assert "driver_lxc" in content def test_xen_packageconfig_gated(self, libvirt_recipe): """libxl PACKAGECONFIG should be gated on xen DISTRO_FEATURE.""" content = libvirt_recipe.read_text() assert "PACKAGECONFIG[libxl]" in content for line in content.splitlines(): if "libxl" in line and "DISTRO_FEATURES" in line: assert "xen" in line break else: pytest.fail("libxl PACKAGECONFIG not gated on xen DISTRO_FEATURE") def test_nftables_rdepends(self, libvirt_recipe): """libvirt-libvirtd should RDEPEND on nftables when configured.""" content = libvirt_recipe.read_text() assert "nftables" in content def test_nftables_or_iptables_rdepends(self, libvirt_recipe): """RDEPENDS should select nftables or iptables based on PACKAGECONFIG.""" content = libvirt_recipe.read_text() assert "bb.utils.contains('PACKAGECONFIG', 'nftables'" in content, \ "RDEPENDS should conditionally select nftables or iptables" def test_packages_split(self, libvirt_recipe): """Recipe should split into libvirt, libvirt-libvirtd, libvirt-virsh.""" content = libvirt_recipe.read_text() assert "${PN}-libvirtd" in content assert "${PN}-virsh" in content def test_cve_status_entries(self, libvirt_recipe): content = libvirt_recipe.read_text() assert "CVE_STATUS" in content cve_count = content.count("CVE_STATUS[CVE-") assert cve_count >= 5, f"Expected at least 5 CVE entries, found {cve_count}" class TestLibvirtHookSupport: """Tests for hook_support.py (the file from the recent bug report).""" def test_hook_script_exists(self, libvirt_files_dir): path = libvirt_files_dir / "hook_support.py" assert path.exists() def test_hook_script_python3_shebang(self, libvirt_files_dir): content = (libvirt_files_dir / "hook_support.py").read_text() first_line = content.splitlines()[0] assert "python3" in first_line or "python" in first_line def test_hook_script_uses_text_mode(self, libvirt_files_dir): """Popen should use text=True for python3 string compatibility.""" content = (libvirt_files_dir / "hook_support.py").read_text() assert "text=True" in content, \ "Popen should use text=True for python3 string/bytes compatibility" def test_hook_script_regex_raw_string(self, libvirt_files_dir): """Regex should use raw string to avoid SyntaxWarning in python3.12+.""" content = (libvirt_files_dir / "hook_support.py").read_text() for line in content.splitlines(): if "re.compile" in line and "\\w" in line: assert "rf\"" in line or "r'" in line or 'r"' in line, \ f"Regex with \\w should use raw string: {line.strip()}" def test_hook_script_syntax_valid(self, libvirt_files_dir): """hook_support.py should parse without syntax errors.""" import py_compile path = libvirt_files_dir / "hook_support.py" try: py_compile.compile(str(path), doraise=True) except py_compile.PyCompileError as e: pytest.fail(f"Syntax error in hook_support.py: {e}") def test_hook_installs_for_all_domains(self, libvirt_recipe): """hook_support.py should be installed for daemon, lxc, network, qemu.""" content = libvirt_recipe.read_text() for hook in ["daemon", "lxc", "network", "qemu"]: assert hook in content, f"Hook not installed for domain: {hook}" class TestLibvirtServiceFiles: """Tests for systemd service and init script files.""" def test_initscript_exists(self, libvirt_files_dir): assert (libvirt_files_dir / "libvirtd.sh").exists() def test_initscript_has_lsb_header(self, libvirt_files_dir): content = (libvirt_files_dir / "libvirtd.sh").read_text() assert "BEGIN INIT INFO" in content def test_initscript_start_stop(self, libvirt_files_dir): content = (libvirt_files_dir / "libvirtd.sh").read_text() assert "start)" in content assert "stop)" in content assert "restart)" in content def test_dnsmasq_conf_exists(self, libvirt_files_dir): assert (libvirt_files_dir / "dnsmasq.conf").exists() def test_libvirtd_conf_exists(self, libvirt_files_dir): assert (libvirt_files_dir / "libvirtd.conf").exists() def test_gnutls_helper_exists(self, libvirt_files_dir): assert (libvirt_files_dir / "gnutls-helper.py").exists() class TestLibvirtPatchFiles: """Verify patches exist and reference valid upstream issues.""" def test_patches_exist(self, libvirt_files_dir): patches = list(libvirt_files_dir.glob("*.patch")) assert len(patches) >= 1, "Expected at least one patch" def test_buildpath_patches_present(self, libvirt_files_dir): """Patches for buildpaths should exist.""" patches = list(libvirt_files_dir.glob("*.patch")) buildpath_patches = [p for p in patches if "build-path" in p.name or "build_path" in p.name or "gendispatch" in p.name] assert len(buildpath_patches) >= 1, \ "Expected at least one buildpath-related patch" class TestKvmImageRecipe: """Static checks on kvm-image-minimal recipe.""" def test_recipe_exists(self, meta_virt_dir): path = meta_virt_dir / "recipes-extended" / "images" / "kvm-image-minimal.bb" assert path.exists() def test_requires_kvm_distro_feature(self, meta_virt_dir): content = (meta_virt_dir / "recipes-extended" / "images" / "kvm-image-minimal.bb").read_text() assert "kvm" in content assert "REQUIRED_DISTRO_FEATURES" in content def test_includes_libvirt(self, meta_virt_dir): content = (meta_virt_dir / "recipes-extended" / "images" / "kvm-image-minimal.bb").read_text() assert "libvirt" in content assert "libvirt-libvirtd" in content assert "libvirt-virsh" in content def test_includes_qemu(self, meta_virt_dir): content = (meta_virt_dir / "recipes-extended" / "images" / "kvm-image-minimal.bb").read_text() assert "qemu" in content def test_includes_kvm_modules(self, meta_virt_dir): content = (meta_virt_dir / "recipes-extended" / "images" / "kvm-image-minimal.bb").read_text() assert "kernel-module-kvm" in content # ============================================================================ # Tier 2: Build verification (requires bitbake) # ============================================================================ @pytest.mark.slow class TestLibvirtBuild: """Build tests for libvirt recipe.""" def test_libvirt_builds(self, build_dir): result = _run_bitbake(build_dir, "libvirt") assert result.returncode == 0, f"libvirt build failed: {result.stderr}" def test_kvm_image_builds(self, build_dir, kvm_image): """kvm-image-minimal builds successfully (via kvm_image fixture).""" pass # ============================================================================ # Tier 3: Boot tests (requires QEMU) # ============================================================================ # Libvirt v12 defaults to modular daemons (virtqemud, virtnetworkd, etc.) # but kvm-image-minimal runs the monolithic libvirtd. virsh must be pointed # at the monolithic socket explicitly. _VIRSH = "virsh -c qemu+unix:///system?socket=/var/run/libvirt/libvirt-sock" @pytest.mark.slow @pytest.mark.boot class TestLibvirtRuntime: """Boot tests verifying libvirt on a running kvm-image-minimal.""" def test_libvirtd_running(self, libvirt_session): """libvirtd systemd service should be active.""" output, rc = run_cmd(libvirt_session, "systemctl is-active libvirtd") assert "active" in output, f"libvirtd not active: {output}" def test_virtlockd_running(self, libvirt_session): """virtlockd should be active (started on demand via socket).""" output, rc = run_cmd(libvirt_session, "systemctl is-active virtlockd.socket") assert "active" in output, f"virtlockd.socket not active: {output}" def test_virsh_available(self, libvirt_session): """virsh command should be available.""" output, rc = run_cmd(libvirt_session, "which virsh") assert rc == 0, f"virsh not found: {output}" def test_virsh_version(self, libvirt_session): """virsh version should report libvirt version.""" output, rc = run_cmd(libvirt_session, "virsh --version") assert rc == 0, f"virsh version failed: {output}" def test_virsh_connect(self, libvirt_session): """virsh should connect to local libvirtd.""" output, rc = run_cmd(libvirt_session, f"{_VIRSH} uri") assert rc == 0, f"virsh uri failed: {output}" def test_virsh_capabilities(self, libvirt_session): """virsh capabilities should return valid XML.""" output, rc = run_cmd(libvirt_session, f"{_VIRSH} capabilities") assert rc == 0, f"virsh capabilities failed: {output}" assert "" in output or "capabilities" in output def test_virsh_nodeinfo(self, libvirt_session): """virsh nodeinfo should report system info.""" output, rc = run_cmd(libvirt_session, f"{_VIRSH} nodeinfo") assert rc == 0, f"virsh nodeinfo failed: {output}" assert "CPU model" in output or "cpu" in output.lower() def test_virsh_list(self, libvirt_session): """virsh list should work (even with no domains).""" output, rc = run_cmd(libvirt_session, f"{_VIRSH} list --all") assert rc == 0, f"virsh list failed: {output}" def test_default_network(self, libvirt_session): """Default network should be defined.""" output, rc = run_cmd(libvirt_session, f"{_VIRSH} net-list --all") assert rc == 0, f"virsh net-list failed: {output}" def test_hook_scripts_installed(self, libvirt_session): """Hook scripts should be installed for all domains.""" output, rc = run_cmd(libvirt_session, "ls /etc/libvirt/hooks/") assert rc == 0 for hook in ["daemon", "lxc", "network", "qemu"]: assert hook in output, f"Hook script missing: {hook}" def test_hook_scripts_executable(self, libvirt_session): """Hook scripts should be executable.""" output, rc = run_cmd(libvirt_session, "test -x /etc/libvirt/hooks/qemu && echo OK") assert "OK" in output, "qemu hook not executable" def test_qemu_user_exists(self, libvirt_session): """qemu user should exist (created by useradd in recipe).""" output, rc = run_cmd(libvirt_session, "grep ^qemu: /etc/passwd") assert rc == 0, f"qemu user not found: {output}" def test_libvirt_group_exists(self, libvirt_session): """libvirt group should exist.""" output, rc = run_cmd(libvirt_session, "grep ^libvirt: /etc/group") assert rc == 0, f"libvirt group not found: {output}"