Import QEMU upstream snapshot d2e570c

Upstream: https://gitlab.com/qemu-project/qemu.git

Upstream-Commit: d2e570cc0f97b936902a5b1b86b73c0f5998b475

tests/functional/generic/meson.build

@@ -0,0 +1,16 @@
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+tests_generic_system = [
+  'empty_cpu_model',
+  'info_usernet',
+  'linters',
+  'monitor_hotplug',
+  'version',
+  'vnc',
+]
+
+tests_generic_linuxuser = [
+]
+
+tests_generic_bsduser = [
+]

tests/functional/generic/test_empty_cpu_model.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+#
+# Check for crash when using empty -cpu option
+#
+# Copyright (c) 2019 Red Hat, Inc.
+#
+# Author:
+#  Eduardo Habkost <[email protected]>
+#
+# 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 qemu_test import QemuSystemTest
+
+class EmptyCPUModel(QemuSystemTest):
+    def test(self):
+        self.vm.add_args('-S', '-display', 'none', '-machine', 'none', '-cpu', '')
+        self.vm.set_qmp_monitor(enabled=False)
+        self.vm.launch()
+        self.vm.wait()
+        self.assertEqual(self.vm.exitcode(), 1, "QEMU exit code should be 1")
+        self.assertRegex(self.vm.get_log(), r'-cpu option cannot be empty')
+
+if __name__ == '__main__':
+    QemuSystemTest.main()

tests/functional/generic/test_info_usernet.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+#
+# Test for the hmp command "info usernet"
+#
+# Copyright (c) 2021 Red Hat, Inc.
+#
+# Author:
+#  Cleber Rosa <[email protected]>
+#
+# 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 qemu_test import QemuSystemTest
+from qemu_test.utils import get_usernet_hostfwd_port
+
+
+class InfoUsernet(QemuSystemTest):
+
+    def test_hostfwd(self):
+        self.require_netdev('user')
+        self.set_machine('none')
+        self.vm.add_args('-netdev', 'user,id=vnet,hostfwd=:127.0.0.1:0-:22')
+        self.vm.launch()
+
+        port = get_usernet_hostfwd_port(self.vm)
+        self.assertIsNotNone(port,
+                             ('"info usernet" output content does not seem to '
+                              'contain the redirected port'))
+        self.assertGreater(port, 0,
+                           ('Found a redirected port that is not greater than'
+                            ' zero'))
+
+if __name__ == '__main__':
+    QemuSystemTest.main()

tests/functional/generic/test_linters.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+'''Python linter tests'''
+
+import os
+
+from pathlib import Path
+from qemu_test import QemuBaseTest, skipIfMissingImports
+
+
+class LinterTest(QemuBaseTest):
+    '''
+    Run python linters on the test *.py files
+    '''
+
+    @skipIfMissingImports("pylint")
+    def test_pylint(self):
+        '''Check source files with pylint'''
+        from pylint.lint import Run as pylint_run
+        from pylint.reporters.collecting_reporter import CollectingReporter
+        srcdir = os.path.join(Path(__file__).parent.parent, self.arch)
+        rcfile = os.path.join(Path(__file__).parent.parent, "pylintrc")
+        self.log.info('Checking files in %s with pylint', srcdir)
+        reporter = CollectingReporter()
+        pylint_run(["--rcfile", rcfile, srcdir], reporter=reporter, exit=False)
+        if reporter.messages:
+            fmt = '"{path}:{line}: {msg_id}: {msg} ({symbol})"'
+            for msg in reporter.messages:
+                if msg.category == "error":
+                    self.log.error(msg.format(fmt))
+                elif msg.category == "warning":
+                    self.log.warning(msg.format(fmt))
+                else:
+                    self.log.info(msg.format(fmt))
+            self.fail("Pylint failed, see base.log for details.")
+
+
+if __name__ == '__main__':
+    QemuBaseTest.main()

tests/functional/generic/test_monitor_hotplug.py

@@ -0,0 +1,305 @@
+#!/usr/bin/env python3
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+# Functional test for dynamic QMP monitor hotplug
+#
+# Copyright (c) 2026 Christian Brauner
+
+import asyncio
+import os
+import random
+import threading
+import time
+
+from qemu_test import QemuSystemTest
+
+from qemu.qmp.legacy import QEMUMonitorProtocol
+from qemu.qmp import QMPClient
+
+
+class MonitorHotplug(QemuSystemTest):
+
+    def setUp(self):
+        super().setUp()
+        sock_dir = self.socket_dir()
+        self._sock_path = os.path.join(sock_dir.name, 'hotplug.sock')
+
+    def _add_monitor(self, autodelete=False):
+        """Create a chardev + monitor and return the socket path."""
+        sock = self._sock_path
+        self.vm.cmd('chardev-add', id='hotplug-chr', backend={
+            'type': 'socket',
+            'data': {
+                'addr': {
+                    'type': 'unix',
+                    'data': {'path': sock}
+                },
+                'server': True,
+                'wait': False,
+            }
+        })
+        if autodelete:
+            self.vm.cmd('object-add', id='hotplug-mon',
+                        qom_type='monitor-qmp',
+                        chardev='hotplug-chr',
+                        close_action='delete')
+        else:
+            self.vm.cmd('object-add', id='hotplug-mon',
+                        qom_type='monitor-qmp',
+                        chardev='hotplug-chr')
+        return sock
+
+    def _remove_monitor(self):
+        """Remove the monitor + chardev."""
+        self.vm.cmd('object-del', id='hotplug-mon')
+        self.vm.cmd('chardev-remove', id='hotplug-chr')
+
+    def _connect_and_handshake(self, sock_path):
+        """
+        Connect to the dynamic monitor socket, perform the QMP
+        greeting and capability negotiation, send a command, then
+        disconnect.
+        """
+        qmp = QEMUMonitorProtocol(sock_path)
+
+        # connect(negotiate=True) receives the greeting, validates it,
+        # and sends qmp_capabilities automatically.
+        greeting = qmp.connect(negotiate=True)
+        self.assertIn('QMP', greeting)
+        self.assertIn('version', greeting['QMP'])
+        self.assertIn('capabilities', greeting['QMP'])
+
+        # Send a real command to prove the session is fully functional
+        resp = qmp.cmd_obj({'execute': 'query-version'})
+        self.assertIn('return', resp)
+        self.assertIn('qemu', resp['return'])
+
+        qmp.close()
+
+    def test_hotplug_cycle(self):
+        """
+        Hotplug a monitor, do the full QMP handshake, unplug it,
+        then repeat the whole cycle a second time.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        # First cycle
+        sock = self._add_monitor()
+        self._connect_and_handshake(sock)
+        self._remove_monitor()
+
+        # Second cycle -- same ids, same path, must work
+        sock = self._add_monitor()
+        self._connect_and_handshake(sock)
+        self._remove_monitor()
+
+    def test_self_removal(self):
+        """
+        A dynamically-added monitor sends object-del targeting
+        itself.  Verify the request is rejected, but the monitor
+        can still be deleted from outside its own context.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        sock = self._add_monitor()
+
+        qmp = QEMUMonitorProtocol(sock)
+        greeting = qmp.connect(negotiate=True)
+        self.assertIn('QMP', greeting)
+
+        # Self-removal: the dynamic monitor raises error
+        resp = qmp.cmd_obj({'execute': 'object-del',
+                            'arguments': {'id': 'hotplug-mon'}})
+        self.assertIn('error', resp)
+
+        qmp.close()
+
+        resp = self.vm.cmd('object-del', id='hotplug-mon')
+
+        # Clean up the chardev
+        self.vm.cmd('chardev-remove', id='hotplug-chr')
+
+    def test_auto_delete(self):
+        """
+        A dynamically-added monitor configured with 'close-action=delete'
+        should see itself deleted when the client is closed.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        sock = self._add_monitor(autodelete=True)
+
+        cdevs = [c["label"] for c in self.vm.cmd('query-chardev')]
+        objs = [o["name"] for o in self.vm.cmd('qom-list', path='/objects')]
+        assert ('hotplug-chr' in cdevs)
+        assert ('hotplug-mon' in objs)
+
+        qmp = QEMUMonitorProtocol(sock)
+        greeting = qmp.connect(negotiate=True)
+        self.assertIn('QMP', greeting)
+
+        cdevs = [c["label"] for c in self.vm.cmd('query-chardev')]
+        objs = [o["name"] for o in self.vm.cmd('qom-list', path='/objects')]
+        assert ('hotplug-chr' in cdevs)
+        assert ('hotplug-mon' in objs)
+
+        qmp.close()
+
+        # Wait upto 10 seconds max for chardev to auto-delete, which
+        # is hopefully enough for reliability under high load
+        for i in range(int(10 / 0.2)):
+            cdevs = [c["label"] for c in self.vm.cmd('query-chardev')]
+            if 'hotplug-chr' not in cdevs:
+                break
+            # Wait a little more then try again
+            time.sleep(0.2)
+
+        cdevs = [c["label"] for c in self.vm.cmd('query-chardev')]
+        objs = [o["name"] for o in self.vm.cmd('qom-list', path='/objects')]
+        assert ('hotplug-chr' not in cdevs)
+        assert ('hotplug-mon' not in objs)
+
+    def test_reconnect(self):
+        """
+        A dynamically-added monitor configured without 'close-action'
+        should allow reconnects after the client is closed.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        sock = self._add_monitor()
+
+        qmp = QEMUMonitorProtocol(sock)
+        qmp.connect(negotiate=True)
+
+        resp = qmp.cmd_obj({'execute': 'query-chardev'})
+        self.assertIn('return', resp)
+
+        qmp.close()
+
+        qmp = QEMUMonitorProtocol(sock)
+        qmp.connect(negotiate=True)
+
+        resp = qmp.cmd_obj({'execute': 'query-chardev'})
+        self.assertIn('return', resp)
+
+        qmp.close()
+        self._remove_monitor()
+
+    def test_large_response(self):
+        """
+        Send a command with a large response (query-qmp-schema) on a
+        dynamically-added monitor to exercise the output buffer flush
+        path.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        sock = self._add_monitor()
+
+        qmp = QEMUMonitorProtocol(sock)
+        qmp.connect(negotiate=True)
+
+        resp = qmp.cmd_obj({'execute': 'query-qmp-schema'})
+        self.assertIn('return', resp)
+        self.assertIsInstance(resp['return'], list)
+        self.assertGreater(len(resp['return']), 0)
+
+        qmp.close()
+        self._remove_monitor()
+
+    def test_events_after_negotiation(self):
+        """
+        Verify that QMP events are delivered on a dynamically-added
+        monitor after capability negotiation completes.
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        sock = self._add_monitor()
+
+        qmp = QEMUMonitorProtocol(sock)
+        qmp.connect(negotiate=True)
+
+        # Trigger a STOP event via the main monitor, then read it
+        # from the dynamic monitor.
+        self.vm.cmd('stop')
+        resp = qmp.pull_event(wait=True)
+        self.assertEqual(resp['event'], 'STOP')
+
+        self.vm.cmd('cont')
+        resp = qmp.pull_event(wait=True)
+        self.assertEqual(resp['event'], 'RESUME')
+
+        qmp.close()
+        self._remove_monitor()
+
+    def stress_mon(self, sock):
+        async def main():
+            qmp = QMPClient('testvm')
+            await qmp.connect(sock)
+            # Run query-version in a tight loop so that the
+            # monitor thread/dispatcher is very busy at the
+            # time we try to delete the monitor
+            while True:
+                try:
+                    # A command which returns a lot of data to make
+                    # it more likely we're in the I/O reply path
+                    # when deleting the monitor
+                    res = await qmp.execute('query-qmp-schema')
+                    # Some commands which generate async events
+                    # as those can trigger different code paths
+                    res = await qmp.execute('stop')
+                    res = await qmp.execute('cont')
+                except:
+                    # we'll get here if the monitor is terminated
+                    # by QEMU in which case we must disconnect
+                    # out side, but....
+                    try:
+                        await qmp.disconnect()
+                    except (ConnectionResetError, EOFError, BrokenPipeError):
+                        # ... disconnect() will probably see
+                        # errors too, but we must try to call it
+                        # regardless to cleanup asyncio state
+                        # and prevent python warnings at GC time
+                        pass
+                    return
+        asyncio.run(main())
+
+    def test_hotplug_stress(self):
+        """
+        Repeatedly hotplug and unplug a monitor, while another thread
+        concurrently issues commands on that monitor. This stresses
+        the synchronization with the monitor thread during cleanup
+        """
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+
+        # Each loop sleeps at most 0.5 seconds, so this should
+        # give an upper bound of approx 5 seconds execution
+        # time which is reasonable to run by default
+        repeat = 10
+        for i in range(repeat):
+            # First cycle
+            sock = self._add_monitor()
+            print ("# stress cycle %02d/%02d" % (i, repeat))
+            stress = threading.Thread(target=self.stress_mon, args=[sock])
+            stress.start()
+            # Sleep upto 1/2 second to vary the races
+            time.sleep(random.random() / 2)
+            self._remove_monitor()
+            stress.join()
+
+
+if __name__ == '__main__':
+    QemuSystemTest.main()

tests/functional/generic/test_version.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+#
+# Version check example test
+#
+# Copyright (c) 2018 Red Hat, Inc.
+#
+# Author:
+#  Cleber Rosa <[email protected]>
+#
+# 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 qemu_test import QemuSystemTest
+
+
+class Version(QemuSystemTest):
+
+    def test_qmp_human_info_version(self):
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults')
+        self.vm.launch()
+        res = self.vm.cmd('human-monitor-command',
+                          command_line='info version')
+        self.assertRegex(res, r'^(\d+\.\d+\.\d)')
+
+if __name__ == '__main__':
+    QemuSystemTest.main()

tests/functional/generic/test_vmstate.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+#
+'''This test runs the vmstate-static-checker script with the current QEMU'''
+
+import subprocess
+
+from qemu_test import QemuSystemTest, skipFlakyTest
+
+
+@skipFlakyTest("vmstate-static-checker can produce false positives")
+class VmStateTest(QemuSystemTest):
+    '''
+    This test helps to check whether there are problems between old
+    reference data and the current QEMU
+    '''
+
+    def test_vmstate_7_2(self):
+        '''Check reference data from QEMU v7.2'''
+
+        target_machine = {
+            'aarch64': 'virt-7.2',
+            'm68k': 'virt-7.2',
+            'ppc64': 'pseries-7.2',
+            's390x': 's390-ccw-virtio-7.2',
+            'x86_64': 'pc-q35-7.2',
+        }
+        self.set_machine(target_machine[self.arch])
+
+        # Run QEMU to get the current vmstate json file:
+        dst_json = self.scratch_file('dest.json')
+        self.log.info('Dumping vmstate from %s', self.qemu_bin)
+        cp = subprocess.run([self.qemu_bin, '-nodefaults',
+                             '-M', target_machine[self.arch],
+                             '-dump-vmstate', dst_json],
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.STDOUT,
+                            text=True, check=True)
+        if cp.stdout:
+            self.log.info('QEMU output: %s', cp.stdout)
+
+        # Check whether the old vmstate json file is still compatible:
+        src_json = self.data_file('..', 'data', 'vmstate-static-checker',
+                                  self.arch,
+                                  target_machine[self.arch] + '.json')
+        self.log.info('Comparing vmstate with %s', src_json)
+        checkerscript = self.data_file('..', '..', 'scripts',
+                                       'vmstate-static-checker.py')
+        cp = subprocess.run([checkerscript, '-s', src_json, '-d', dst_json],
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.STDOUT,
+                            text=True, check=False)
+        if cp.returncode != 0:
+            self.fail('Running vmstate-static-checker failed:\n' + cp.stdout +
+                      '\nThis either means that there is a migration bug '
+                      'that needs to be fixed, or\nvmstate-static-checker.py '
+                      'needs to be improved (e.g. extend the changed_names\n'
+                      'in case a field has been renamed), or drop the '
+                      'problematic field from\n' + src_json +
+                      '\nin case the script cannot be fixed easily.')
+        if cp.stdout:
+            self.log.warning('vmstate-static-checker output: %s', cp.stdout)
+
+
+if __name__ == '__main__':
+    QemuSystemTest.main()

tests/functional/generic/test_vnc.py

@@ -0,0 +1,116 @@
+#!/usr/bin/env python3
+#
+# Simple functional tests for VNC functionality
+#
+# Copyright (c) 2018 Red Hat, Inc.
+#
+# Author:
+#  Cleber Rosa <[email protected]>
+#
+# 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 socket
+
+from qemu.machine.machine import VMLaunchFailure
+from qemu_test import QemuSystemTest
+from qemu_test.ports import Ports
+
+
+VNC_ADDR = '127.0.0.1'
+
+def check_connect(port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        try:
+            sock.connect((VNC_ADDR, port))
+        except ConnectionRefusedError:
+            return False
+
+    return True
+
+class Vnc(QemuSystemTest):
+
+    def test_no_vnc_change_password(self):
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults', '-S')
+        self.vm.launch()
+
+        query_vnc_response = self.vm.qmp('query-vnc')
+        if 'error' in query_vnc_response:
+            self.assertEqual(query_vnc_response['error']['class'],
+                             'CommandNotFound')
+            self.skipTest('VNC support not available')
+        self.assertFalse(query_vnc_response['return']['enabled'])
+
+        set_password_response = self.vm.qmp('change-vnc-password',
+                                            password='new_password')
+        self.assertIn('error', set_password_response)
+        self.assertEqual(set_password_response['error']['class'],
+                         'GenericError')
+        self.assertEqual(set_password_response['error']['desc'],
+                         'No VNC display is present');
+
+    def launch_guarded(self):
+        try:
+            self.vm.launch()
+        except VMLaunchFailure as excp:
+            if "-vnc: invalid option" in excp.output:
+                self.skipTest("VNC support not available")
+            elif "Cipher backend does not support DES algorithm" in excp.output:
+                self.skipTest("No cryptographic backend available")
+            else:
+                self.log.info("unhandled launch failure: %s", excp.output)
+                raise excp
+
+    def test_change_password_requires_a_password(self):
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999')
+        self.launch_guarded()
+        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
+        set_password_response = self.vm.qmp('change-vnc-password',
+                                            password='new_password')
+        self.assertIn('error', set_password_response)
+        self.assertEqual(set_password_response['error']['class'],
+                         'GenericError')
+        self.assertEqual(set_password_response['error']['desc'],
+                         'VNC password authentication is disabled')
+
+    def test_change_password(self):
+        self.set_machine('none')
+        self.vm.add_args('-nodefaults', '-S', '-vnc', ':1,to=999,password=on')
+        self.launch_guarded()
+        self.assertTrue(self.vm.qmp('query-vnc')['return']['enabled'])
+        self.vm.cmd('change-vnc-password',
+                    password='new_password')
+
+    def do_test_change_listen(self, a, b, c):
+        self.assertFalse(check_connect(a))
+        self.assertFalse(check_connect(b))
+        self.assertFalse(check_connect(c))
+
+        self.vm.add_args('-nodefaults', '-S', '-vnc', f'{VNC_ADDR}:{a - 5900}')
+        self.launch_guarded()
+        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(a))
+        self.assertTrue(check_connect(a))
+        self.assertFalse(check_connect(b))
+        self.assertFalse(check_connect(c))
+
+        self.vm.cmd('display-update', type='vnc',
+                    addresses=[{'type': 'inet', 'host': VNC_ADDR,
+                                'port': str(b)},
+                               {'type': 'inet', 'host': VNC_ADDR,
+                                'port': str(c)}])
+        self.assertEqual(self.vm.qmp('query-vnc')['return']['service'], str(b))
+        self.assertFalse(check_connect(a))
+        self.assertTrue(check_connect(b))
+        self.assertTrue(check_connect(c))
+
+    def test_change_listen(self):
+        self.set_machine('none')
+        with Ports() as ports:
+            a, b, c = ports.find_free_ports(3)
+            self.do_test_change_listen(a, b, c)
+
+
+if __name__ == '__main__':
+    QemuSystemTest.main()