tests: add libvirtaio test coverage

Signed-off-by: Chris Gunn <chrisgun@microsoft.com>

3 files changed, 129 insertions, 1 deletions

diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..b23017c
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,3 @@
+[pytest]
+markers =
+    separate_process: mark test as requiring its own process
diff --git a/setup.py b/setup.py
index 0bb60e1..3e99ac4 100755
--- a/setup.py
+++ b/setup.py
@@ -310,8 +310,19 @@ class my_test(Command):
             os.environ["PYTHONPATH"] = self.build_platlib
 
         pytest = self.find_pytest_path()
-        subprocess.check_call([pytest])
 
+        # Run the normal tests.
+        subprocess.check_call([pytest, "-m", "not separate_process"])
+
+        # Run the tests that require their own process.
+        testlist = subprocess.run(
+            [pytest, "--collect-only", "--quiet", "-m", "separate_process"],
+            check=True, stdout=subprocess.PIPE)
+        testlist = testlist.stdout.decode("utf-8").splitlines()
+        testlist = filter(
+            lambda test: test and "tests collected" not in test, testlist)
+        for test in testlist:
+            subprocess.check_call([pytest, test])
 
 class my_clean(Command):
     def run(self):
diff --git a/tests/test_aio.py b/tests/test_aio.py
new file mode 100644
index 0000000..fb8c6b6
--- /dev/null
+++ b/tests/test_aio.py
@@ -0,0 +1,114 @@
+import asyncio
+import libvirt
+import libvirtaio
+import sys
+import unittest
+import pytest
+
+
+class TestLibvirtAio(unittest.TestCase):
+    async def _run(self, register):
+        def lifecycleCallback(conn, dom, event, detail, domainChangedEvent):
+            if (event == libvirt.VIR_DOMAIN_EVENT_STOPPED or
+                    event == libvirt.VIR_DOMAIN_EVENT_STARTED):
+                domainChangedEvent.set()
+
+        if register:
+            libvirtEvents = libvirtaio.virEventRegisterAsyncIOImpl()
+        else:
+            libvirtEvents = libvirtaio.getCurrentImpl()
+
+        conn = libvirt.open("test:///default")
+        dom = conn.lookupByName("test")
+
+        eventRegistered = False
+        domainStopped = False
+        try:
+            # Ensure the VM is running.
+            self.assertEqual([libvirt.VIR_DOMAIN_RUNNING, libvirt.VIR_DOMAIN_RUNNING_UNKNOWN], dom.state())
+            self.assertTrue(libvirtEvents.is_idle())
+
+            # Register VM start/stopped event handler.
+            domainChangedEvent = asyncio.Event()
+            conn.domainEventRegisterAny(dom, libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE, lifecycleCallback, domainChangedEvent)
+            eventRegistered = True
+
+            self.assertFalse(libvirtEvents.is_idle())
+
+            # Stop the VM.
+            dom.destroy()
+            domainStopped = True
+
+            # Ensure domain stopped event is received.
+            await asyncio.wait_for(domainChangedEvent.wait(), 2)
+            self.assertEqual([libvirt.VIR_DOMAIN_SHUTOFF, libvirt.VIR_DOMAIN_SHUTOFF_DESTROYED], dom.state())
+
+            # Start the VM.
+            domainChangedEvent.clear()
+            domainStopped = False
+            dom.create()
+
+            # Ensure domain started event is received.
+            await asyncio.wait_for(domainChangedEvent.wait(), 2)
+            self.assertEqual([libvirt.VIR_DOMAIN_RUNNING, libvirt.VIR_DOMAIN_RUNNING_BOOTED], dom.state())
+            self.assertFalse(libvirtEvents.is_idle())
+
+            # Deregister the VM start/stopped event handler.
+            eventRegistered = False
+            conn.domainEventDeregisterAny(libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE)
+
+            # Wait for event queue to clear.
+            await libvirtEvents.drain()
+
+            # Make sure event queue is cleared.
+            self.assertTrue(libvirtEvents.is_idle())
+
+        finally:
+            if eventRegistered:
+                conn.domainEventDeregisterAny(libvirt.VIR_DOMAIN_EVENT_ID_LIFECYCLE)
+
+            if domainStopped:
+                dom.create()
+
+    @pytest.mark.separate_process
+    def testEventsWithManualLoopSetup(self):
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        loop.run_until_complete(self._run(register=True))
+
+        loop.close()
+        asyncio.set_event_loop(None)
+
+    @pytest.mark.separate_process
+    @unittest.skipIf(sys.version_info < (3,7), "test requires Python 3.7+")
+    def testEventsWithAsyncioRun(self):
+        asyncio.run(self._run(register=True))
+
+    @pytest.mark.separate_process
+    @unittest.skipIf(sys.version_info >= (3,10), "test not compatible with Python 3.10+")
+    def testEventsPreInit(self):
+        # Initialize libvirt events before setting the event loop. This is not recommended.
+        # But is supported in older version of Python for the sake of back-compat.
+        loop = asyncio.new_event_loop()
+        libvirtaio.virEventRegisterAsyncIOImpl(loop)
+        asyncio.set_event_loop(loop)
+
+        loop.run_until_complete(self._run(register=False))
+
+        loop.close()
+        asyncio.set_event_loop(None)
+
+    @pytest.mark.separate_process
+    def testEventsImplicitLoopInit(self):
+        # Allow virEventRegisterAsyncIOImpl() to init the event loop by calling
+        # asyncio.get_event_loop(). This is not recommended and probably only works by
+        # accident. But is supported for now for the sake of back-compat. For Python
+        # 3.10+, asyncio will report deprecation warnings.
+        libvirtaio.virEventRegisterAsyncIOImpl()
+        loop = asyncio.get_event_loop()
+
+        loop.run_until_complete(self._run(register=False))
+
+        loop.close()
+        asyncio.set_event_loop(None)