diff --git a/.flake8 b/.flake8
index 80676bc..74aad7c 100644
--- a/.flake8
+++ b/.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
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..b91e13d
--- /dev/null
+++ b/.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"
diff --git a/pyproject.toml b/pyproject.toml
index 7e0c4c9..4d2f303 100644
--- a/pyproject.toml
+++ b/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"]
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..c7cf36b
--- /dev/null
+++ b/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",
+ )
diff --git a/tests/test_libvirt_manager.py b/tests/test_libvirt_manager.py
new file mode 100644
index 0000000..4ebc8bb
--- /dev/null
+++ b/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("")
+
+
+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
diff --git a/tests/test_vm_manager_libvirt.py b/tests/test_vm_manager_libvirt.py
new file mode 100644
index 0000000..eeb1301
--- /dev/null
+++ b/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 "myvm" 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 "" 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()
diff --git a/vm_manager/helpers/pacemaker.py b/vm_manager/helpers/pacemaker.py
index 47bdf10..fc6b0db 100644
--- a/vm_manager/helpers/pacemaker.py
+++ b/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}'"
)
diff --git a/vm_manager/helpers/tests/rbd_manager/rollback_rbd.py b/vm_manager/helpers/tests/rbd_manager/rollback_rbd.py
index c473651..1c71f99 100755
--- a/vm_manager/helpers/tests/rbd_manager/rollback_rbd.py
+++ b/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 (
diff --git a/vm_manager/testdata/vm.xml b/vm_manager/testdata/vm.xml
index aef1aee..00ac1c8 100644
--- a/vm_manager/testdata/vm.xml
+++ b/vm_manager/testdata/vm.xml
@@ -5,7 +5,9 @@
hvm
-
+
+ qemu64
+
diff --git a/vm_manager/testdata/wrong_vm_config.xml b/vm_manager/testdata/wrong_vm_config.xml
index 597c32a..25cc22d 100644
--- a/vm_manager/testdata/wrong_vm_config.xml
+++ b/vm_manager/testdata/wrong_vm_config.xml
@@ -3,7 +3,9 @@
hvm
-
+
+ qemu64
+
diff --git a/vm_manager/vm_manager_cmd.py b/vm_manager/vm_manager_cmd.py
index 264e1a2..8cab7f3 100755
--- a/vm_manager/vm_manager_cmd.py
+++ b/vm_manager/vm_manager_cmd.py
@@ -471,9 +471,7 @@ def main():
if vm_manager.cluster_mode:
vm_manager.start(args.name)
else:
- vm_manager.start(
- args.name, autostart=not args.no_autostart
- )
+ vm_manager.start(args.name, autostart=not args.no_autostart)
elif args.command == "stop":
vm_manager.stop(args.name, force=args.force)
elif args.command == "remove":