Refactor duplicate XML/UUID logic into shared module

tests/test_vm_manager_cluster.py

@@ -11,6 +11,7 @@
 import pytest
 
 from vm_manager import vm_manager_cluster as vmc
+from vm_manager.exceptions import UuidConflictError
 from vm_manager.helpers.libvirt import LibVirtManager
 
 TESTDATA_XML_PATH = os.path.join(
@@ -23,6 +24,22 @@
 
 
 def _read_test_xml():
+    """Read the test XML template with UUID stripped.
+
+    Removing the UUID ensures each test VM gets a fresh random UUID
+    and avoids collisions with existing VMs on the cluster.
+    """
+    with open(TESTDATA_XML_PATH) as f:
+        xml = f.read()
+    root = ElementTree.fromstring(xml)
+    uuid_el = root.find("uuid")
+    if uuid_el is not None:
+        root.remove(uuid_el)
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _read_test_xml_with_uuid():
+    """Read the test XML template preserving the predefined UUID."""
     with open(TESTDATA_XML_PATH) as f:
         return f.read()
 
@@ -252,19 +269,33 @@ def test_name_replaced(self):
         result = vmc._create_xml(xml, "myvm")
         assert "<name>myvm</name>" in result
 
-    def test_uuid_preserved_when_provided(self):
-        xml = _read_test_xml()
+    def test_predefined_uuid_preserved(self):
+        xml = _read_test_xml_with_uuid()
         result = vmc._create_xml(xml, "myvm")
-        assert "7b48b1fe-066a-41a6-aef4-f0a9c028f719" in result
+        root = ElementTree.fromstring(result)
+        assert root.findtext("uuid") == "7b48b1fe-066a-41a6-aef4-f0a9c028f719"
 
-    def test_uuid_generated_when_not_provided(self):
-        xml = _read_test_xml()
-        xml = xml.replace(
-            "<uuid>7b48b1fe-066a-41a6-aef4-f0a9c028f719</uuid>", ""
-        )
+    def test_empty_uuid_generates_new(self):
+        xml = _read_test_xml_with_uuid()
+        xml = xml.replace("7b48b1fe-066a-41a6-aef4-f0a9c028f719", "")
         result = vmc._create_xml(xml, "myvm")
-        assert "<uuid>" in result
-        assert "7b48b1fe-066a-41a6-aef4-f0a9c028f719" not in result
+        root = ElementTree.fromstring(result)
+        vm_uuid = root.findtext("uuid")
+        assert vm_uuid
+        assert vm_uuid != ""
+
+    def test_missing_uuid_generates_new(self):
+        xml = _read_test_xml()
+        root = ElementTree.fromstring(xml)
+        uuid_el = root.find("uuid")
+        if uuid_el is not None:
+            root.remove(uuid_el)
+        xml_no_uuid = ElementTree.tostring(root).decode()
+        result = vmc._create_xml(xml_no_uuid, "myvm")
+        root = ElementTree.fromstring(result)
+        vm_uuid = root.findtext("uuid")
+        assert vm_uuid
+        assert vm_uuid != ""
 
     def test_rbd_disk_added(self):
         xml = _read_test_xml()
@@ -396,6 +427,51 @@ def test_remove_disabled_vm(self, vm_name, qcow2_image):
         assert vmc.status(vm_name) == "Undefined"
 
 
+# ── list_all_uuids ──────────────────────────────────────────────────
+
+
+class TestListAllUuids:
+    def test_returns_dict(self):
+        result = vmc.list_all_uuids()
+        assert isinstance(result, dict)
+
+    def test_created_vm_uuid_appears(self, created_vm):
+        uuids = vmc.list_all_uuids()
+        assert created_vm in uuids.values()
+
+
+# ── UUID collision ──────────────────────────────────────────────────
+
+
+class TestUuidCollision:
+    def test_duplicate_uuid_raises(self, vm_name, qcow2_image):
+        xml = _read_test_xml_with_uuid()
+        # Create first VM with the predefined UUID
+        vmc.create(
+            {
+                "name": vm_name,
+                "image": qcow2_image,
+                "base_xml": xml,
+                "force": True,
+            }
+        )
+        # Attempt to create second VM with same UUID — should conflict
+        second_name = vm_name + "dup"
+        with pytest.raises(UuidConflictError):
+            vmc.create(
+                {
+                    "name": second_name,
+                    "image": qcow2_image,
+                    "base_xml": xml,
+                }
+            )
+        # Clean up second VM if somehow created
+        try:
+            vmc.remove(second_name)
+        except Exception:
+            pass
+
+
 # ── Snapshots ────────────────────────────────────────────────────────
 
 

tests/test_vm_manager_libvirt.py

@@ -1,9 +1,13 @@
 # Copyright (C) 2025, RTE (http://www.rte-france.com)
 # SPDX-License-Identifier: Apache-2.0
 
+import xml.etree.ElementTree as ElementTree
+
 import libvirt
+import pytest
 
 from vm_manager import vm_manager_libvirt as vml
+from vm_manager.exceptions import UuidConflictError
 
 
 class TestListVms:
@@ -19,22 +23,37 @@ def test_name_replaced(self, vm_xml_path):
         result = vml._create_xml(xml, "myvm")
         assert "<name>myvm</name>" in result
 
-    def test_uuid_preserved_when_provided(self, vm_xml_path):
+    def test_predefined_uuid_preserved(self, vm_xml_path):
         with open(vm_xml_path) as f:
             xml = f.read()
         result = vml._create_xml(xml, "myvm")
-        assert "7b48b1fe-066a-41a6-aef4-f0a9c028f719" in result
+        root = ElementTree.fromstring(result)
+        assert root.findtext("uuid") == "7b48b1fe-066a-41a6-aef4-f0a9c028f719"
 
-    def test_uuid_generated_when_not_provided(self, vm_xml_path):
+    def test_empty_uuid_generates_new(self, vm_xml_path):
         with open(vm_xml_path) as f:
             xml = f.read()
-        # Remove the uuid element from the XML
-        xml = xml.replace(
-            "<uuid>7b48b1fe-066a-41a6-aef4-f0a9c028f719</uuid>", ""
-        )
+        xml = xml.replace("7b48b1fe-066a-41a6-aef4-f0a9c028f719", "")
         result = vml._create_xml(xml, "myvm")
-        assert "<uuid>" in result
-        assert "7b48b1fe-066a-41a6-aef4-f0a9c028f719" not in result
+        root = ElementTree.fromstring(result)
+        vm_uuid = root.findtext("uuid")
+        assert vm_uuid
+        assert vm_uuid != ""
+
+    def test_missing_uuid_generates_new(self, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        # Remove the entire <uuid> element
+        root = ElementTree.fromstring(xml)
+        uuid_el = root.find("uuid")
+        if uuid_el is not None:
+            root.remove(uuid_el)
+        xml_no_uuid = ElementTree.tostring(root).decode()
+        result = vml._create_xml(xml_no_uuid, "myvm")
+        root = ElementTree.fromstring(result)
+        vm_uuid = root.findtext("uuid")
+        assert vm_uuid
+        assert vm_uuid != ""
 
 
 class TestCreate:
@@ -114,6 +133,39 @@ def test_started_after_start(self, vm_name, vm_xml_path):
         assert vml.status(vm_name) == "Started"
 
 
+class TestListAllUuids:
+    def test_returns_dict(self):
+        result = vml.list_all_uuids()
+        assert isinstance(result, dict)
+
+    def test_created_vm_uuid_appears(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})
+        uuids = vml.list_all_uuids()
+        assert vm_name in uuids.values()
+
+
+class TestUuidCollision:
+    def test_duplicate_uuid_raises(self, vm_name, vm_xml_path):
+        with open(vm_xml_path) as f:
+            xml = f.read()
+        # Create first VM with the predefined UUID
+        vml.create({"base_xml": xml, "name": vm_name, "autostart": False})
+        # Attempt to create second VM with same UUID
+        with pytest.raises(UuidConflictError):
+            vml.create(
+                {
+                    "base_xml": xml,
+                    "name": vm_name + "dup",
+                    "autostart": False,
+                }
+            )
+        # Clean up the second VM if it was somehow created
+        if vm_name + "dup" in vml.list_vms():
+            vml.remove(vm_name + "dup")
+
+
 class TestAutostart:
     def test_enable_autostart(self, vm_name, vm_xml_path):
         with open(vm_xml_path) as f:

vm_manager/__init__.py

@@ -1,6 +1,8 @@
 # Copyright (C) 2021, RTE (http://www.rte-france.com)
 # SPDX-License-Identifier: Apache-2.0
 
+from .exceptions import UuidConflictError  # noqa: F401
+
 try:
     from .helpers.rbd_manager import RbdManager
     from .helpers.pacemaker import Pacemaker
@@ -12,6 +14,7 @@
 if cluster_mode:
     from .vm_manager_cluster import (
         list_vms,
+        list_all_uuids,
         start,
         stop,
         create,
@@ -38,6 +41,7 @@
 else:
     from .vm_manager_libvirt import (
         list_vms,
+        list_all_uuids,
         console,
         create,
         remove,

vm_manager/exceptions.py

@@ -0,0 +1,10 @@
+# Copyright (C) 2025, RTE (http://www.rte-france.com)
+# SPDX-License-Identifier: Apache-2.0
+
+
+class VmManagerException(Exception):
+    """Base exception for vm_manager errors."""
+
+
+class UuidConflictError(VmManagerException):
+    """Raised when a VM UUID conflicts with an existing VM."""

vm_manager/helpers/libvirt.py

@@ -52,6 +52,12 @@ def list(self):
         """
         return [x.name() for x in self._conn.listAllDomains()]
 
+    def list_uuids(self):
+        """
+        Return dict mapping UUID strings to domain names.
+        """
+        return {d.UUIDString(): d.name() for d in self._conn.listAllDomains()}
+
     def get_virsh_secrets(self):
         """
         Get the virsh secrets

vm_manager/vm_manager_cluster.py

@@ -18,6 +18,7 @@
 from .helpers.rbd_manager import RbdManager
 from .helpers.pacemaker import Pacemaker
 from .helpers.libvirt import LibVirtManager
+from .exceptions import UuidConflictError
 
 XML_PACEMAKER_PATH = "/etc/pacemaker"
 
@@ -35,6 +36,29 @@
 """
 
 
+def list_all_uuids():
+    """
+    Return dict mapping UUID strings to VM names by reading XML
+    metadata from each VM's system disk in the RBD cluster.
+    """
+    uuids = {}
+    with RbdManager(CEPH_CONF, POOL_NAME, NAMESPACE) as rbd:
+        for vm_name in rbd.list_groups():
+            disk_name = OS_DISK_PREFIX + vm_name
+            try:
+                xml_str = rbd.get_image_metadata(disk_name, "xml")
+                xml_root = ElementTree.fromstring(xml_str)
+                vm_uuid = xml_root.findtext("uuid")
+                if vm_uuid:
+                    uuids[vm_uuid] = vm_name
+            except Exception:
+                logger.warning(
+                    "Could not read UUID for VM %s, skipping",
+                    vm_name,
+                )
+    return uuids
+
+
 def _check_name(name):
     """
     Raise ValueError if name is an empty string, contains special
@@ -348,6 +372,23 @@ def create(vm_options_with_nones):
     except KeyError:
         progress = False
 
+    # Check for UUID collision before importing the disk
+    xml = _create_xml(
+        vm_options["base_xml"],
+        vm_options["name"],
+        vm_options.get("disk_bus", "virtio"),
+    )
+    xml_root = ElementTree.fromstring(xml)
+    vm_uuid = xml_root.findtext("uuid")
+    if vm_uuid:
+        existing = list_all_uuids()
+        if vm_uuid in existing:
+            raise UuidConflictError(
+                "UUID {} is already used by VM {}".format(
+                    vm_uuid, existing[vm_uuid]
+                )
+            )
+
     # Create VM group
     if "force" not in vm_options:
         vm_options["force"] = False

vm_manager/vm_manager_libvirt.py

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from .helpers.libvirt import LibVirtManager
+from .exceptions import UuidConflictError
 import xml.etree.ElementTree as ElementTree
 import uuid
 import logging
@@ -22,6 +23,14 @@ def list_vms():
         return lvm.list()
 
 
+def list_all_uuids():
+    """
+    Return dict mapping UUID strings to VM names for all local VMs.
+    """
+    with LibVirtManager() as lvm:
+        return lvm.list_uuids()
+
+
 def _create_xml(xml, vm_name):
     """
     Creates a libvirt configuration file according to xml and
@@ -55,6 +64,17 @@ def create(args):
     """
     xml = _create_xml(args.get("base_xml"), args.get("name"))
 
+    xml_root = ElementTree.fromstring(xml)
+    vm_uuid = xml_root.findtext("uuid")
+    if vm_uuid:
+        existing = list_all_uuids()
+        if vm_uuid in existing:
+            raise UuidConflictError(
+                "UUID {} is already used by VM {}".format(
+                    vm_uuid, existing[vm_uuid]
+                )
+            )
+
     with LibVirtManager() as lvm:
         lvm.define(xml)
         if args.get("autostart"):