tests/test_aio.py

   1 import asyncio
   2 import libvirt
   3 import libvirtaio
   4 import sys
   5 import unittest
   6 from unittest import mock
   7
   8 import eventmock
   9
  10
  11 class TestLibvirtAio(unittest.TestCase):
  12     async def _run(self, register):
  13         def lifecycleCallback(conn, dom, event, detail, domainChangedEvent):
  14             if (event == libvirt.VIR_DOMAIN_EVENT_STOPPED or
  15                     event == libvirt.VIR_DOMAIN_EVENT_STARTED):
  16                 domainChangedEvent.set()
  17
  18         if register:
  19             libvirtEvents = libvirtaio.virEventRegisterAsyncIOImpl()
  20         else:
  21             libvirtEvents = libvirtaio.getCurrentImpl()
  22
  23         conn = libvirt.open("test:///default")
  24         dom = conn.lookupByName("test")
  25
  26         eventRegistered = False
  27         domainStopped = False
  28         try:
  29             # Ensure the VM is running.
  30             self.assertEqual([libvirt.VIR_DOMAIN_RUNNING, libvirt.VIR_DOMAIN_RUNNING_UNKNOWN], dom.state())
  31             self.assertTrue(libvirtEvents.is_idle())
  32
  33             # Register VM start/stopped event handler.
  34             domainChangedEvent = asyncio.Event()
  35             conn.domainEventRegisterAny(dom, libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE, lifecycleCallback, domainChangedEvent)
  36             eventRegistered = True
  37
  38             self.assertFalse(libvirtEvents.is_idle())
  39
  40             # Stop the VM.
  41             dom.destroy()
  42             domainStopped = True
  43
  44             # Ensure domain stopped event is received.
  45             await asyncio.wait_for(domainChangedEvent.wait(), 2)
  46             self.assertEqual([libvirt.VIR_DOMAIN_SHUTOFF, libvirt.VIR_DOMAIN_SHUTOFF_DESTROYED], dom.state())
  47
  48             # Start the VM.
  49             domainChangedEvent.clear()
  50             domainStopped = False
  51             dom.create()
  52
  53             # Ensure domain started event is received.
  54             await asyncio.wait_for(domainChangedEvent.wait(), 2)
  55             self.assertEqual([libvirt.VIR_DOMAIN_RUNNING, libvirt.VIR_DOMAIN_RUNNING_BOOTED], dom.state())
  56             self.assertFalse(libvirtEvents.is_idle())
  57
  58             # Deregister the VM start/stopped event handler.
  59             eventRegistered = False
  60             conn.domainEventDeregisterAny(libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE)
  61
  62             # Wait for event queue to clear.
  63             await libvirtEvents.drain()
  64
  65             # Make sure event queue is cleared.
  66             self.assertTrue(libvirtEvents.is_idle())
  67
  68         finally:
  69             if eventRegistered:
  70                 conn.domainEventDeregisterAny(libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE)
  71
  72             if domainStopped:
  73                 dom.create()
  74
  75     @mock.patch('libvirt.virEventRegisterImpl',
  76                 side_effect=eventmock.virEventRegisterImplMock)
  77     def testEventsWithManualLoopSetup(self, mock_event_register):
  78         # Register libvirt events after starting the asyncio loop.
  79         #
  80         # Manually create and set the event loop against this
  81         # thread.
  82         loop = asyncio.new_event_loop()
  83         asyncio.set_event_loop(loop)
  84
  85         loop.run_until_complete(self._run(register=True))
  86
  87         loop.close()
  88         asyncio.set_event_loop(None)
  89         mock_event_register.assert_called_once()
  90
  91     @mock.patch('libvirt.virEventRegisterImpl',
  92                 side_effect=eventmock.virEventRegisterImplMock)
  93     @unittest.skipIf(sys.version_info < (3,7), "test requires Python 3.7+")
  94     def testEventsWithAsyncioRun(self, mock_event_register):
  95         # Register libvirt events after starting the asyncio loop.
  96         #
  97         # Use asyncio helper to create and set the event loop
  98         # against this thread.
  99         asyncio.run(self._run(register=True))
 100         mock_event_register.assert_called_once()
 101
 102     @mock.patch('libvirt.virEventRegisterImpl',
 103                 side_effect=eventmock.virEventRegisterImplMock)
 104     def testEventsPreInitExplicit(self, mock_event_register):
 105         # Register libvirt events before starting the asyncio loop.
 106         #
 107         # Tell virEventRegisterAsyncIOImpl() explicitly what loop
 108         # to use before we set a loop for this thread.
 109         loop = asyncio.new_event_loop()
 110         libvirtaio.virEventRegisterAsyncIOImpl(loop)
 111         asyncio.set_event_loop(loop)
 112
 113         loop.run_until_complete(self._run(register=False))
 114
 115         loop.close()
 116         asyncio.set_event_loop(None)
 117         mock_event_register.assert_called_once()
 118
 119     @mock.patch('libvirt.virEventRegisterImpl',
 120                 side_effect=eventmock.virEventRegisterImplMock)
 121     @unittest.skipIf(sys.version_info >= (3,10), "test incompatible with 3.10+")
 122     def testEventsPreInitImplicit(self, mock_event_register):
 123         # Register libvirt events before starting the asyncio loop.
 124         #
 125         # Allow virEventRegisterAsyncIOImpl() to implicitly find the
 126         # loop we set for this thread.
 127         loop = asyncio.new_event_loop()
 128         asyncio.set_event_loop(loop)
 129         libvirtaio.virEventRegisterAsyncIOImpl()
 130
 131         loop.run_until_complete(self._run(register=False))
 132
 133         loop.close()
 134         asyncio.set_event_loop(None)
 135         mock_event_register.assert_called_once()
 136
 137     @mock.patch('libvirt.virEventRegisterImpl',
 138                 side_effect=eventmock.virEventRegisterImplMock)
 139     @unittest.skipIf(sys.version_info >= (3,10), "test incompatible with 3.10+")
 140     def testEventsImplicitLoopInit(self, mock_event_register):
 141         # Register libvirt events before starting the asyncio loop.
 142         #
 143         # Let virEventRegisterAsyncIOImpl() auto-create a default
 144         # event loop, which we then register against this thread.
 145         #
 146         # Historically this often worked if called from the main thead,
 147         # but since Python 3.10 this triggers a deprecation warning,
 148         # which will turn into a RuntimeError in a later release.
 149         libvirtaio.virEventRegisterAsyncIOImpl()
 150         loop = asyncio.get_event_loop()
 151
 152         loop.run_until_complete(self._run(register=False))
 153
 154         loop.close()
 155         asyncio.set_event_loop(None)
 156         mock_event_register.assert_called_once()