Add GitHub actions

.flake8

@@ -1,2 +1,5 @@
 [flake8]
-per-file-ignores = __init__.py:F401
+exclude = build
+per-file-ignores =
+    __init__.py:F401
+    vm_manager_cmd.py:F841

.github/workflows/ci.yml

@@ -0,0 +1,57 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  lint:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Install cqfd
+        run: |
+          git clone https://github.com/savoirfairelinux/cqfd.git /tmp/cqfd
+          cd /tmp/cqfd
+          sudo make install
+
+      - name: Initialize cqfd container
+        run: cqfd init
+
+      - name: Check formatting
+        run: cqfd -b check_format
+
+      - name: Run flake8
+        run: cqfd -b flake
+
+  test:
+    runs-on: ubuntu-24.04
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends \
+            qemu-system-x86 \
+            libvirt-daemon-system \
+            libvirt-dev \
+            pkg-config
+
+      - name: Start libvirtd and grant access
+        run: |
+          sudo systemctl start libvirtd
+          sudo usermod -aG libvirt "$USER"
+
+      - name: Install package with test deps
+        run: pip install ".[test]"
+
+      - name: Run tests
+        run: sg libvirt -c "pytest tests/ -v --tb=short"

pyproject.toml

@@ -17,6 +17,9 @@ dependencies = [
 ]
 readme = "README.md"
 
+[project.optional-dependencies]
+test = ["pytest>=7.0"]
+
 [tool.setuptools]
 packages = ["vm_manager", "vm_manager.helpers", "vm_manager.helpers.tests.pacemaker", "vm_manager.helpers.tests.rbd_manager"]
 

tests/conftest.py

@@ -0,0 +1,45 @@
+# Copyright (C) 2025, RTE (http://www.rte-france.com)
+# SPDX-License-Identifier: Apache-2.0
+
+import os
+import secrets
+
+import pytest
+
+from vm_manager.helpers.libvirt import LibVirtManager
+
+
+@pytest.fixture
+def vm_name():
+    """Generate a unique VM name and ensure cleanup after test."""
+    name = "testvm" + secrets.token_hex(4)
+    yield name
+    with LibVirtManager() as lvm:
+        if name in lvm.list():
+            try:
+                lvm.force_stop(name)
+            except Exception:
+                pass
+            try:
+                lvm.undefine(name)
+            except Exception:
+                pass
+
+
+@pytest.fixture
+def libvirt_conn():
+    """Provide a LibVirtManager connection for the test."""
+    with LibVirtManager() as lvm:
+        yield lvm
+
+
+@pytest.fixture
+def vm_xml_path():
+    """Return the path to the test VM XML template."""
+    return os.path.join(
+        os.path.dirname(__file__),
+        "..",
+        "vm_manager",
+        "testdata",
+        "vm.xml",
+    )

tests/test_libvirt_manager.py

@@ -0,0 +1,113 @@
+# Copyright (C) 2025, RTE (http://www.rte-france.com)
+# SPDX-License-Identifier: Apache-2.0
+
+import libvirt
+import pytest
+
+from vm_manager.helpers.libvirt import LibVirtManager
+
+
+class TestConnection:
+    def test_context_manager(self):
+        with LibVirtManager() as lvm:
+            assert lvm._conn is not None
+            assert lvm._conn.isAlive()
+
+    def test_close(self):
+        lvm = LibVirtManager()
+        lvm.close()
+
+
+class TestList:
+    def test_list_returns_list(self, libvirt_conn):
+        result = libvirt_conn.list()
+        assert isinstance(result, list)
+
+    def test_list_contains_defined_vm(
+        self, libvirt_conn, vm_name, vm_xml_path
+    ):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        assert vm_name in libvirt_conn.list()
+
+
+class TestDefine:
+    def test_define_valid_xml(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        assert vm_name in libvirt_conn.list()
+
+    def test_define_invalid_xml_raises(self, libvirt_conn):
+        with pytest.raises(libvirt.libvirtError):
+            libvirt_conn.define("<invalid/>")
+
+
+class TestUndefine:
+    def test_undefine_removes_domain(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        libvirt_conn.undefine(vm_name)
+        assert vm_name not in libvirt_conn.list()
+
+    def test_undefine_nonexistent_raises(self, libvirt_conn):
+        with pytest.raises(libvirt.libvirtError):
+            libvirt_conn.undefine("nonexistent_vm_xyz")
+
+
+class TestStartStop:
+    def test_start_and_force_stop(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        libvirt_conn.start(vm_name)
+        assert libvirt_conn.status(vm_name) == "Started"
+        libvirt_conn.force_stop(vm_name)
+        assert libvirt_conn.status(vm_name) == "Stopped"
+
+
+class TestStatus:
+    def test_status_stopped(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        assert libvirt_conn.status(vm_name) == "Stopped"
+
+    def test_status_started(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        libvirt_conn.start(vm_name)
+        assert libvirt_conn.status(vm_name) == "Started"
+
+    def test_status_undefined(self, libvirt_conn):
+        assert libvirt_conn.status("nonexistent_vm_xyz") == "Undefined"
+
+
+class TestAutostart:
+    def test_enable_autostart(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        libvirt_conn.set_autostart(vm_name, True)
+        domain = libvirt_conn._conn.lookupByName(vm_name)
+        assert domain.autostart() == 1
+
+    def test_disable_autostart(self, libvirt_conn, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        xml = xml.replace("test0", vm_name)
+        libvirt_conn.define(xml)
+        libvirt_conn.set_autostart(vm_name, True)
+        libvirt_conn.set_autostart(vm_name, False)
+        domain = libvirt_conn._conn.lookupByName(vm_name)
+        assert domain.autostart() == 0

tests/test_vm_manager_libvirt.py

@@ -0,0 +1,127 @@
+# Copyright (C) 2025, RTE (http://www.rte-france.com)
+# SPDX-License-Identifier: Apache-2.0
+
+import libvirt
+
+from vm_manager import vm_manager_libvirt as vml
+
+
+class TestListVms:
+    def test_returns_list(self):
+        result = vml.list_vms()
+        assert isinstance(result, list)
+
+
+class TestCreateXml:
+    def test_name_replaced(self, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        result = vml._create_xml(xml, "myvm")
+        assert "<name>myvm</name>" in result
+
+    def test_uuid_replaced(self, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        result = vml._create_xml(xml, "myvm")
+        # Original UUID should be gone
+        assert "7b48b1fe-066a-41a6-aef4-f0a9c028f719" not in result
+        assert "<uuid>" in result
+
+
+class TestCreate:
+    def test_create_vm(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        args = {
+            "base_xml": xml,
+            "name": vm_name,
+            "autostart": False,
+        }
+        vml.create(args)
+        assert vm_name in vml.list_vms()
+
+    def test_create_with_autostart(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        args = {
+            "base_xml": xml,
+            "name": vm_name,
+            "autostart": True,
+        }
+        vml.create(args)
+        assert vm_name in vml.list_vms()
+        conn = libvirt.open("qemu:///system")
+        domain = conn.lookupByName(vm_name)
+        assert domain.autostart() == 1
+        conn.close()
+
+
+class TestRemove:
+    def test_remove_stopped_vm(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.remove(vm_name)
+        assert vm_name not in vml.list_vms()
+
+    def test_remove_running_vm(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.start(vm_name)
+        vml.remove(vm_name)
+        assert vm_name not in vml.list_vms()
+
+
+class TestStartStop:
+    def test_start(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.start(vm_name)
+        assert vml.status(vm_name) == "Started"
+
+    def test_force_stop(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.start(vm_name)
+        vml.stop(vm_name, force=True)
+        assert vml.status(vm_name) == "Stopped"
+
+
+class TestStatus:
+    def test_stopped_after_create(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        assert vml.status(vm_name) == "Stopped"
+
+    def test_started_after_start(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.start(vm_name)
+        assert vml.status(vm_name) == "Started"
+
+
+class TestAutostart:
+    def test_enable_autostart(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        vml.autostart(vm_name, True)
+        conn = libvirt.open("qemu:///system")
+        domain = conn.lookupByName(vm_name)
+        assert domain.autostart() == 1
+        conn.close()
+
+    def test_disable_autostart(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": True})
+        vml.autostart(vm_name, False)
+        conn = libvirt.open("qemu:///system")
+        domain = conn.lookupByName(vm_name)
+        assert domain.autostart() == 0
+        conn.close()

vm_manager/helpers/pacemaker.py

@@ -430,7 +430,7 @@ def find_resource(resource):
         """
         command = (
             "crm status | "
-            + f'grep -E "^  \* {resource}\\b" | '
+            + f'grep -E "^  \\* {resource}\\b" | '
             + "grep Started | "
             + "awk 'NF>1{print $NF}'"
         )

vm_manager/helpers/tests/rbd_manager/rollback_rbd.py

@@ -85,6 +85,7 @@ def main():
             ts = rbd.get_image_snapshot_timestamp(IMG_NAME, SNAP)
             if ts.tzinfo is None:
                 from datetime import timezone
+
                 ts = ts.replace(tzinfo=timezone.utc)
             print("Snapshot " + SNAP + " timestamp: " + str(ts))
             if (

vm_manager/testdata/vm.xml

@@ -5,7 +5,9 @@
   <os>
     <type arch="x86_64" machine="q35">hvm</type>
   </os>
-  <cpu mode="host-model" check="full"/>
+  <cpu mode="custom">
+    <model fallback="forbid">qemu64</model>
+  </cpu>
   <clock offset="utc">
     <timer name="rtc" tickpolicy="catchup"/>
     <timer name="pit" tickpolicy="delay"/>