diff --git a/README.md b/README.md index 578d52b5..a32637b6 100644 --- a/README.md +++ b/README.md @@ -325,6 +325,8 @@ datasource: metadata_urls: [ "http://wvc.example.com/datasource" ] ``` +Without the network: tick **Cloud-init seed** (Advanced) when creating a VM. The VM gets a CD-ROM with its hostname and your SSH keys (a NoCloud seed, `cidata-.iso` in the disk's directory pool), which the image's cloud-init reads at first boot; it is deleted with the VM's disks. + ### LDAP The options below are set in `webvirtcloud/settings.py`. Variants for Active Directory and OpenLDAP are shown; this is a minimal config, see the [django-auth-ldap documentation](https://django-auth-ldap.readthedocs.io) for more. diff --git a/conf/requirements.txt b/conf/requirements.txt index 2c09946b..e8b51520 100644 --- a/conf/requirements.txt +++ b/conf/requirements.txt @@ -18,6 +18,7 @@ lxml>=6.1 openpyxl==3.1.5 paramiko==5.0.0 psycopg[binary]==3.3.6 +pycdlib==1.22.0 python-engineio==4.14.0 python-socketio==5.17.0 segno==1.6.6 diff --git a/instances/api/serializers.py b/instances/api/serializers.py index ce4d2451..28b85a08 100644 --- a/instances/api/serializers.py +++ b/instances/api/serializers.py @@ -83,6 +83,7 @@ class CreateInstanceSerializer(serializers.ModelSerializer): "meta_prealloc", "virtio", "qemu_ga", + "cloud_init", "video", "listener_addr", "submission_key", diff --git a/instances/jobs.py b/instances/jobs.py index c6a367fe..d3170397 100644 --- a/instances/jobs.py +++ b/instances/jobs.py @@ -5,6 +5,7 @@ import re import time from django.core.exceptions import PermissionDenied +from django.template.loader import render_to_string from django.utils.translation import gettext_noop from django.utils.translation import gettext_noop as _ from libvirt import (VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE, VIR_DOMAIN_JOB_NONE, VIR_DOMAIN_SNAPSHOT_LIST_EXTERNAL, @@ -14,7 +15,7 @@ from vrtManager.connection import wvmConnect from vrtManager.create import wvmCreate from vrtManager.instance import wvmInstance -from accounts.models import UserInstance +from accounts.models import UserInstance, UserSSHKey from appsettings.settings import app_settings from computes.models import Compute from instances import utils @@ -613,6 +614,17 @@ class CloneInstance(Kind): return _log(op, state, f"clone from {source}", vm=op.intent["name"]) +# pools whose volumes are files a CD-ROM can read +SEED_POOL_TYPES = ("dir", "fs", "netfs") + + +def _seed_pool(conn, data): + """The pool of a new VM's cloud-init seed: the one of its disk.""" + if data["storage"]: + return conn.get_storage(data["storage"]) + return conn.get_storage_by_vol_path(conn.get_volume_path(data["images"].split(",")[0])) + + @register class CreateInstance(Kind): name = "instance.create" @@ -622,7 +634,7 @@ class CreateInstance(Kind): def normalize(self, raw): # what the API leaves out means what create_instance did by default params = {"template": "", "images": "", "firmware": "", "net_model": "virtio", "add_cdrom": "sata", - "add_input": "default", "meta_prealloc": False, "hdd_size": 0} + "add_input": "default", "meta_prealloc": False, "hdd_size": 0, "cloud_init": False} params.update({k: v for k, v in raw.items() if k not in ("submission_key", "create", "csrfmiddlewaretoken")}) params.update(compute=int(raw["compute"]), devices=[list(d) for d in raw.get("devices", [])]) if "UEFI" in params["firmware"] and "secboot" in params["firmware"].partition(":")[2]: @@ -653,6 +665,12 @@ class CreateInstance(Kind): raise util.OperationError(_("Invalid cache mode")) if data["add_cdrom"] != "None" and data["add_cdrom"] not in cdrom_buses: raise util.OperationError(_("A CD-ROM cannot use the %(bus)s bus") % {"bus": data["add_cdrom"]}) + if data["cloud_init"]: + if data["add_cdrom"] == "None": + raise util.OperationError(_("The cloud-init seed is read from a CD-ROM: choose its bus")) + pool = _seed_pool(conn, data) + if util.get_xml_path(pool.XMLDesc(0), "/pool/@type") not in SEED_POOL_TYPES: + raise util.OperationError(_("The cloud-init seed needs a directory storage pool for the disk")) try: utils.nic_macs(data["mac"], data["networks"]) except ValueError as err: @@ -715,6 +733,14 @@ class CreateInstance(Kind): for vol, (device, bus) in zip(data["images"].split(","), data["devices"]): volume_list.append(volume(conn.get_volume_path(vol), device, bus)) + if data["cloud_init"]: + ctx.set_phase(gettext_noop("Writing the cloud-init seed")) + keys = UserSSHKey.objects.filter(user=op.user).values_list("keypublic", flat=True) + user_data = render_to_string("user_data", {"instance_keys": list(keys)}) + seed = conn.create_seed(_seed_pool(conn, data).name(), op.intent["uuid"], data["name"], user_data) + created.append(seed) + volume_list.append(volume(seed, "cdrom", data["add_cdrom"])) + ctx.set_phase(gettext_noop("Defining the VM")) conn.create_instance( name=data["name"], @@ -754,8 +780,10 @@ class CreateInstance(Kind): def reconcile(self, op): data = op.params - names = [data["name"] + ext for ext in ("", ".img", ".qcow", ".qcow2")] - return vm_outcome(op, owner=False, volumes=names if data["hdd_size"] or data["template"] else []) + names = [data["name"] + ext for ext in ("", ".img", ".qcow", ".qcow2")] if data["hdd_size"] or data["template"] else [] + if data.get("cloud_init"): + names.append(f"cidata-{op.intent['uuid']}.iso") + return vm_outcome(op, owner=False, volumes=names) def log_entry(self, op, state): return _log(op, state, "create") diff --git a/instances/models.py b/instances/models.py index 3295bd59..a025893e 100644 --- a/instances/models.py +++ b/instances/models.py @@ -375,6 +375,8 @@ class CreateInstance(models.Model): qemu_ga = models.BooleanField(default=False) mac = models.CharField(max_length=255, blank=True) # one per NIC, comma separated add_cdrom = models.CharField(max_length=16) + # a NoCloud seed on a CD-ROM: the hostname and the creating user's SSH keys + cloud_init = models.BooleanField(default=False) add_input = models.CharField(max_length=16) video = models.CharField( max_length=16, error_messages={"required": _("Please select a video driver")} diff --git a/instances/templates/create_instance_w2.html b/instances/templates/create_instance_w2.html index f584460f..d981103a 100644 --- a/instances/templates/create_instance_w2.html +++ b/instances/templates/create_instance_w2.html @@ -329,6 +329,11 @@ +
+ + +
{% trans "The hostname and your SSH keys on the CD-ROM, for cloud images" %}
+
{% if virtio_support %}
diff --git a/instances/tests/test_cloud_init_create.py b/instances/tests/test_cloud_init_create.py new file mode 100644 index 00000000..c9a6ac15 --- /dev/null +++ b/instances/tests/test_cloud_init_create.py @@ -0,0 +1,98 @@ +"""A new VM can get a cloud-init NoCloud seed: its hostname and the creating +user's SSH keys on a CD-ROM, read at first boot without the network.""" + +from types import SimpleNamespace +from unittest.mock import patch + +import libvirt +from accounts.models import UserSSHKey +from appsettings.settings import get_settings +from computes.models import Compute +from django.contrib.auth import get_user_model +from django.test import TransactionTestCase +from instances.jobs import CreateInstance +from operations.jobs import Context +from operations.models import Operation, acquire +from vrtManager.util import OperationError + +from instances.tests.test_create_job import BODY + +SEED = "/srv/default/cidata-x.iso" + + +class CloudInitCreateTestCase(TransactionTestCase): + def setUp(self): + get_settings() + self.admin = get_user_model().objects.create_superuser("ci_admin", "a@example.com", "pw") + self.client.force_login(self.admin) + UserSSHKey.objects.create(user=self.admin, keyname="laptop", keypublic="ssh-ed25519 AAAA laptop") + self.compute = Compute.objects.create(name="ci-c", hostname="192.0.2.41", login="root", password="", type=1) + patcher = patch("instances.jobs.wvmCreate") + self.conn = patcher.start().return_value + self.addCleanup(patcher.stop) + self.conn.get_instances.return_value = [] + self.conn.get_cache_modes.return_value = {"none": "Disable cache"} + self.conn.get_disk_bus_types.return_value = ["virtio", "sata"] + self.conn.create_volume.return_value = "/srv/default/api-vm.qcow2" + self.conn.create_seed.return_value = SEED + self.conn.get_storage.return_value.XMLDesc.return_value = "" + self.conn.get_storage.return_value.name.return_value = "default" + + def post(self, **fields): + url = f"/api/v1/computes/{self.compute.pk}/instances/create/x86_64/q35/" + return self.client.post(url, {**BODY, **fields}, content_type="application/json") + + def run_job(self): + op = Operation.objects.get() + acquire("wvc-test:1:a", "wvc-test") + Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker="wvc-test:1:a") + op.refresh_from_db() + CreateInstance().run(op, Context(op, SimpleNamespace(identity="wvc-test:1:a"))) + return op + + def test_the_seed_goes_into_the_disks_pool_and_a_cdrom(self): + self.assertEqual(self.post(cloud_init=True).status_code, 202) + op = self.run_job() + pool, uuid, hostname, user_data = self.conn.create_seed.call_args.args + self.assertEqual((pool, uuid, hostname), ("default", op.intent["uuid"], "api-vm")) + # the same user-data the HTTP datasource serves + self.assertEqual(user_data.split(), ["#cloud-config", "ssh_authorized_keys:", "-", "ssh-ed25519", "AAAA", "laptop"]) + cdrom = self.conn.create_instance.call_args.kwargs["volumes"][-1] + self.assertEqual((cdrom["path"], cdrom["device"], cdrom["bus"]), (SEED, "cdrom", "sata")) + + def test_without_cloud_init_there_is_no_seed(self): + self.post() + self.run_job() + self.conn.create_seed.assert_not_called() + + def test_a_seed_needs_a_cdrom_and_a_directory_pool(self): + kind = CreateInstance() + params = kind.normalize({**BODY, "compute": self.compute.pk, "arch": "x86_64", "machine": "q35", + "cloud_init": True, "add_cdrom": "None"}) + with self.assertRaisesRegex(OperationError, "CD-ROM"): + kind.validate(self.admin, params) + self.conn.get_storage.return_value.XMLDesc.return_value = "" + self.assertEqual(self.post(cloud_init=True, submission_key="key-rbd00001").status_code, 400) + self.assertFalse(Operation.objects.exists()) + + def test_a_failed_create_deletes_the_seed_too(self): + self.post(cloud_init=True) + self.conn.create_instance.side_effect = libvirt.libvirtError("define failed") + with self.assertRaises(libvirt.libvirtError): + self.run_job() + deleted = [c.args[0] for c in self.conn.delete_volume.call_args_list] + self.assertEqual(deleted, ["/srv/default/api-vm.qcow2", SEED]) + + @patch("instances.jobs.wvmConnect") + def test_recovery_names_a_seed_left_behind(self, conn): + op = Operation(params={**BODY, "template": "", "cloud_init": True}, + intent={"compute": self.compute.pk, "uuid": "abababab-0000-1111-2222-333333333333", + "name": "api-vm"}) + missing = libvirt.libvirtError("no domain") + missing.get_error_code = lambda: libvirt.VIR_ERR_NO_DOMAIN + conn.return_value.wvm.lookupByUUIDString.side_effect = missing + conn.return_value.wvm.listAllStoragePools.return_value = [ + SimpleNamespace(listVolumes=lambda: ["cidata-abababab-0000-1111-2222-333333333333.iso"])] + outcome = CreateInstance().reconcile(op) + self.assertEqual(outcome.state, Operation.UNCERTAIN) + self.assertIn("cidata-abababab-0000-1111-2222-333333333333.iso", outcome.error) diff --git a/instances/tests/test_create_form_input.py b/instances/tests/test_create_form_input.py index a1d5e137..841ddd68 100644 --- a/instances/tests/test_create_form_input.py +++ b/instances/tests/test_create_form_input.py @@ -76,6 +76,16 @@ class CreateFormInputTests(TestCase): self.assertRedirects(response, reverse("instances", args=[compute.id]), fetch_redirect_response=False) self.assertEqual(Operation.objects.get().kind, "instance.create") + def test_the_cloud_init_seed_is_chosen_on_the_form(self): + _, response = render_create_page(self) + (box,) = html.fromstring(response.content).xpath("//input[@name='cloud_init']") + self.assertIsNone(box.get("checked")) + # the test driver has no CD-ROM bus to pass the job's checks: what the form hands over + queued = Operation(pk=1, params={"machine": "pc"}) + with patch("instances.views.jobs.submit", return_value=queued) as submit: + render_create_page(self, valid(disk_source="new", hdd_size="10", storage="default-pool", cloud_init="true")) + self.assertIs(submit.call_args.args[2]["cloud_init"], True) + def test_two_disk_sources_are_refused(self): # the page sends only the chosen source; a crafted request with two queues nothing _, response = render_create_page(self, valid( diff --git a/vrtManager/connection.py b/vrtManager/connection.py index 45b8036b..bb2a03fc 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -1,4 +1,5 @@ import contextlib +import io import os import re import socket @@ -919,6 +920,25 @@ class wvmConnect(object): with contextlib.suppress(libvirtError): pool.destroy() + def upload_volume(self, pool, name, data): + """A new raw volume in pool holding data, written through a libvirt + stream, which every connection type carries. A failed write leaves + no volume.""" + vol = pool.createXML( + f"{util.xml_escape(name)}{len(data)}" + "", + 0, + ) + try: + stream = self.wvm.newStream(0) + vol.upload(stream, 0, len(data), 0) + stream.sendAll(lambda _stream, size, buffer: buffer.read(size), io.BytesIO(data)) + stream.finish() + except Exception: + vol.delete(0) + raise + return vol + @staticmethod def _find_volume(pool, name): try: diff --git a/vrtManager/create.py b/vrtManager/create.py index f3f47bed..6c047ec2 100644 --- a/vrtManager/create.py +++ b/vrtManager/create.py @@ -1,10 +1,32 @@ import contextlib +import io +import json import string +import pycdlib from vrtManager import util from vrtManager.connection import wvmConnect +def nocloud_seed(instance_id, hostname, user_data): + """A cloud-init NoCloud seed: an ISO labelled cidata with the VM's + meta-data and user-data, read at first boot without the network. + cloud-init runs again only for a new instance-id.""" + iso = pycdlib.PyCdlib() + iso.new(interchange_level=3, joliet=3, rock_ridge="1.09", vol_ident="cidata") + # JSON is YAML whose values stay strings: a VM named 123 or yes keeps its name + meta_data = json.dumps({"instance-id": instance_id, "local-hostname": hostname}) + files = {"meta-data": meta_data, "user-data": user_data} + for name, text in files.items(): + data = text.encode() + iso_name = name.replace("-", "").upper() + iso.add_fp(io.BytesIO(data), len(data), f"/{iso_name}.;1", rr_name=name, joliet_path=f"/{name}") + out = io.BytesIO() + iso.write_fp(out) + iso.close() + return out.getvalue() + + def get_rbd_storage_data(stg): xml = stg.XMLDesc(0) ceph_user = util.get_xml_path(xml, "/pool/source/auth/@username") @@ -144,6 +166,11 @@ class wvmCreate(wvmConnect): vol = self.get_volume_by_path(path) vol.delete() + def create_seed(self, pool, uuid, hostname, user_data): + """The VM's NoCloud seed as a volume cidata-.iso in the pool; its path.""" + seed = nocloud_seed(uuid, hostname, user_data) + return self.upload_volume(self.get_storage(pool), f"cidata-{uuid}.iso", seed).path() + def create_instance( self, name, diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 227da6af..451c6e18 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -1,6 +1,5 @@ import base64 import contextlib -import io import ipaddress import json import logging @@ -200,20 +199,7 @@ class wvmInstances(wvmConnect): vol.download(stream, 0, 0, 0) stream.recvAll(lambda _stream, data, out: out.append(data), chunks) stream.finish() - data = b"".join(chunks) - copy = target.createXML( - f"{util.xml_escape(name)}{len(data)}" - "", - 0, - ) - try: - stream = self.wvm.newStream(0) - copy.upload(stream, 0, len(data), 0) - stream.sendAll(lambda _stream, size, buffer: buffer.read(size), io.BytesIO(data)) - stream.finish() - except Exception: - copy.delete(0) - raise + self.upload_volume(target, name, b"".join(chunks)) return name def delete_nvram(self, name): @@ -1525,10 +1511,14 @@ class wvmInstance(wvmConnect): def split_disk_paths_by_use(self): """This VM's disk paths, split into (only this VM, also used by another domain). With external snapshots, the files below a disk that the - snapshots were taken on, and their memory files, are the VM's too.""" + snapshots were taken on, and their memory files, are the VM's too, and + so is its cloud-init seed, in a CD-ROM drive or ejected.""" self.refresh_instance_pools() own = [disk["path"] for disk in self.get_disk_devices() if disk["path"]] own += self._snapshot_chain_files(own) + seed = self._seed_path() + if seed: + own.append(seed) used_elsewhere = self.paths_used_by_other_domains(own) # a kept file keeps the files below it for path in [path for path in own if path in used_elsewhere]: @@ -1539,6 +1529,15 @@ class wvmInstance(wvmConnect): [path for path in own if path in used_elsewhere], ) + def _seed_path(self): + """The path of this VM's cloud-init seed (cidata-.iso) in an active pool, or None.""" + name = f"cidata-{self.get_uuid()}.iso" + for pool in self.wvm.listAllStoragePools(VIR_CONNECT_LIST_STORAGE_POOLS_ACTIVE): + vol = self._find_volume(pool, name) + if vol is not None: + return vol.path() + return None + def _snapshot_chain_files(self, tops): """The files of this VM's external snapshots that these disk images grow from or that grow from them: the backing files snapshots were @@ -1974,6 +1973,11 @@ class wvmInstance(wvmConnect): raise ValueError("An instance with the clone name already exists") tree = etree.fromstring(self._XMLDesc(PERSISTENT_XML)) + # the cloud-init seed holds the VM's own instance-id and hostname: the clone's drive stays empty + seed = f"cidata-{tree.findtext('uuid')}.iso" + for source in tree.xpath("devices/disk[@device='cdrom']/source"): + if os.path.basename(source.get("file") or "") == seed: + source.getparent().remove(source) tree.find("name").text = clone_data["name"] uuid_el = tree.find("uuid") if uuid_el is None: diff --git a/vrtManager/tests/test_cloud_init_seed.py b/vrtManager/tests/test_cloud_init_seed.py new file mode 100644 index 00000000..c610c19f --- /dev/null +++ b/vrtManager/tests/test_cloud_init_seed.py @@ -0,0 +1,131 @@ +"""A new VM's cloud-init NoCloud seed: an ISO labelled cidata with its +meta-data and user-data, written into a storage pool as a volume.""" + +import io +import json +import unittest +from unittest.mock import MagicMock + +from django.conf import settings + +if not settings.configured: + settings.configure(MAC_OUI="52:54:10") + +import pycdlib +from libvirt import VIR_ERR_NO_STORAGE_VOL, libvirtError +from vrtManager.connection import wvmConnect +from vrtManager.create import nocloud_seed, wvmCreate + +UUID = "11111111-2222-3333-4444-555555555555" +USER_DATA = "#cloud-config\nssh_authorized_keys:\n - ssh-ed25519 AAAA laptop\n" + + +def read_iso(data): + iso = pycdlib.PyCdlib() + iso.open_fp(io.BytesIO(data)) + files = {} + for name in ("meta-data", "user-data"): + out = io.BytesIO() + iso.get_file_from_iso_fp(out, rr_path=f"/{name}") + files[name] = out.getvalue().decode() + out = io.BytesIO() + iso.get_file_from_iso_fp(out, joliet_path=f"/{name}") + assert out.getvalue().decode() == files[name] + label = iso.pvd.volume_identifier.decode().strip() + iso.close() + return label, files + + +class SeedTestCase(unittest.TestCase): + def test_the_seed_carries_the_vms_identity_and_user_data(self): + label, files = read_iso(nocloud_seed(UUID, "web-1", USER_DATA)) + # cloud-init finds the seed by its label, and runs again only for a new instance-id + self.assertEqual(label, "cidata") + self.assertEqual(json.loads(files["meta-data"]), {"instance-id": UUID, "local-hostname": "web-1"}) + self.assertEqual(files["user-data"], USER_DATA) + + def test_any_vm_name_stays_a_string_hostname(self): + # YAML would read 123 as a number, yes as true and null as nothing + import yaml + + for name in ("123", "yes", "null", "1e3", "web.example"): + with self.subTest(name=name): + meta = yaml.safe_load(read_iso(nocloud_seed(UUID, name, USER_DATA))[1]["meta-data"]) + self.assertEqual(meta, {"instance-id": UUID, "local-hostname": name}) + + def test_the_seed_goes_into_the_pool_as_a_volume(self): + conn = wvmCreate.__new__(wvmCreate) + conn.wvm = MagicMock() + pool = conn.wvm.storagePoolLookupByName.return_value + pool.createXML.return_value.path.return_value = f"/var/lib/libvirt/images/cidata-{UUID}.iso" + sent = [] + conn.wvm.newStream.return_value.sendAll.side_effect = lambda handler, buffer: sent.append(buffer.read()) + path = conn.create_seed("default", UUID, "web-1", USER_DATA) + self.assertEqual(path, f"/var/lib/libvirt/images/cidata-{UUID}.iso") + conn.wvm.storagePoolLookupByName.assert_called_once_with("default") + xml = pool.createXML.call_args.args[0] + self.assertIn(f"cidata-{UUID}.iso", xml) + self.assertEqual(json.loads(read_iso(sent[0])[1]["meta-data"])["instance-id"], UUID) + self.assertIn(f"{len(sent[0])}", xml) + + +class UploadVolumeTestCase(unittest.TestCase): + def test_a_failed_write_leaves_no_volume(self): + conn = wvmConnect.__new__(wvmConnect) + conn.wvm = MagicMock() + conn.wvm.newStream.return_value.finish.side_effect = RuntimeError("stream broke") + pool = MagicMock() + with self.assertRaises(RuntimeError): + conn.upload_volume(pool, "a.iso", b"data") + pool.createXML.return_value.delete.assert_called_once_with(0) + + +class DeleteWithDisksTestCase(unittest.TestCase): + def vm(self, seed_in_pool=True): + from vrtManager.instance import wvmInstance + + vm = wvmInstance.__new__(wvmInstance) + vm.wvm = MagicMock() + vm.refresh_instance_pools = MagicMock() + vm.get_uuid = lambda: UUID + vm.get_disk_devices = lambda: [{"path": "/pool/web-1.qcow2"}] + vm._snapshot_chain_files = lambda own: [] + vm.paths_used_by_other_domains = lambda own: set() + pool = MagicMock() + if seed_in_pool: + pool.storageVolLookupByName.return_value.path.return_value = f"/pool/cidata-{UUID}.iso" + else: + missing = libvirtError("no volume") + missing.get_error_code = lambda: VIR_ERR_NO_STORAGE_VOL + pool.storageVolLookupByName.side_effect = missing + vm.wvm.listAllStoragePools.return_value = [pool] + return vm, pool + + def test_the_vms_own_seed_goes_with_its_disks_even_when_ejected(self): + # found in the pools by its name, not in a CD-ROM drive: one may hold an installer now + vm, pool = self.vm() + self.assertEqual(vm.split_disk_paths_by_use(), (["/pool/web-1.qcow2", f"/pool/cidata-{UUID}.iso"], [])) + pool.storageVolLookupByName.assert_called_once_with(f"cidata-{UUID}.iso") + + def test_a_vm_without_a_seed_deletes_only_its_disks(self): + vm, _ = self.vm(seed_in_pool=False) + self.assertEqual(vm.split_disk_paths_by_use(), (["/pool/web-1.qcow2"], [])) + + +class CloneTestCase(unittest.TestCase): + def test_a_clone_does_not_take_the_vms_seed(self): + # the seed holds the VM's instance-id and hostname: the clone's CD-ROM is left empty + from vrtManager.instance import wvmInstance + + vm = wvmInstance.__new__(wvmInstance) + vm.get_instances = lambda: [] + vm.get_uuid = lambda: UUID + vm._XMLDesc = lambda flags: ( + "web-1" + UUID + "" + "" + "" + "") + tree, _ = vm._clone_definition({"name": "web-2"}, "99999999-2222-3333-4444-555555555555") + sources = [d.find("source").get("file") if d.find("source") is not None else None + for d in tree.findall("devices/disk")] + self.assertEqual(sources, [None, "/iso/alma10.iso"]) diff --git a/vrtManager/tests/test_external_snapshots.py b/vrtManager/tests/test_external_snapshots.py index 0b892b38..9881f6f8 100644 --- a/vrtManager/tests/test_external_snapshots.py +++ b/vrtManager/tests/test_external_snapshots.py @@ -765,6 +765,9 @@ class DestroyFilesTestCase(unittest.TestCase): p.get_volume_by_path = volume pool = MagicMock(**{"isActive.return_value": True}) pool.listAllVolumes.return_value = [volume(path) for path in backing] + no_seed = libvirtError("no volume") + no_seed.err = (libvirt.VIR_ERR_NO_STORAGE_VOL,) + pool.storageVolLookupByName.side_effect = no_seed # the VM has no cloud-init seed p.wvm.listAllStoragePools.return_value = [pool] p.wvm.listAllDomains.return_value = [] return p.split_disk_paths_by_use()