diff --git a/tests/functional/aarch64/test_device_passthrough.py b/tests/functional/aarch64/test_device_passthrough.py index 05a3f52d5e..10c73728f3 100755 --- a/tests/functional/aarch64/test_device_passthrough.py +++ b/tests/functional/aarch64/test_device_passthrough.py @@ -10,13 +10,13 @@ # SPDX-License-Identifier: GPL-2.0-or-later from os.path import join - -from qemu_test import QemuSystemTest, Asset -from qemu_test import exec_command, wait_for_console_pattern -from qemu_test import exec_command_and_wait_for_pattern from random import randbytes -guest_script = ''' +from qemu_test import QemuSystemTest, Asset +from qemu_test import wait_for_console_pattern + + +GUEST_SCRIPT = ''' #!/usr/bin/env bash set -euo pipefail @@ -56,7 +56,7 @@ qemu-system-aarch64 \ -device vfio-pci,host=$pci_iommufd,iommufd=iommufd0 ''' -nested_guest_script = ''' +NESTED_GUEST_SCRIPT = ''' #!/usr/bin/env bash set -euo pipefail @@ -75,6 +75,7 @@ diff $image_iommufd /dev/$dev_iommufd echo device_passthrough_test_ok ''' + class Aarch64DevicePassthrough(QemuSystemTest): # https://github.com/pbo-linaro/qemu-linux-stack/tree/device_passthrough @@ -86,7 +87,7 @@ class Aarch64DevicePassthrough(QemuSystemTest): ASSET_DEVICE_PASSTHROUGH_STACK = Asset( ('https://github.com/pbo-linaro/qemu-linux-stack/' 'releases/download/build/device_passthrough-a9612a2.tar.xz'), - 'f7d2f70912e7231986e6e293e1a2c4786dd02bec113a7acb6bfc619e96155455') + 'f7d2f70912e7231986e6e293e1a2c4786dd02bec113a7acb6bfc619e96155455') # This tests the device passthrough implementation, by booting a VM # supporting it with two nvme disks attached, and launching a nested VM @@ -108,10 +109,14 @@ class Aarch64DevicePassthrough(QemuSystemTest): guest_cmd = join(stack, 'guest.sh') nested_guest_cmd = join(stack, 'nested_guest.sh') # we generate two random disks - with open(disk_vfio, "wb") as d: d.write(randbytes(512)) - with open(disk_iommufd, "wb") as d: d.write(randbytes(1024)) - with open(guest_cmd, 'w') as s: s.write(guest_script) - with open(nested_guest_cmd, 'w') as s: s.write(nested_guest_script) + with open(disk_vfio, "wb") as d: + d.write(randbytes(512)) + with open(disk_iommufd, "wb") as d: + d.write(randbytes(1024)) + with open(guest_cmd, 'w', encoding='utf-8') as s: + s.write(GUEST_SCRIPT) + with open(nested_guest_cmd, 'w', encoding='utf-8') as s: + s.write(NESTED_GUEST_SCRIPT) self.vm.add_args('-cpu', 'max') self.vm.add_args('-m', '2G') @@ -139,5 +144,6 @@ class Aarch64DevicePassthrough(QemuSystemTest): wait_for_console_pattern(self, 'device_passthrough_test_ok', failure_message='Kernel panic') + if __name__ == '__main__': QemuSystemTest.main() diff --git a/tests/functional/m68k/test_nextcube.py b/tests/functional/m68k/test_nextcube.py index e5e1c69dcb..71c160d026 100755 --- a/tests/functional/m68k/test_nextcube.py +++ b/tests/functional/m68k/test_nextcube.py @@ -40,7 +40,7 @@ class NextCubeMachine(QemuSystemTest): time.sleep(0.1) res = self.vm.cmd('human-monitor-command', - command_line='screendump %s' % screenshot_path) + command_line=f"screendump {screenshot_path}") if 'unknown command' in res: self.skipTest('screendump not available') @@ -52,7 +52,7 @@ class NextCubeMachine(QemuSystemTest): from PIL import Image with Image.open(screenshot_path) as image: - width, height = image.size + width, height = image.size self.assertEqual(width, 1120) self.assertEqual(height, 832) diff --git a/tests/functional/mips64el/test_malta.py b/tests/functional/mips64el/test_malta.py index e37463dc29..bc750cb7ad 100755 --- a/tests/functional/mips64el/test_malta.py +++ b/tests/functional/mips64el/test_malta.py @@ -10,7 +10,6 @@ # SPDX-License-Identifier: GPL-2.0-or-later import os -import logging from qemu_test import LinuxKernelTest, Asset from qemu_test import exec_command_and_wait_for_pattern @@ -50,7 +49,7 @@ class MaltaMachineConsole(LinuxKernelTest): self.vm.add_args('-kernel', kernel_path, '-append', kernel_command_line) self.vm.launch() - console_pattern = 'Kernel command line: %s' % kernel_command_line + console_pattern = f'Kernel command line: {kernel_command_line}' self.wait_for_console_pattern(console_pattern) ASSET_KERNEL_3_19_3 = Asset( @@ -66,7 +65,7 @@ class MaltaMachineConsole(LinuxKernelTest): '75ba10cd35fb44e32948eeb26974f061b703c81c4ba2fab1ebcacf1d1bec3b61') @skipUntrustedTest() - def test_mips64el_malta_5KEc_cpio(self): + def test_mips64el_malta_5kec_cpio(self): kernel_path = self.ASSET_KERNEL_3_19_3.fetch() initrd_path = self.uncompress(self.ASSET_CPIO_R1) @@ -134,8 +133,8 @@ class MaltaMachineFramebuffer(LinuxKernelTest): Boot Linux kernel and check Tux logo is displayed on the framebuffer. """ - import numpy as np - import cv2 + import numpy as np # pylint: disable=import-outside-toplevel + import cv2 # pylint: disable=import-outside-toplevel screendump_path = self.scratch_file('screendump.pbm') @@ -149,7 +148,7 @@ class MaltaMachineFramebuffer(LinuxKernelTest): 'clocksource=GIC console=tty0 console=ttyS0') self.vm.add_args('-kernel', kernel_path, '-cpu', 'I6400', - '-smp', '%u' % cpu_cores_count, + '-smp', str(cpu_cores_count), '-vga', 'std', '-append', kernel_command_line) self.vm.launch() @@ -157,7 +156,7 @@ class MaltaMachineFramebuffer(LinuxKernelTest): self.wait_for_console_pattern(framebuffer_ready) self.vm.cmd('human-monitor-command', command_line='stop') res = self.vm.cmd('human-monitor-command', - command_line='screendump %s' % screendump_path) + command_line=f'screendump {screendump_path}') if 'unknown command' in res: self.skipTest('screendump not available') @@ -191,6 +190,8 @@ class MaltaMachineFramebuffer(LinuxKernelTest): self.do_test_i6400_framebuffer_logo(8) +# Add the tests from the 32-bit mipsel file here, too. +# pylint: disable=unused-import,wrong-import-position from mipsel.test_malta import MaltaMachineYAMON if __name__ == '__main__': diff --git a/tests/functional/mips64el/test_replay.py b/tests/functional/mips64el/test_replay.py index e9318448fa..9a432be5ce 100755 --- a/tests/functional/mips64el/test_replay.py +++ b/tests/functional/mips64el/test_replay.py @@ -23,7 +23,7 @@ class Mips64elReplay(ReplayKernelBase): kernel_path = self.archive_extract(self.ASSET_KERNEL_2_63_2, member='boot/vmlinux-2.6.32-5-5kc-malta') kernel_command_line = self.KERNEL_COMMON_COMMAND_LINE + 'console=ttyS0' - console_pattern = 'Kernel command line: %s' % kernel_command_line + console_pattern = f'Kernel command line: {kernel_command_line}' self.run_rr(kernel_path, kernel_command_line, console_pattern, shift=5) @@ -41,7 +41,7 @@ class Mips64elReplay(ReplayKernelBase): @skipUntrustedTest() @skipFlakyTest("https://gitlab.com/qemu-project/qemu/-/issues/2013") - def test_replay_mips64el_malta_5KEc_cpio(self): + def test_replay_mips64el_malta_5kec_cpio(self): self.set_machine('malta') self.cpu = '5KEc' kernel_path = self.ASSET_KERNEL_3_19_3.fetch() diff --git a/tests/functional/qemu_test/testcase.py b/tests/functional/qemu_test/testcase.py index 1d773dd697..58f2740100 100644 --- a/tests/functional/qemu_test/testcase.py +++ b/tests/functional/qemu_test/testcase.py @@ -14,7 +14,6 @@ import logging import os from pathlib import Path -import pycotap import shutil from subprocess import run import sys @@ -23,6 +22,8 @@ import warnings import unittest import uuid +import pycotap + from qemu.machine import QEMUMachine from qemu.utils import hvf_available, kvm_available, tcg_available @@ -34,50 +35,50 @@ from .uncompress import uncompress class QemuBaseTest(unittest.TestCase): - ''' - @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): + ''' + @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 + ''' self.log.debug(f"Uncompress {compressed} format={format}") - if type(compressed) == Asset: + if isinstance(compressed, Asset): compressed.fetch() - (name, ext) = os.path.splitext(str(compressed)) + (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): + ''' + @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 + ''' self.log.debug(f"Extract {archive} format={format}" + f"sub_dir={sub_dir} member={member}") - if type(archive) == Asset: + if isinstance(archive, Asset): archive.fetch() if sub_dir is None: archive_extract(archive, self.scratch_file(), format, member) @@ -89,108 +90,108 @@ class QemuBaseTest(unittest.TestCase): 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): + ''' + Create a temporary directory suitable for storing UNIX + socket paths. + + Returns: a tempfile.TemporaryDirectory instance + ''' 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): + ''' + @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 + ''' 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): + ''' + @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 + ''' 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): + ''' + @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 + ''' 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): + ''' + @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 + ''' return str(Path(self.outputdir, *args)) - ''' - @params plugin name - - Return the full path to the plugin taking into account any host OS - specific suffixes. - ''' def plugin_file(self, plugin_name): + ''' + @params plugin name + + Return the full path to the plugin taking into account any host OS + specific suffixes. + ''' sfx = dso_suffix() return os.path.join('tests', 'tcg', 'plugins', f'{plugin_name}.{sfx}') def assets_available(self): for name, asset in vars(self.__class__).items(): - if name.startswith("ASSET_") and type(asset) == Asset: + if name.startswith("ASSET_") and isinstance(asset, Asset): if not asset.available(): self.log.debug(f"Asset {asset.url} not available") return False @@ -216,9 +217,9 @@ class QemuBaseTest(unittest.TestCase): self.log.setLevel(logging.DEBUG) self._log_fh = logging.FileHandler(self.log_filename, mode='w') self._log_fh.setLevel(logging.DEBUG) - fileFormatter = logging.Formatter( + file_formatter = logging.Formatter( '%(asctime)s - %(levelname)s: %(name)s.%(funcName)s %(message)s') - self._log_fh.setFormatter(fileFormatter) + self._log_fh.setFormatter(file_formatter) self.log.addHandler(self._log_fh) # Capture QEMUMachine logging @@ -260,7 +261,7 @@ class QemuBaseTest(unittest.TestCase): res = unittest.main(module = None, testRunner = tr, exit = False, argv=[sys.argv[0], path] + sys.argv[1:]) failed = {} - for (test, message) in res.result.errors + res.result.failures: + for (test, _message) in res.result.errors + res.result.failures: if hasattr(test, "log_filename") and not test.id() in failed: print('More information on ' + test.id() + ' could be found here:' '\n %s' % test.log_filename, file=sys.stderr) @@ -279,7 +280,9 @@ class QemuUserTest(QemuBaseTest): def add_ldpath(self, ldpath): self._ldpath.append(os.path.abspath(ldpath)) - def run_cmd(self, bin_path, args=[]): + def run_cmd(self, bin_path, args=None): + if args is None: + args = [] return run([self.qemu_bin] + ["-L %s" % ldpath for ldpath in self._ldpath] + [bin_path] @@ -304,8 +307,8 @@ class QemuSystemTest(QemuBaseTest): self._console_log_fh = logging.FileHandler(self.console_log_name, mode='w') self._console_log_fh.setLevel(logging.DEBUG) - fileFormatter = logging.Formatter('%(asctime)s: %(message)s') - self._console_log_fh.setFormatter(fileFormatter) + file_formatter = logging.Formatter('%(asctime)s: %(message)s') + self._console_log_fh.setFormatter(file_formatter) console_log.addHandler(self._console_log_fh) def set_machine(self, machinename): @@ -343,17 +346,15 @@ class QemuSystemTest(QemuBaseTest): "available" % accelerator) def require_netdev(self, netdevname): - help = run([self.qemu_bin, - '-M', 'none', '-netdev', 'help'], - capture_output=True, check=True, encoding='utf8').stdout; - if help.find('\n' + netdevname + '\n') < 0: + helptxt = run([self.qemu_bin, '-M', 'none', '-netdev', 'help'], + capture_output=True, check=True, encoding='utf8').stdout + if helptxt.find('\n' + netdevname + '\n') < 0: self.skipTest('no support for " + netdevname + " networking') def require_device(self, devicename): - help = run([self.qemu_bin, - '-M', 'none', '-device', 'help'], - capture_output=True, check=True, encoding='utf8').stdout; - if help.find(devicename) < 0: + helptxt = run([self.qemu_bin, '-M', 'none', '-device', 'help'], + capture_output=True, check=True, encoding='utf8').stdout + if helptxt.find(devicename) < 0: self.skipTest('no support for device ' + devicename) def _new_vm(self, name, *args): @@ -415,7 +416,7 @@ class QemuSystemTest(QemuBaseTest): try: vm.shutdown() except Exception as ex: - self.log.error("Failed to teardown VM: %s" % ex) + self.log.error("Failed to teardown VM: %s", ex) logging.getLogger('console').removeHandler(self._console_log_fh) self._console_log_fh.close() super().tearDown()