From: Christian Brauner <brauner@kernel.org>
Add functional tests that exercise dynamic monitor hotplug with real
socket connections:
- Hotplug cycle: chardev-add a unix socket, object-add, connect to the
socket, receive the QMP greeting, negotiate capabilities, send
query-version, disconnect, remove the monitor and chardev, then repeat
the entire cycle a second time to verify cleanup and reuse.
- Self-removal: a dynamically-added monitor sends object-del
targeting itself, verifying that the request is rejected
- Large response: send query-qmp-schema on a dynamic monitor to
exercise the output buffer flush path with a large response payload.
- Events after negotiation: trigger STOP/RESUME events via the main
monitor and verify they are delivered on the dynamic monitor.
This complements the qtest unit tests by verifying that a real QMP
client can connect to a dynamically-added monitor and exchange messages.
Signed-off-by: Christian Brauner (Amutable) <brauner@kernel.org>
[DB: modified to use object-add/object-del; adjust self-removal test
to validate rejection of request]
Reviewed-by: Marc-André Lureau <marcandre.lureau@redhat.com>
Tested-by: Peter Krempa <pkrempa@redhat.com>
Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>
---
MAINTAINERS | 1 +
tests/functional/generic/meson.build | 1 +
.../generic/test_monitor_hotplug.py | 168 ++++++++++++++++++
3 files changed, 170 insertions(+)
create mode 100755 tests/functional/generic/test_monitor_hotplug.py
diff --git a/MAINTAINERS b/MAINTAINERS
index 97dcc78ded..eb0f9fc9ba 100644
--- a/MAINTAINERS
+++ b/MAINTAINERS
@@ -3669,6 +3669,7 @@ F: docs/interop/*qmp-*
F: scripts/qmp/
F: tests/qtest/qmp-test.c
F: tests/qtest/qmp-cmd-test.c
+F: tests/functional/generic/test_monitor_hotplug.py
T: git https://repo.or.cz/qemu/armbru.git qapi-next
qtest
diff --git a/tests/functional/generic/meson.build b/tests/functional/generic/meson.build
index 09763c5d22..c94105c62e 100644
--- a/tests/functional/generic/meson.build
+++ b/tests/functional/generic/meson.build
@@ -4,6 +4,7 @@ tests_generic_system = [
'empty_cpu_model',
'info_usernet',
'linters',
+ 'monitor_hotplug',
'version',
'vnc',
]
diff --git a/tests/functional/generic/test_monitor_hotplug.py b/tests/functional/generic/test_monitor_hotplug.py
new file mode 100755
index 0000000000..5d8a159eb0
--- /dev/null
+++ b/tests/functional/generic/test_monitor_hotplug.py
@@ -0,0 +1,168 @@
+#!/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 os
+
+from qemu_test import QemuSystemTest
+
+from qemu.qmp.legacy import QEMUMonitorProtocol
+
+
+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):
+ """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,
+ }
+ })
+ 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_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()
+
+
+if __name__ == '__main__':
+ QemuSystemTest.main()
--
2.55.0