Merge tag 'pull-request-2024-12-18' of https://gitlab.com/thuth/qemu into staging

* Lots of functional test improvements (clean-ups, don't fail on temporary download errors, etc.) * Convert some more avocado tests to the functional framework * Disallow building with libnfs v6 due to an API breakage # -----BEGIN PGP SIGNATURE----- # # iQJFBAABCAAvFiEEJ7iIR+7gJQEY8+q5LtnXdP5wLbUFAmdirOIRHHRodXRoQHJl # ZGhhdC5jb20ACgkQLtnXdP5wLbU0NRAAke8X0B6OOD+99lY5nc7Hrh7N1m+sw5Lw # TVwIpxdhxU11vgdlCodfdoVJCV1NGVHwkR57lLNr+bdspWDBBwlmUWn0+t2QCXGe # oyQsV+boznsjG9pan6v6DcU/gOu7/7ZydhJi+M8Msf8ah0lcn/otAdC4ZFB93JLh # 6xPnj69y8HomCW+wMyXl7WTjcWX0wQFzweEYY8p7X7p1rtjYyseiZlRjNAvPgTMI # jznZ6v9/qU54xR9RnKdW+0m1Qu06nx26Wz+ZBlvrJS1Llloe23X9+LY1tDD0Xh1D # 9P0v9PuaBWRRF+UjVjl37LMyn9h1aaKFKBoWQiKMbyvOVr4ncobjRgN8r5kdNxDP # FZ/fA1GiX8O3foN9uB9JLKd6Hl49LAqQSPzAneEc3pfQLH3NdAjPxJDbJH5fyMa7 # qVOQC0Bdy8+2kCxFfKbemrwDOFcyq1fVYcADPDZySjMiPnwFJ1Qpni1tXY1PZ+Tl # Q18AsFJanyAAn7L+8R3Yl54983SuR5eXIFxO+Tq9mw1V1V2h+Cm09HGcS8y5bxFG # Xh+jhMsMB98NFLR87W6olwl57gKllSbTYuGtiz9TrbnuT/THhUJ0k/B76L7C9HWE # ZefkFxC5Zy8jrcz3pgarO+19V+eXg5rwGtEngRQrji/3cY5CbK7Jeh5nvZQeASpb # nZ/gJ/gC8Gs= # =SWw6 # -----END PGP SIGNATURE----- # gpg: Signature made Wed 18 Dec 2024 06:07:14 EST # gpg: using RSA key 27B88847EEE0250118F3EAB92ED9D774FE702DB5 # gpg: issuer "thuth@redhat.com" # gpg: Good signature from "Thomas Huth <th.huth@gmx.de>" [full] # gpg: aka "Thomas Huth <thuth@redhat.com>" [full] # gpg: aka "Thomas Huth <huth@tuxfamily.org>" [full] # gpg: aka "Thomas Huth <th.huth@posteo.de>" [unknown] # Primary key fingerprint: 27B8 8847 EEE0 2501 18F3 EAB9 2ED9 D774 FE70 2DB5 * tag 'pull-request-2024-12-18' of https://gitlab.com/thuth/qemu: (38 commits) meson.build: Disallow libnfs v6 to fix the broken macOS build tests/functional: Convert the hotplug_cpu avocado test tests/functional: Convert the intel_iommu avocado test tests/functional: Add a helper function for retrieving the hostfwd port tests/functional: Convert the arm virt avocado test tests/functional: Convert the quanta-gsj avocado test MAINTAINERS: add myself as reviewer for functional test suite tests/functional: ignore errors when caching assets, except for 404 tests/functional: skip tests if assets are not available tests/functional: remove now unused 'run_cmd' helper tests/functional: replace 'run_cmd' with subprocess helpers tests/functional: drop back compat imports from utils.py tests/functional: convert tests to new uncompress helper tests/functional: add 'uncompress' to QemuBaseTest tests/functional: add a generalized uncompress helper tests/functional: convert tests to new archive_extract helper tests/functional: add 'archive_extract' to QemuBaseTest tests/functional: add a generalized archive_extract tests/functional: let cpio_extract accept filenames tests/functional: add common deb_extract helper ... Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
2026-07-06 00:20:22 -07:00 · 2024-12-18 20:24:51 -05:00
parent 877fad2a3e e2d98f2571
commit ba182a693f
79 changed files with 1275 additions and 1029 deletions
@@ -872,6 +872,7 @@ F: tests/qtest/adm1266-test.c
 F: pc-bios/npcm7xx_bootrom.bin
 F: roms/vbootrom
 F: docs/system/arm/nuvoton.rst
+F: tests/functional/test_arm_quanta_gsj.py

 Raspberry Pi
 M: Peter Maydell <peter.maydell@linaro.org>
@@ -3681,6 +3682,7 @@ S: Supported
 F: hw/i386/intel_iommu.c
 F: hw/i386/intel_iommu_internal.h
 F: include/hw/i386/intel_iommu.h
+F: tests/functional/test_intel_iommu.py

 AMD-Vi Emulation
 S: Orphan
@@ -4157,6 +4159,7 @@ W: https://cirrus-ci.com/github/qemu/qemu
 Functional testing framework
 M: Thomas Huth <thuth@redhat.com>
 R: Philippe Mathieu-Daudé <philmd@linaro.org>
+R: Daniel P. Berrange <berrange@redhat.com>
 F: tests/functional/qemu_test/

 Windows Hosted Continuous Integration
@@ -1145,7 +1145,7 @@ endif

 libnfs = not_found
 if not get_option('libnfs').auto() or have_block
-  libnfs = dependency('libnfs', version: '>=1.9.3',
+  libnfs = dependency('libnfs', version: ['>=1.9.3', '<6.0.0'],
                      required: get_option('libnfs'),
                      method: 'pkg-config')
 endif
@@ -94,110 +94,3 @@ class BootLinuxConsole(LinuxKernelTest):
        self.vm.launch()
        console_pattern = 'Kernel command line: %s' % kernel_command_line
        self.wait_for_console_pattern(console_pattern)
-
-    def test_arm_virt(self):
-        """
-        :avocado: tags=arch:arm
-        :avocado: tags=machine:virt
-        :avocado: tags=accel:tcg
-        """
-        kernel_url = ('https://archives.fedoraproject.org/pub/archive/fedora'
-                      '/linux/releases/29/Everything/armhfp/os/images/pxeboot'
-                      '/vmlinuz')
-        kernel_hash = 'e9826d741b4fb04cadba8d4824d1ed3b7fb8b4d4'
-        kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
-
-        self.vm.set_console()
-        kernel_command_line = (self.KERNEL_COMMON_COMMAND_LINE +
-                               'console=ttyAMA0')
-        self.vm.add_args('-kernel', kernel_path,
-                         '-append', kernel_command_line)
-        self.vm.launch()
-        console_pattern = 'Kernel command line: %s' % kernel_command_line
-        self.wait_for_console_pattern(console_pattern)
-
-    @skipUnless(os.getenv('AVOCADO_TIMEOUT_EXPECTED'), 'Test might timeout')
-    def test_arm_quanta_gsj(self):
-        """
-        :avocado: tags=arch:arm
-        :avocado: tags=machine:quanta-gsj
-        :avocado: tags=accel:tcg
-        """
-        # 25 MiB compressed, 32 MiB uncompressed.
-        image_url = (
-                'https://github.com/hskinnemoen/openbmc/releases/download/'
-                '20200711-gsj-qemu-0/obmc-phosphor-image-gsj.static.mtd.gz')
-        image_hash = '14895e634923345cb5c8776037ff7876df96f6b1'
-        image_path_gz = self.fetch_asset(image_url, asset_hash=image_hash)
-        image_name = 'obmc.mtd'
-        image_path = os.path.join(self.workdir, image_name)
-        archive.gzip_uncompress(image_path_gz, image_path)
-
-        self.vm.set_console()
-        drive_args = 'file=' + image_path + ',if=mtd,bus=0,unit=0'
-        self.vm.add_args('-drive', drive_args)
-        self.vm.launch()
-
-        # Disable drivers and services that stall for a long time during boot,
-        # to avoid running past the 90-second timeout. These may be removed
-        # as the corresponding device support is added.
-        kernel_command_line = self.KERNEL_COMMON_COMMAND_LINE + (
-                'console=${console} '
-                'mem=${mem} '
-                'initcall_blacklist=npcm_i2c_bus_driver_init '
-                'systemd.mask=systemd-random-seed.service '
-                'systemd.mask=dropbearkey.service '
-        )
-
-        self.wait_for_console_pattern('> BootBlock by Nuvoton')
-        self.wait_for_console_pattern('>Device: Poleg BMC NPCM730')
-        self.wait_for_console_pattern('>Skip DDR init.')
-        self.wait_for_console_pattern('U-Boot ')
-        interrupt_interactive_console_until_pattern(
-                self, 'Hit any key to stop autoboot:', 'U-Boot>')
-        exec_command_and_wait_for_pattern(
-                self, "setenv bootargs ${bootargs} " + kernel_command_line,
-                'U-Boot>')
-        exec_command_and_wait_for_pattern(
-                self, 'run romboot', 'Booting Kernel from flash')
-        self.wait_for_console_pattern('Booting Linux on physical CPU 0x0')
-        self.wait_for_console_pattern('CPU1: thread -1, cpu 1, socket 0')
-        self.wait_for_console_pattern('OpenBMC Project Reference Distro')
-        self.wait_for_console_pattern('gsj login:')
-
-    def test_arm_quanta_gsj_initrd(self):
-        """
-        :avocado: tags=arch:arm
-        :avocado: tags=machine:quanta-gsj
-        :avocado: tags=accel:tcg
-        """
-        initrd_url = (
-                'https://github.com/hskinnemoen/openbmc/releases/download/'
-                '20200711-gsj-qemu-0/obmc-phosphor-initramfs-gsj.cpio.xz')
-        initrd_hash = '98fefe5d7e56727b1eb17d5c00311b1b5c945300'
-        initrd_path = self.fetch_asset(initrd_url, asset_hash=initrd_hash)
-        kernel_url = (
-                'https://github.com/hskinnemoen/openbmc/releases/download/'
-                '20200711-gsj-qemu-0/uImage-gsj.bin')
-        kernel_hash = 'fa67b2f141d56d39b3c54305c0e8a899c99eb2c7'
-        kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
-        dtb_url = (
-                'https://github.com/hskinnemoen/openbmc/releases/download/'
-                '20200711-gsj-qemu-0/nuvoton-npcm730-gsj.dtb')
-        dtb_hash = '18315f7006d7b688d8312d5c727eecd819aa36a4'
-        dtb_path = self.fetch_asset(dtb_url, asset_hash=dtb_hash)
-
-        self.vm.set_console()
-        kernel_command_line = (self.KERNEL_COMMON_COMMAND_LINE +
-                               'console=ttyS0,115200n8 '
-                               'earlycon=uart8250,mmio32,0xf0001000')
-        self.vm.add_args('-kernel', kernel_path,
-                         '-initrd', initrd_path,
-                         '-dtb', dtb_path,
-                         '-append', kernel_command_line)
-        self.vm.launch()
-
-        self.wait_for_console_pattern('Booting Linux on physical CPU 0x0')
-        self.wait_for_console_pattern('CPU1: thread -1, cpu 1, socket 0')
-        self.wait_for_console_pattern(
-                'Give root password for system maintenance')
@@ -1,37 +0,0 @@
-# Functional test that hotplugs a CPU and checks it on a Linux guest
-#
-# Copyright (c) 2021 Red Hat, Inc.
-#
-# Author:
-#  Cleber Rosa <crosa@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later.  See the COPYING file in the top-level directory.
-
-from avocado_qemu.linuxtest import LinuxTest
-
-
-class HotPlugCPU(LinuxTest):
-
-    def test(self):
-        """
-        :avocado: tags=arch:x86_64
-        :avocado: tags=machine:q35
-        :avocado: tags=accel:kvm
-        """
-        self.require_accelerator('kvm')
-        self.vm.add_args('-accel', 'kvm')
-        self.vm.add_args('-cpu', 'Haswell')
-        self.vm.add_args('-smp', '1,sockets=1,cores=2,threads=1,maxcpus=2')
-        self.launch_and_wait()
-
-        self.ssh_command('test -e /sys/devices/system/cpu/cpu0')
-        with self.assertRaises(AssertionError):
-            self.ssh_command('test -e /sys/devices/system/cpu/cpu1')
-
-        self.vm.cmd('device_add',
-                    driver='Haswell-x86_64-cpu',
-                    socket_id=0,
-                    core_id=1,
-                    thread_id=0)
-        self.ssh_command('test -e /sys/devices/system/cpu/cpu1')
@@ -1,122 +0,0 @@
-# INTEL_IOMMU Functional tests
-#
-# Copyright (c) 2021 Red Hat, Inc.
-#
-# Author:
-#  Eric Auger <eric.auger@redhat.com>
-#
-# This work is licensed under the terms of the GNU GPL, version 2 or
-# later.  See the COPYING file in the top-level directory.
-import os
-
-from avocado import skipUnless
-from avocado_qemu.linuxtest import LinuxTest
-
-@skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'), 'Test is unstable on GitLab')
-class IntelIOMMU(LinuxTest):
-    """
-    :avocado: tags=arch:x86_64
-    :avocado: tags=distro:fedora
-    :avocado: tags=distro_version:31
-    :avocado: tags=machine:q35
-    :avocado: tags=accel:kvm
-    :avocado: tags=intel_iommu
-    :avocado: tags=flaky
-    """
-
-    IOMMU_ADDON = ',iommu_platform=on,disable-modern=off,disable-legacy=on'
-    kernel_path = None
-    initrd_path = None
-    kernel_params = None
-
-    def set_up_boot(self):
-        path = self.download_boot()
-        self.vm.add_args('-device', 'virtio-blk-pci,bus=pcie.0,' +
-                         'drive=drv0,id=virtio-disk0,bootindex=1,'
-                         'werror=stop,rerror=stop' + self.IOMMU_ADDON)
-        self.vm.add_args('-device', 'virtio-gpu-pci' + self.IOMMU_ADDON)
-        self.vm.add_args('-drive',
-                         'file=%s,if=none,cache=writethrough,id=drv0' % path)
-
-    def setUp(self):
-        super(IntelIOMMU, self).setUp(None, 'virtio-net-pci' + self.IOMMU_ADDON)
-
-    def add_common_args(self):
-        self.vm.add_args('-device', 'virtio-rng-pci,rng=rng0')
-        self.vm.add_args('-object',
-                         'rng-random,id=rng0,filename=/dev/urandom')
-
-    def common_vm_setup(self, custom_kernel=None):
-        self.require_accelerator("kvm")
-        self.add_common_args()
-        self.vm.add_args("-accel", "kvm")
-
-        if custom_kernel is None:
-            return
-
-        kernel_url = self.distro.pxeboot_url + 'vmlinuz'
-        kernel_hash = '5b6f6876e1b5bda314f93893271da0d5777b1f3c'
-        initrd_url = self.distro.pxeboot_url + 'initrd.img'
-        initrd_hash = 'dd0340a1b39bd28f88532babd4581c67649ec5b1'
-        self.kernel_path = self.fetch_asset(kernel_url, asset_hash=kernel_hash)
-        self.initrd_path = self.fetch_asset(initrd_url, asset_hash=initrd_hash)
-
-    def run_and_check(self):
-        if self.kernel_path:
-            self.vm.add_args('-kernel', self.kernel_path,
-                             '-append', self.kernel_params,
-                             '-initrd', self.initrd_path)
-        self.launch_and_wait()
-        self.ssh_command('cat /proc/cmdline')
-        self.ssh_command('dmesg | grep -e DMAR -e IOMMU')
-        self.ssh_command('find /sys/kernel/iommu_groups/ -type l')
-        self.ssh_command('dnf -y install numactl-devel')
-
-    def test_intel_iommu(self):
-        """
-        :avocado: tags=intel_iommu_intremap
-        """
-
-        self.common_vm_setup(True)
-        self.vm.add_args('-device', 'intel-iommu,intremap=on')
-        self.vm.add_args('-machine', 'kernel_irqchip=split')
-
-        self.kernel_params = (self.distro.default_kernel_params +
-                              ' quiet intel_iommu=on')
-        self.run_and_check()
-
-    def test_intel_iommu_strict(self):
-        """
-        :avocado: tags=intel_iommu_strict
-        """
-
-        self.common_vm_setup(True)
-        self.vm.add_args('-device', 'intel-iommu,intremap=on')
-        self.vm.add_args('-machine', 'kernel_irqchip=split')
-        self.kernel_params = (self.distro.default_kernel_params +
-                              ' quiet intel_iommu=on,strict')
-        self.run_and_check()
-
-    def test_intel_iommu_strict_cm(self):
-        """
-        :avocado: tags=intel_iommu_strict_cm
-        """
-
-        self.common_vm_setup(True)
-        self.vm.add_args('-device', 'intel-iommu,intremap=on,caching-mode=on')
-        self.vm.add_args('-machine', 'kernel_irqchip=split')
-        self.kernel_params = (self.distro.default_kernel_params +
-                              ' quiet intel_iommu=on,strict')
-        self.run_and_check()
-
-    def test_intel_iommu_pt(self):
-        """
-        :avocado: tags=intel_iommu_pt
-        """
-
-        self.common_vm_setup(True)
-        self.vm.add_args('-device', 'intel-iommu,intremap=on')
-        self.vm.add_args('-machine', 'kernel_irqchip=split')
-        self.kernel_params = (self.distro.default_kernel_params +
-                              ' quiet intel_iommu=on iommu=pt')
-        self.run_and_check()
@@ -27,9 +27,11 @@ test_timeouts = {
  'arm_collie' : 180,
  'arm_cubieboard' : 360,
  'arm_orangepi' : 540,
+  'arm_quanta_gsj' : 240,
  'arm_raspi2' : 120,
  'arm_tuxrun' : 240,
  'arm_sx1' : 360,
+  'intel_iommu': 300,
  'mips_malta' : 120,
  'netdev_ethtool' : 180,
  'ppc_40p' : 240,
@@ -85,10 +87,12 @@ tests_arm_system_thorough = [
  'arm_emcraft_sf2',
  'arm_integratorcp',
  'arm_orangepi',
+  'arm_quanta_gsj',
  'arm_raspi2',
  'arm_smdkc210',
  'arm_sx1',
  'arm_vexpress',
+  'arm_virt',
  'arm_tuxrun',
 ]

@@ -224,11 +228,13 @@ tests_x86_64_system_quick = [

 tests_x86_64_system_thorough = [
  'acpi_bits',
-  'x86_64_tuxrun',
+  'intel_iommu',
  'linux_initrd',
  'multiprocess',
  'netdev_ethtool',
  'virtio_gpu',
+  'x86_64_hotplug_cpu',
+  'x86_64_tuxrun',
 ]

 tests_xtensa_system_thorough = [
@@ -8,8 +8,13 @@

 from .asset import Asset
 from .config import BUILD_DIR
-from .cmd import has_cmd, has_cmds, run_cmd, is_readable_executable_file, \
+from .cmd import is_readable_executable_file, \
    interrupt_interactive_console_until_pattern, wait_for_console_pattern, \
-    exec_command, exec_command_and_wait_for_pattern, get_qemu_img
+    exec_command, exec_command_and_wait_for_pattern, get_qemu_img, which
 from .testcase import QemuBaseTest, QemuUserTest, QemuSystemTest
 from .linuxkernel import LinuxKernelTest
+from .decorators import skipIfMissingCommands, skipIfNotMachine, \
+    skipFlakyTest, skipUntrustedTest, skipBigDataTest, \
+    skipIfMissingImports
+from .archive import archive_extract
+from .uncompress import uncompress
@@ -0,0 +1,117 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Utilities for python-based QEMU tests
+#
+# Copyright 2024 Red Hat, Inc.
+#
+# Authors:
+#  Thomas Huth <thuth@redhat.com>
+
+import os
+from subprocess import check_call, run, DEVNULL
+import tarfile
+from urllib.parse import urlparse
+import zipfile
+
+from .asset import Asset
+
+
+def tar_extract(archive, dest_dir, member=None):
+    with tarfile.open(archive) as tf:
+        if hasattr(tarfile, 'data_filter'):
+            tf.extraction_filter = getattr(tarfile, 'data_filter',
+                                           (lambda member, path: member))
+        if member:
+            tf.extract(member=member, path=dest_dir)
+        else:
+            tf.extractall(path=dest_dir)
+
+def cpio_extract(archive, output_path):
+    cwd = os.getcwd()
+    os.chdir(output_path)
+    # Not passing 'check=True' as cpio exits with non-zero
+    # status if the archive contains any device nodes :-(
+    if type(archive) == str:
+        run(['cpio', '-i', '-F', archive],
+            stdout=DEVNULL, stderr=DEVNULL)
+    else:
+        run(['cpio', '-i'],
+            input=archive.read(),
+            stdout=DEVNULL, stderr=DEVNULL)
+    os.chdir(cwd)
+
+def zip_extract(archive, dest_dir, member=None):
+    with zipfile.ZipFile(archive, 'r') as zf:
+        if member:
+            zf.extract(member=member, path=dest_dir)
+        else:
+            zf.extractall(path=dest_dir)
+
+def deb_extract(archive, dest_dir, member=None):
+    cwd = os.getcwd()
+    os.chdir(dest_dir)
+    try:
+        proc = run(['ar', 't', archive],
+                   check=True, capture_output=True, encoding='utf8')
+        file_path = proc.stdout.split()[2]
+        check_call(['ar', 'x', archive, file_path],
+                   stdout=DEVNULL, stderr=DEVNULL)
+        tar_extract(file_path, dest_dir, member)
+    finally:
+        os.chdir(cwd)
+
+'''
+@params archive: filename, Asset, or file-like object to extract
+@params dest_dir: target directory to extract into
+@params member: optional member file to limit extraction to
+
+Extracts @archive into @dest_dir. All files are extracted
+unless @member specifies a limit.
+
+If @format is None, heuristics will be applied to guess the format
+from the filename or Asset URL. @format must be non-None if @archive
+is a file-like object.
+'''
+def archive_extract(archive, dest_dir, format=None, member=None):
+    if format is None:
+        format = guess_archive_format(archive)
+    if type(archive) == Asset:
+        archive = str(archive)
+
+    if format == "tar":
+        tar_extract(archive, dest_dir, member)
+    elif format == "zip":
+        zip_extract(archive, dest_dir, member)
+    elif format == "cpio":
+        if member is not None:
+            raise Exception("Unable to filter cpio extraction")
+        cpio_extract(archive, dest_dir)
+    elif format == "deb":
+        if type(archive) != str:
+            raise Exception("Unable to use file-like object with deb archives")
+        deb_extract(archive, dest_dir, "./" + member)
+    else:
+        raise Exception(f"Unknown archive format {format}")
+
+'''
+@params archive: filename, or Asset to guess
+
+Guess the format of @compressed, raising an exception if
+no format can be determined
+'''
+def guess_archive_format(archive):
+    if type(archive) == Asset:
+        archive = urlparse(archive.url).path
+    elif type(archive) != str:
+        raise Exception(f"Unable to guess archive format for {archive}")
+
+    if ".tar." in archive or archive.endswith("tgz"):
+        return "tar"
+    elif archive.endswith(".zip"):
+        return "zip"
+    elif archive.endswith(".cpio"):
+        return "cpio"
+    elif archive.endswith(".deb") or archive.endswith(".udeb"):
+        return "deb"
+    else:
+        raise Exception(f"Unknown archive format for {archive}")
@@ -9,13 +9,13 @@ import hashlib
 import logging
 import os
 import stat
-import subprocess
 import sys
 import unittest
 import urllib.request
 from time import sleep
 from pathlib import Path
 from shutil import copyfileobj
+from urllib.error import HTTPError


 # Instances of this class must be declared as class level variables
@@ -40,6 +40,9 @@ class Asset:
        return "Asset: url=%s hash=%s cache=%s" % (
            self.url, self.hash, self.cache_file)

+    def __str__(self):
+        return str(self.cache_file)
+
    def _check(self, cache_file):
        if self.hash is None:
            return True
@@ -63,6 +66,12 @@ class Asset:
    def valid(self):
        return self.cache_file.exists() and self._check(self.cache_file)

+    def fetchable(self):
+        return not os.environ.get("QEMU_TEST_NO_DOWNLOAD", False)
+
+    def available(self):
+        return self.valid() or self.fetchable()
+
    def _wait_for_other_download(self, tmp_cache_file):
        # Another thread already seems to download the asset, so wait until
        # it is done, while also checking the size to see whether it is stuck
@@ -101,7 +110,7 @@ class Asset:
                           self.cache_file, self.url)
            return str(self.cache_file)

-        if os.environ.get("QEMU_TEST_NO_DOWNLOAD", False):
+        if not self.fetchable():
            raise Exception("Asset cache is invalid and downloads disabled")

        self.log.info("Downloading %s to %s...", self.url, self.cache_file)
@@ -162,7 +171,18 @@ class Asset:
        for name, asset in vars(test.__class__).items():
            if name.startswith("ASSET_") and type(asset) == Asset:
                log.info("Attempting to cache '%s'" % asset)
-                asset.fetch()
+                try:
+                    asset.fetch()
+                except HTTPError as e:
+                    # Treat 404 as fatal, since it is highly likely to
+                    # indicate a broken test rather than a transient
+                    # server or networking problem
+                    if e.code == 404:
+                        raise
+
+                    log.debug(f"HTTP error {e.code} from {asset.url} " +
+                              "skipping asset precache")
+
        log.removeHandler(handler)

    def precache_suite(suite):
@@ -14,66 +14,18 @@
 import logging
 import os
 import os.path
-import subprocess
-
-from .config import BUILD_DIR


-def has_cmd(name, args=None):
+def which(tool):
+    """ looks up the full path for @tool, returns None if not found
+        or if @tool does not have executable permissions.
    """
-    This function is for use in a @skipUnless decorator, e.g.:
-
-        @skipUnless(*has_cmd('sudo -n', ('sudo', '-n', 'true')))
-        def test_something_that_needs_sudo(self):
-            ...
-    """
-
-    if args is None:
-        args = ('which', name)
-
-    try:
-        _, stderr, exitcode = run_cmd(args)
-    except Exception as e:
-        exitcode = -1
-        stderr = str(e)
-
-    if exitcode != 0:
-        cmd_line = ' '.join(args)
-        err = f'{name} required, but "{cmd_line}" failed: {stderr.strip()}'
-        return (False, err)
-    else:
-        return (True, '')
-
-def has_cmds(*cmds):
-    """
-    This function is for use in a @skipUnless decorator and
-    allows checking for the availability of multiple commands, e.g.:
-
-        @skipUnless(*has_cmds(('cmd1', ('cmd1', '--some-parameter')),
-                              'cmd2', 'cmd3'))
-        def test_something_that_needs_cmd1_and_cmd2(self):
-            ...
-    """
-
-    for cmd in cmds:
-        if isinstance(cmd, str):
-            cmd = (cmd,)
-
-        ok, errstr = has_cmd(*cmd)
-        if not ok:
-            return (False, errstr)
-
-    return (True, '')
-
-def run_cmd(args):
-    subp = subprocess.Popen(args,
-                            stdout=subprocess.PIPE,
-                            stderr=subprocess.PIPE,
-                            universal_newlines=True)
-    stdout, stderr = subp.communicate()
-    ret = subp.returncode
-
-    return (stdout, stderr, ret)
+    paths=os.getenv('PATH')
+    for p in paths.split(os.path.pathsep):
+        p = os.path.join(p, tool)
+        if os.access(p, os.X_OK):
+            return p
+    return None

 def is_readable_executable_file(path):
    return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
@@ -241,10 +193,10 @@ def get_qemu_img(test):

    # If qemu-img has been built, use it, otherwise the system wide one
    # will be used.
-    qemu_img = os.path.join(BUILD_DIR, 'qemu-img')
+    qemu_img = test.build_file('qemu-img')
    if os.path.exists(qemu_img):
        return qemu_img
-    (has_system_qemu_img, errmsg) = has_cmd('qemu-img')
-    if has_system_qemu_img:
-        return 'qemu-img'
-    test.skipTest(errmsg)
+    qemu_img = which('qemu-img')
+    if qemu_img is not None:
+        return qemu_img
+    test.skipTest(f"qemu-img not found in build dir or '$PATH'")
@@ -0,0 +1,107 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Decorators useful in functional tests
+
+import os
+import platform
+from unittest import skipUnless
+
+from .cmd import which
+
+'''
+Decorator to skip execution of a test if the list
+of command binaries is not available in $PATH.
+Example:
+
+  @skipIfMissingCommands("mkisofs", "losetup")
+'''
+def skipIfMissingCommands(*args):
+    def has_cmds(cmdlist):
+        for cmd in cmdlist:
+            if not which(cmd):
+                return False
+        return True
+
+    return skipUnless(lambda: has_cmds(args),
+                      'required command(s) "%s" not installed' %
+                      ", ".join(args))
+
+'''
+Decorator to skip execution of a test if the current
+host machine does not match one of the permitted
+machines.
+Example
+
+  @skipIfNotMachine("x86_64", "aarch64")
+'''
+def skipIfNotMachine(*args):
+    return skipUnless(lambda: platform.machine() in args,
+                        'not running on one of the required machine(s) "%s"' %
+                        ", ".join(args))
+
+'''
+Decorator to skip execution of flaky tests, unless
+the $QEMU_TEST_FLAKY_TESTS environment variable is set.
+A bug URL must be provided that documents the observed
+failure behaviour, so it can be tracked & re-evaluated
+in future.
+
+Historical tests may be providing "None" as the bug_url
+but this should not be done for new test.
+
+Example:
+
+  @skipFlakyTest("https://gitlab.com/qemu-project/qemu/-/issues/NNN")
+'''
+def skipFlakyTest(bug_url):
+    if bug_url is None:
+        bug_url = "FIXME: reproduce flaky test and file bug report or remove"
+    return skipUnless(os.getenv('QEMU_TEST_FLAKY_TESTS'),
+                      f'Test is unstable: {bug_url}')
+
+'''
+Decorator to skip execution of tests which are likely
+to execute untrusted commands on the host, or commands
+which process untrusted code, unless the
+$QEMU_TEST_ALLOW_UNTRUSTED_CODE env var is set.
+Example:
+
+  @skipUntrustedTest()
+'''
+def skipUntrustedTest():
+    return skipUnless(os.getenv('QEMU_TEST_ALLOW_UNTRUSTED_CODE'),
+                      'Test runs untrusted code / processes untrusted data')
+
+'''
+Decorator to skip execution of tests which need large
+data storage (over around 500MB-1GB mark) on the host,
+unless the $QEMU_TEST_ALLOW_LARGE_STORAGE environment
+variable is set
+
+Example:
+
+  @skipBigDataTest()
+'''
+def skipBigDataTest():
+    return skipUnless(os.getenv('QEMU_TEST_ALLOW_LARGE_STORAGE'),
+                      'Test requires large host storage space')
+
+'''
+Decorator to skip execution of a test if the list
+of python imports is not available.
+Example:
+
+  @skipIfMissingImports("numpy", "cv2")
+'''
+def skipIfMissingImports(*args):
+    def has_imports(importlist):
+        for impname in importlist:
+            try:
+                import impname
+            except ImportError:
+                return False
+        return True
+
+    return skipUnless(lambda: has_imports(args),
+                      'required import(s) "%s" not installed' %
+                      ", ".join(args))
@@ -3,11 +3,9 @@
 # This work is licensed under the terms of the GNU GPL, version 2 or
 # later.  See the COPYING file in the top-level directory.

-import os
-
 from .testcase import QemuSystemTest
-from .cmd import run_cmd, wait_for_console_pattern
-from .utils import archive_extract
+from .cmd import wait_for_console_pattern
+

 class LinuxKernelTest(QemuSystemTest):
    KERNEL_COMMON_COMMAND_LINE = 'printk.time=0 '
@@ -28,26 +26,3 @@ class LinuxKernelTest(QemuSystemTest):
        self.vm.launch()
        if wait_for:
                self.wait_for_console_pattern(wait_for)
-
-    def extract_from_deb(self, deb_path, path):
-        """
-        Extracts a file from a deb package into the test workdir
-
-        :param deb_path: path to the deb archive
-        :param path: path within the deb archive of the file to be extracted
-        :returns: path of the extracted file
-        """
-        cwd = os.getcwd()
-        os.chdir(self.workdir)
-        (stdout, stderr, ret) = run_cmd(['ar', 't', deb_path])
-        file_path = stdout.split()[2]
-        run_cmd(['ar', 'x', deb_path, file_path])
-        archive_extract(file_path, self.workdir)
-        os.chdir(cwd)
-        # Return complete path to extracted file.  Because callers to
-        # extract_from_deb() specify 'path' with a leading slash, it is
-        # necessary to use os.path.relpath() as otherwise os.path.join()
-        # interprets it as an absolute path and drops the self.workdir part.
-        return os.path.normpath(os.path.join(self.workdir,
-                                             os.path.relpath(path, '/')))
-
@@ -5,30 +5,19 @@
 # This work is licensed under the terms of the GNU GPL, version 2 or
 # later. See the COPYING file in the top-level directory.

-import re
 import logging
+from subprocess import run

-from . import has_cmd, run_cmd
-
-def tesseract_available(expected_version):
-    (has_tesseract, _) = has_cmd('tesseract')
-    if not has_tesseract:
-        return False
-    (stdout, stderr, ret) = run_cmd([ 'tesseract', '--version'])
-    if ret:
-        return False
-    version = stdout.split()[1]
-    return int(version.split('.')[0]) >= expected_version

 def tesseract_ocr(image_path, tesseract_args=''):
    console_logger = logging.getLogger('console')
    console_logger.debug(image_path)
-    (stdout, stderr, ret) = run_cmd(['tesseract', image_path,
-                                     'stdout'])
-    if ret:
+    proc = run(['tesseract', image_path, 'stdout'],
+               capture_output=True, encoding='utf8')
+    if proc.returncode:
        return None
    lines = []
-    for line in stdout.split('\n'):
+    for line in proc.stdout.split('\n'):
        sline = line.strip()
        if len(sline):
            console_logger.debug(sline)
@@ -13,19 +13,22 @@

 import logging
 import os
+from pathlib import Path
 import pycotap
 import shutil
-import subprocess
+from subprocess import run
 import sys
+import tempfile
 import unittest
 import uuid

 from qemu.machine import QEMUMachine
 from qemu.utils import kvm_available, tcg_available

+from .archive import archive_extract
 from .asset import Asset
-from .cmd import run_cmd
 from .config import BUILD_DIR
+from .uncompress import uncompress


 class QemuBaseTest(unittest.TestCase):
@@ -37,17 +40,169 @@ class QemuBaseTest(unittest.TestCase):
    log = None
    logdir = None

+    '''
+    @params compressed: filename, Asset, or file-like object to uncompress
+    @params format: optional compression format (gzip, lzma)
+
+    Uncompresses @compressed into the scratch directory.
+
+    If @format is None, heuristics will be applied to guess the format
+    from the filename or Asset URL. @format must be non-None if @uncompressed
+    is a file-like object.
+
+    Returns the fully qualified path to the uncompressed file
+    '''
+    def uncompress(self, compressed, format=None):
+        self.log.debug(f"Uncompress {compressed} format={format}")
+        if type(compressed) == Asset:
+            compressed.fetch()
+
+        (name, ext) = os.path.splitext(str(compressed))
+        uncompressed = self.scratch_file(os.path.basename(name))
+
+        uncompress(compressed, uncompressed, format)
+
+        return uncompressed
+
+    '''
+    @params archive: filename, Asset, or file-like object to extract
+    @params format: optional archive format (tar, zip, deb, cpio)
+    @params sub_dir: optional sub-directory to extract into
+    @params member: optional member file to limit extraction to
+
+    Extracts @archive into the scratch directory, or a directory beneath
+    named by @sub_dir. All files are extracted unless @member specifies
+    a limit.
+
+    If @format is None, heuristics will be applied to guess the format
+    from the filename or Asset URL. @format must be non-None if @archive
+    is a file-like object.
+
+    If @member is non-None, returns the fully qualified path to @member
+    '''
+    def archive_extract(self, archive, format=None, sub_dir=None, member=None):
+        self.log.debug(f"Extract {archive} format={format}" +
+                       f"sub_dir={sub_dir} member={member}")
+        if type(archive) == Asset:
+            archive.fetch()
+        if sub_dir is None:
+            archive_extract(archive, self.scratch_file(), format, member)
+        else:
+            archive_extract(archive, self.scratch_file(sub_dir),
+                            format, member)
+
+        if member is not None:
+            return self.scratch_file(member)
+        return None
+
+    '''
+    Create a temporary directory suitable for storing UNIX
+    socket paths.
+
+    Returns: a tempfile.TemporaryDirectory instance
+    '''
+    def socket_dir(self):
+        if self.socketdir is None:
+            self.socketdir = tempfile.TemporaryDirectory(
+                prefix="qemu_func_test_sock_")
+        return self.socketdir
+
+    '''
+    @params args list of zero or more subdirectories or file
+
+    Construct a path for accessing a data file located
+    relative to the source directory that is the root for
+    functional tests.
+
+    @args may be an empty list to reference the root dir
+    itself, may be a single element to reference a file in
+    the root directory, or may be multiple elements to
+    reference a file nested below. The path components
+    will be joined using the platform appropriate path
+    separator.
+
+    Returns: string representing a file path
+    '''
+    def data_file(self, *args):
+        return str(Path(Path(__file__).parent.parent, *args))
+
+    '''
+    @params args list of zero or more subdirectories or file
+
+    Construct a path for accessing a data file located
+    relative to the build directory root.
+
+    @args may be an empty list to reference the build dir
+    itself, may be a single element to reference a file in
+    the build directory, or may be multiple elements to
+    reference a file nested below. The path components
+    will be joined using the platform appropriate path
+    separator.
+
+    Returns: string representing a file path
+    '''
+    def build_file(self, *args):
+        return str(Path(BUILD_DIR, *args))
+
+    '''
+    @params args list of zero or more subdirectories or file
+
+    Construct a path for accessing/creating a scratch file
+    located relative to a temporary directory dedicated to
+    this test case. The directory and its contents will be
+    purged upon completion of the test.
+
+    @args may be an empty list to reference the scratch dir
+    itself, may be a single element to reference a file in
+    the scratch directory, or may be multiple elements to
+    reference a file nested below. The path components
+    will be joined using the platform appropriate path
+    separator.
+
+    Returns: string representing a file path
+    '''
+    def scratch_file(self, *args):
+        return str(Path(self.workdir, *args))
+
+    '''
+    @params args list of zero or more subdirectories or file
+
+    Construct a path for accessing/creating a log file
+    located relative to a temporary directory dedicated to
+    this test case. The directory and its log files will be
+    preserved upon completion of the test.
+
+    @args may be an empty list to reference the log dir
+    itself, may be a single element to reference a file in
+    the log directory, or may be multiple elements to
+    reference a file nested below. The path components
+    will be joined using the platform appropriate path
+    separator.
+
+    Returns: string representing a file path
+    '''
+    def log_file(self, *args):
+        return str(Path(self.outputdir, *args))
+
+    def assets_available(self):
+        for name, asset in vars(self.__class__).items():
+            if name.startswith("ASSET_") and type(asset) == Asset:
+                if not asset.available():
+                    self.log.debug(f"Asset {asset.url} not available")
+                    return False
+        return True
+
    def setUp(self, bin_prefix):
        self.assertIsNotNone(self.qemu_bin, 'QEMU_TEST_QEMU_BINARY must be set')
        self.arch = self.qemu_bin.split('-')[-1]
+        self.socketdir = None

-        self.outputdir = os.path.join(BUILD_DIR, 'tests', 'functional',
-                                      self.arch, self.id())
+        self.outputdir = self.build_file('tests', 'functional',
+                                         self.arch, self.id())
        self.workdir = os.path.join(self.outputdir, 'scratch')
        os.makedirs(self.workdir, exist_ok=True)

-        self.logdir = self.outputdir
-        self.log_filename = os.path.join(self.logdir, 'base.log')
+        self.log_filename = self.log_file('base.log')
        self.log = logging.getLogger('qemu-test')
        self.log.setLevel(logging.DEBUG)
        self._log_fh = logging.FileHandler(self.log_filename, mode='w')
@@ -62,9 +217,15 @@ class QemuBaseTest(unittest.TestCase):
        self.machinelog.setLevel(logging.DEBUG)
        self.machinelog.addHandler(self._log_fh)

+        if not self.assets_available():
+            self.skipTest('One or more assets is not available')
+
    def tearDown(self):
        if "QEMU_TEST_KEEP_SCRATCH" not in os.environ:
            shutil.rmtree(self.workdir)
+        if self.socketdir is not None:
+            shutil.rmtree(self.socketdir.name)
+            self.socketdir = None
        self.machinelog.removeHandler(self._log_fh)
        self.log.removeHandler(self._log_fh)

@@ -100,11 +261,11 @@ class QemuUserTest(QemuBaseTest):
        self._ldpath.append(os.path.abspath(ldpath))

    def run_cmd(self, bin_path, args=[]):
-        return subprocess.run([self.qemu_bin]
-                              + ["-L %s" % ldpath for ldpath in self._ldpath]
-                              + [bin_path]
-                              + args,
-                              text=True, capture_output=True)
+        return run([self.qemu_bin]
+                   + ["-L %s" % ldpath for ldpath in self._ldpath]
+                   + [bin_path]
+                   + args,
+                   text=True, capture_output=True)

 class QemuSystemTest(QemuBaseTest):
    """Facilitates system emulation tests."""
@@ -120,7 +281,7 @@ class QemuSystemTest(QemuBaseTest):

        console_log = logging.getLogger('console')
        console_log.setLevel(logging.DEBUG)
-        self.console_log_name = os.path.join(self.logdir, 'console.log')
+        self.console_log_name = self.log_file('console.log')
        self._console_log_fh = logging.FileHandler(self.console_log_name,
                                                   mode='w')
        self._console_log_fh.setLevel(logging.DEBUG)
@@ -131,7 +292,9 @@ class QemuSystemTest(QemuBaseTest):
    def set_machine(self, machinename):
        # TODO: We should use QMP to get the list of available machines
        if not self._machinehelp:
-            self._machinehelp = run_cmd([self.qemu_bin, '-M', 'help'])[0];
+            self._machinehelp = run(
+                [self.qemu_bin, '-M', 'help'],
+                capture_output=True, check=True, encoding='utf8').stdout
        if self._machinehelp.find(machinename) < 0:
            self.skipTest('no support for machine ' + machinename)
        self.machine = machinename
@@ -159,22 +322,24 @@ class QemuSystemTest(QemuBaseTest):
                          "available" % accelerator)

    def require_netdev(self, netdevname):
-        netdevhelp = run_cmd([self.qemu_bin,
-                             '-M', 'none', '-netdev', 'help'])[0];
-        if netdevhelp.find('\n' + netdevname + '\n') < 0:
+        help = run([self.qemu_bin,
+                    '-M', 'none', '-netdev', 'help'],
+                   capture_output=True, check=True, encoding='utf8').stdout;
+        if help.find('\n' + netdevname + '\n') < 0:
            self.skipTest('no support for " + netdevname + " networking')

    def require_device(self, devicename):
-        devhelp = run_cmd([self.qemu_bin,
-                           '-M', 'none', '-device', 'help'])[0];
-        if devhelp.find(devicename) < 0:
+        help = run([self.qemu_bin,
+                    '-M', 'none', '-device', 'help'],
+                   capture_output=True, check=True, encoding='utf8').stdout;
+        if help.find(devicename) < 0:
            self.skipTest('no support for device ' + devicename)

    def _new_vm(self, name, *args):
        vm = QEMUMachine(self.qemu_bin,
                         name=name,
                         base_temp_dir=self.workdir,
-                         log_dir=self.logdir)
+                         log_dir=self.log_file())
        self.log.debug('QEMUMachine "%s" created', name)
        self.log.debug('QEMUMachine "%s" temp_dir: %s', name, vm.temp_dir)

@@ -11,12 +11,12 @@

 import os
 import stat
-import time
+from subprocess import check_call, DEVNULL

 from qemu_test import QemuSystemTest
-from qemu_test import exec_command, exec_command_and_wait_for_pattern
+from qemu_test import exec_command_and_wait_for_pattern
 from qemu_test import wait_for_console_pattern
-from qemu_test import has_cmd, run_cmd, get_qemu_img
+from qemu_test import which, get_qemu_img

 class TuxRunBaselineTest(QemuSystemTest):

@@ -39,10 +39,8 @@ class TuxRunBaselineTest(QemuSystemTest):
        super().setUp()

        # We need zstd for all the tuxrun tests
-        (has_zstd, msg) = has_cmd('zstd')
-        if has_zstd is False:
-            self.skipTest(msg)
-        self.zstd = 'zstd'
+        if which('zstd') is None:
+            self.skipTest("zstd not found in $PATH")

        # Pre-init TuxRun specific settings: Most machines work with
        # reasonable defaults but we sometimes need to tweak the
@@ -77,10 +75,11 @@ class TuxRunBaselineTest(QemuSystemTest):
        kernel_image =  kernel_asset.fetch()
        disk_image_zst = rootfs_asset.fetch()

-        disk_image = self.workdir + "/rootfs.ext4"
+        disk_image = self.scratch_file("rootfs.ext4")

-        run_cmd([self.zstd, "-f", "-d", disk_image_zst,
-                 "-o", disk_image])
+        check_call(['zstd', "-f", "-d", disk_image_zst,
+                    "-o", disk_image],
+                   stdout=DEVNULL, stderr=DEVNULL)
        # zstd copies source archive permissions for the output
        # file, so must make this writable for QEMU
        os.chmod(disk_image, stat.S_IRUSR | stat.S_IWUSR)
@@ -0,0 +1,83 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Utilities for python-based QEMU tests
+#
+# Copyright 2024 Red Hat, Inc.
+#
+# Authors:
+#  Thomas Huth <thuth@redhat.com>
+
+import gzip
+import lzma
+import os
+import shutil
+from urllib.parse import urlparse
+
+from .asset import Asset
+
+
+def gzip_uncompress(gz_path, output_path):
+    if os.path.exists(output_path):
+        return
+    with gzip.open(gz_path, 'rb') as gz_in:
+        try:
+            with open(output_path, 'wb') as raw_out:
+                shutil.copyfileobj(gz_in, raw_out)
+        except:
+            os.remove(output_path)
+            raise
+
+def lzma_uncompress(xz_path, output_path):
+    if os.path.exists(output_path):
+        return
+    with lzma.open(xz_path, 'rb') as lzma_in:
+        try:
+            with open(output_path, 'wb') as raw_out:
+                shutil.copyfileobj(lzma_in, raw_out)
+        except:
+            os.remove(output_path)
+            raise
+
+'''
+@params compressed: filename, Asset, or file-like object to uncompress
+@params uncompressed: filename to uncompress into
+@params format: optional compression format (gzip, lzma)
+
+Uncompresses @compressed into @uncompressed
+
+If @format is None, heuristics will be applied to guess the format
+from the filename or Asset URL. @format must be non-None if @uncompressed
+is a file-like object.
+
+Returns the fully qualified path to the uncompessed file
+'''
+def uncompress(compressed, uncompressed, format=None):
+    if format is None:
+        format = guess_uncompress_format(compressed)
+
+    if format == "xz":
+        lzma_uncompress(str(compressed), uncompressed)
+    elif format == "gz":
+        gzip_uncompress(str(compressed), uncompressed)
+    else:
+        raise Exception(f"Unknown compression format {format}")
+
+'''
+@params compressed: filename, Asset, or file-like object to guess
+
+Guess the format of @compressed, raising an exception if
+no format can be determined
+'''
+def guess_uncompress_format(compressed):
+    if type(compressed) == Asset:
+        compressed = urlparse(compressed.url).path
+    elif type(compressed) != str:
+        raise Exception(f"Unable to guess compression cformat for {compressed}")
+
+    (name, ext) = os.path.splitext(compressed)
+    if ext == ".xz":
+        return "xz"
+    elif ext == ".gz":
+        return "gz"
+    else:
+        raise Exception(f"Unknown compression format for {compressed}")
@@ -8,12 +8,14 @@
 # This work is licensed under the terms of the GNU GPL, version 2 or
 # later.  See the COPYING file in the top-level directory.

-import gzip
-import lzma
 import os
-import shutil
-import subprocess
-import tarfile
+
+from qemu.utils import get_info_usernet_hostfwd_port
+
+
+def get_usernet_hostfwd_port(vm):
+    res = vm.cmd('human-monitor-command', command_line='info usernet')
+    return get_info_usernet_hostfwd_port(res)

 """
 Round up to next power of 2
@@ -35,43 +37,3 @@ def image_pow2ceil_expand(path):
        if size != size_aligned:
            with open(path, 'ab+') as fd:
                fd.truncate(size_aligned)
-
-def archive_extract(archive, dest_dir, member=None):
-    with tarfile.open(archive) as tf:
-        if hasattr(tarfile, 'data_filter'):
-            tf.extraction_filter = getattr(tarfile, 'data_filter',
-                                           (lambda member, path: member))
-        if member:
-            tf.extract(member=member, path=dest_dir)
-        else:
-            tf.extractall(path=dest_dir)
-
-def gzip_uncompress(gz_path, output_path):
-    if os.path.exists(output_path):
-        return
-    with gzip.open(gz_path, 'rb') as gz_in:
-        try:
-            with open(output_path, 'wb') as raw_out:
-                shutil.copyfileobj(gz_in, raw_out)
-        except:
-            os.remove(output_path)
-            raise
-
-def lzma_uncompress(xz_path, output_path):
-    if os.path.exists(output_path):
-        return
-    with lzma.open(xz_path, 'rb') as lzma_in:
-        try:
-            with open(output_path, 'wb') as raw_out:
-                shutil.copyfileobj(lzma_in, raw_out)
-        except:
-            os.remove(output_path)
-            raise
-
-def cpio_extract(cpio_handle, output_path):
-    cwd = os.getcwd()
-    os.chdir(output_path)
-    subprocess.run(['cpio', '-i'],
-                   input=cpio_handle.read(),
-                   stderr=subprocess.DEVNULL)
-    os.chdir(cwd)
@@ -6,13 +6,12 @@
 #
 # SPDX-License-Identifier: GPL-2.0-or-later

-import sys
 import os

 from qemu_test import QemuSystemTest, Asset
 from qemu_test import wait_for_console_pattern
 from qemu_test import exec_command_and_wait_for_pattern
-from qemu_test.utils import archive_extract
+

 class AST2x00MachineSDK(QemuSystemTest):

@@ -35,30 +34,31 @@ class AST2x00MachineSDK(QemuSystemTest):
    def test_aarch64_ast2700_evb_sdk_v09_02(self):
        self.set_machine('ast2700-evb')

-        image_path = self.ASSET_SDK_V902_AST2700.fetch()
-        archive_extract(image_path, self.workdir)
+        self.archive_extract(self.ASSET_SDK_V902_AST2700)

        num_cpu = 4
-        image_dir = self.workdir + '/ast2700-default/'
-        uboot_size = os.path.getsize(image_dir + 'u-boot-nodtb.bin')
+        uboot_size = os.path.getsize(self.scratch_file('ast2700-default',
+                                                       'u-boot-nodtb.bin'))
        uboot_dtb_load_addr = hex(0x400000000 + uboot_size)

        load_images_list = [
            {
                'addr': '0x400000000',
-                'file': image_dir + 'u-boot-nodtb.bin'
+                'file': self.scratch_file('ast2700-default',
+                                          'u-boot-nodtb.bin')
            },
            {
                'addr': str(uboot_dtb_load_addr),
-                'file': image_dir + 'u-boot.dtb'
+                'file': self.scratch_file('ast2700-default', 'u-boot.dtb')
            },
            {
                'addr': '0x430000000',
-                'file': image_dir + 'bl31.bin'
+                'file': self.scratch_file('ast2700-default', 'bl31.bin')
            },
            {
                'addr': '0x430080000',
-                'file': image_dir + 'optee/tee-raw.bin'
+                'file': self.scratch_file('ast2700-default', 'optee',
+                                          'tee-raw.bin')
            }
        ]

@@ -75,7 +75,8 @@ class AST2x00MachineSDK(QemuSystemTest):
        self.vm.add_args('-smp', str(num_cpu))
        self.vm.add_args('-device',
                         'tmp105,bus=aspeed.i2c.bus.1,address=0x4d,id=tmp-test')
-        self.do_test_aarch64_aspeed_sdk_start(image_dir + 'image-bmc')
+        self.do_test_aarch64_aspeed_sdk_start(
+            self.scratch_file('ast2700-default', 'image-bmc'))

        wait_for_console_pattern(self, 'ast2700-default login:')

@@ -7,9 +7,6 @@
 #
 # SPDX-License-Identifier: GPL-2.0-or-later

-import os
-from zipfile import ZipFile
-
 from qemu_test import LinuxKernelTest, Asset


@@ -22,11 +19,7 @@ class Aarch64Raspi3Machine(LinuxKernelTest):

    def test_aarch64_raspi3_atf(self):
        efi_name = 'RPI_EFI.fd'
-        zip_path = self.ASSET_RPI3_UEFI.fetch()
-
-        with ZipFile(zip_path, 'r') as zf:
-                     zf.extract(efi_name, path=self.workdir)
-        efi_fd = os.path.join(self.workdir, efi_name)
+        efi_fd = self.archive_extract(self.ASSET_RPI3_UEFI, member=efi_name)

        self.set_machine('raspi3b')
        self.vm.set_console(console_index=1)
@@ -5,11 +5,8 @@
 #
 # SPDX-License-Identifier: GPL-2.0-or-later

-import os
-
 from qemu_test import LinuxKernelTest, Asset
 from qemu_test import exec_command_and_wait_for_pattern
-from qemu_test.utils import gzip_uncompress


 class Aarch64Raspi4Machine(LinuxKernelTest):
@@ -32,9 +29,10 @@ class Aarch64Raspi4Machine(LinuxKernelTest):
        '7c0b16d1853772f6f4c3ca63e789b3b9ff4936efac9c8a01fb0c98c05c7a7648')

    def test_arm_raspi4(self):
-        deb_path = self.ASSET_KERNEL_20190215.fetch()
-        kernel_path = self.extract_from_deb(deb_path, '/boot/kernel8.img')
-        dtb_path = self.extract_from_deb(deb_path, '/boot/bcm2711-rpi-4-b.dtb')
+        kernel_path = self.archive_extract(self.ASSET_KERNEL_20190215,
+                                           member='boot/kernel8.img')
+        dtb_path = self.archive_extract(self.ASSET_KERNEL_20190215,
+                                        member='boot/bcm2711-rpi-4-b.dtb')

        self.set_machine('raspi4b')
        self.vm.set_console()
@@ -60,12 +58,11 @@ class Aarch64Raspi4Machine(LinuxKernelTest):


    def test_arm_raspi4_initrd(self):
-        deb_path = self.ASSET_KERNEL_20190215.fetch()
-        kernel_path = self.extract_from_deb(deb_path, '/boot/kernel8.img')
-        dtb_path = self.extract_from_deb(deb_path, '/boot/bcm2711-rpi-4-b.dtb')
-        initrd_path_gz = self.ASSET_INITRD.fetch()
-        initrd_path = os.path.join(self.workdir, 'rootfs.cpio')
-        gzip_uncompress(initrd_path_gz, initrd_path)
+        kernel_path = self.archive_extract(self.ASSET_KERNEL_20190215,
+                                           member='boot/kernel8.img')
+        dtb_path = self.archive_extract(self.ASSET_KERNEL_20190215,
+                                        member='boot/bcm2711-rpi-4-b.dtb')
+        initrd_path = self.uncompress(self.ASSET_INITRD)

        self.set_machine('raspi4b')
        self.vm.set_console()
--- a/Show More
+++ b/Show More