Add functional tests that exercise dynamic monitor hotplug with real
socket connections:

- Hotplug cycle: chardev-add a unix socket, monitor-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 monitor-remove
  targeting itself, verifying that the response is delivered before
  the connection drops and that the monitor is gone afterwards.

- 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>
---
 tests/functional/generic/meson.build             |   1 +
 tests/functional/generic/test_monitor_hotplug.py | 183 +++++++++++++++++++++++
 2 files changed, 184 insertions(+)

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 100644
index 0000000000..d65d33c2f6
--- /dev/null
+++ b/tests/functional/generic/test_monitor_hotplug.py
@@ -0,0 +1,183 @@
+#!/usr/bin/env python3
+#
+# Functional test for dynamic QMP monitor hotplug
+#
+# Copyright (c) 2026 Christian Brauner
+#
+# 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 os
+import tempfile
+
+from qemu.qmp.legacy import QEMUMonitorProtocol
+
+from qemu_test import QemuSystemTest
+
+
+class MonitorHotplug(QemuSystemTest):
+
+    def setUp(self):
+        super().setUp()
+        # Use /tmp to avoid UNIX socket path length limit (108 bytes).
+        # The scratch_file() path is too deep for socket names.
+        fd, self._sock_path = tempfile.mkstemp(
+            prefix='qemu-mon-', suffix='.sock')
+        os.close(fd)
+        os.unlink(self._sock_path)
+
+    def tearDown(self):
+        try:
+            os.unlink(self._sock_path)
+        except FileNotFoundError:
+            pass
+        super().tearDown()
+
+    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('monitor-add', id='hotplug-mon',
+                    chardev='hotplug-chr')
+        return sock
+
+    def _remove_monitor(self):
+        """Remove the monitor + chardev."""
+        self.vm.cmd('monitor-remove', 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 monitor-remove targeting
+        itself.  Verify the response is delivered before the
+        connection drops, and that the monitor is gone afterwards.
+        """
+        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 removes itself
+        resp = qmp.cmd_obj({'execute': 'monitor-remove',
+                            'arguments': {'id': 'hotplug-mon'}})
+        self.assertIn('return', resp)
+
+        qmp.close()
+
+        # The main monitor should no longer list the removed monitor
+        monitors = self.vm.cmd('query-monitors')
+        for m in monitors:
+            self.assertNotEqual(m.get('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.47.3