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