From f004552148c850af6a3759fc4b46d982dcd478ce Mon Sep 17 00:00:00 2001 From: catborise Date: Sun, 25 Oct 2026 19:42:19 +0300 Subject: [PATCH] feat(instances): a new VM can get a cloud-init seed with its hostname and SSH keys Cloud images read their first-boot settings with cloud-init. The panel served them only over HTTP, which a VM reaches only with the network and a reverse DNS name for its address. A new VM can now carry them itself: Cloud-init seed (Advanced) on the create form, or cloud_init in the API. The job writes a NoCloud seed, an ISO labelled cidata, into the disk's directory pool as cidata-.iso and puts it in the VM's CD-ROM drive. Its meta-data holds the VM's UUID as instance-id and its name as hostname, written as JSON so a VM named 123 or yes keeps a string hostname; its user-data is what the HTTP datasource serves, with the creating user's SSH keys. The ISO is built with pycdlib and written through a libvirt stream, as the NVRAM copy is, now one helper for both. A seed needs a CD-ROM bus and a directory pool, else the create is refused. A failed create deletes the seed with the new disk, recovery names one left behind, and deleting the VM with its disks deletes its seed, found in the pools by name even when ejected. A clone's drive is left empty: the seed is the source VM's identity. --- README.md | 2 + conf/requirements.txt | 1 + instances/api/serializers.py | 1 + instances/jobs.py | 36 +++++- instances/models.py | 2 + instances/templates/create_instance_w2.html | 5 + instances/tests/test_cloud_init_create.py | 98 +++++++++++++++ instances/tests/test_create_form_input.py | 10 ++ vrtManager/connection.py | 20 +++ vrtManager/create.py | 27 ++++ vrtManager/instance.py | 36 +++--- vrtManager/tests/test_cloud_init_seed.py | 131 ++++++++++++++++++++ vrtManager/tests/test_external_snapshots.py | 3 + 13 files changed, 352 insertions(+), 20 deletions(-) create mode 100644 instances/tests/test_cloud_init_create.py create mode 100644 vrtManager/tests/test_cloud_init_seed.py 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()