| #!/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() |