* Fix some issues in the functional tests that pylint complains about

-----BEGIN PGP SIGNATURE-----
 
 iQJFBAABCgAvFiEEJ7iIR+7gJQEY8+q5LtnXdP5wLbUFAmkTDfQRHHRodXRoQHJl
 ZGhhdC5jb20ACgkQLtnXdP5wLbVj8RAAhOSNyBa81eFJXydkqp0qrQYw6WGT/mAP
 Zn5oTm6NhsgLbUKgbqYQIAivE7VNVWfdhj7aOO9wYM1GfhCk/LOHZWBTNXxFF/uH
 m7ICV5dtSF2zE1AdsWn2rB6vPocc/VMDCHhIzfC7AYlEA7AGuu/O2QALE8H/qOS5
 mQ3+Fuq2EYkOKxKsSnUcj+ZPnUA3NlIF2CTeY0jTQFrwO5RKU3jsScm+uOZZJycn
 DTOzJTymIBGNSlFMNEoj4AhoY43SDdcQcZhwvAPzHZZTVhotJxHf5Fvr7XnDW5VA
 zTA7xZgnY0eAtvzZ4ihyT9BfAHdk62WgBrUeohQ1Ggf/Bo11DVCJtkQ4iY5bY4uI
 yalO7QSMi04PudeIRJmKTAhR6zhDZb/XijtrIcFn6ypTnOEMw8V7MJt9qXB76I/X
 HDZ9859a0//8F70I3mAxDKj8ve/Y6ACuY7pOwKR1Ea0iuM47Dgw9jsuUKRRPUZ+p
 rhJiQ10j8B6mxI0HCqEr8S47zMbW7uJViVYLT7yYKL7vokr96mm08/gEOI07cc88
 CKw3FocW2/suOdFCJVsIrjjq/ySVv0GTAkIeGUaefnY13dmq8ZILmT+GOOf695s9
 PDCoPWzdCY5n0OxToMUosJkQKbFp2F2ls5IGcEHUwxkqPT68/gsqb1VeC8W7x6Gs
 nJGM9ZR7XcM=
 =FhJ1
 -----END PGP SIGNATURE-----

Merge tag 'pull-request-2025-11-11' of https://gitlab.com/thuth/qemu into staging

* Fix some issues in the functional tests that pylint complains about

# -----BEGIN PGP SIGNATURE-----
#
# iQJFBAABCgAvFiEEJ7iIR+7gJQEY8+q5LtnXdP5wLbUFAmkTDfQRHHRodXRoQHJl
# ZGhhdC5jb20ACgkQLtnXdP5wLbVj8RAAhOSNyBa81eFJXydkqp0qrQYw6WGT/mAP
# Zn5oTm6NhsgLbUKgbqYQIAivE7VNVWfdhj7aOO9wYM1GfhCk/LOHZWBTNXxFF/uH
# m7ICV5dtSF2zE1AdsWn2rB6vPocc/VMDCHhIzfC7AYlEA7AGuu/O2QALE8H/qOS5
# mQ3+Fuq2EYkOKxKsSnUcj+ZPnUA3NlIF2CTeY0jTQFrwO5RKU3jsScm+uOZZJycn
# DTOzJTymIBGNSlFMNEoj4AhoY43SDdcQcZhwvAPzHZZTVhotJxHf5Fvr7XnDW5VA
# zTA7xZgnY0eAtvzZ4ihyT9BfAHdk62WgBrUeohQ1Ggf/Bo11DVCJtkQ4iY5bY4uI
# yalO7QSMi04PudeIRJmKTAhR6zhDZb/XijtrIcFn6ypTnOEMw8V7MJt9qXB76I/X
# HDZ9859a0//8F70I3mAxDKj8ve/Y6ACuY7pOwKR1Ea0iuM47Dgw9jsuUKRRPUZ+p
# rhJiQ10j8B6mxI0HCqEr8S47zMbW7uJViVYLT7yYKL7vokr96mm08/gEOI07cc88
# CKw3FocW2/suOdFCJVsIrjjq/ySVv0GTAkIeGUaefnY13dmq8ZILmT+GOOf695s9
# PDCoPWzdCY5n0OxToMUosJkQKbFp2F2ls5IGcEHUwxkqPT68/gsqb1VeC8W7x6Gs
# nJGM9ZR7XcM=
# =FhJ1
# -----END PGP SIGNATURE-----
# gpg: Signature made Tue 11 Nov 2025 11:20:36 AM CET
# gpg:                using RSA key 27B88847EEE0250118F3EAB92ED9D774FE702DB5
# gpg:                issuer "thuth@redhat.com"
# gpg: Good signature from "Thomas Huth <th.huth@gmx.de>" [unknown]
# gpg:                 aka "Thomas Huth <thuth@redhat.com>" [unknown]
# gpg:                 aka "Thomas Huth <th.huth@posteo.de>" [unknown]
# gpg:                 aka "Thomas Huth <huth@tuxfamily.org>" [unknown]
# gpg: WARNING: The key's User ID is not certified with a trusted signature!
# gpg:          There is no indication that the signature belongs to the owner.
# Primary key fingerprint: 27B8 8847 EEE0 2501 18F3  EAB9 2ED9 D774 FE70 2DB5

* tag 'pull-request-2025-11-11' of https://gitlab.com/thuth/qemu:
  tests/functional/m68k/test_nextcube: Fix issues reported by pylint
  tests/functional/mips64el: Silence issues reported by pylint
  tests/functional/aarch64/test_device_passthrough: Fix warnings from pylint
  tests/functional: Fix problems in testcase.py reported by pylint

Signed-off-by: Richard Henderson <richard.henderson@linaro.org>
This commit is contained in:
Richard Henderson 2025-11-11 17:24:13 +01:00
commit 4481234e98
5 changed files with 155 additions and 147 deletions

View file

@ -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()

View file

@ -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)

View file

@ -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__':

View file

@ -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()

View file

@ -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()