diff --git a/instances/test_live_dataloss.py b/instances/test_live_dataloss.py index 6764013a..494fa932 100644 --- a/instances/test_live_dataloss.py +++ b/instances/test_live_dataloss.py @@ -12,6 +12,7 @@ import unittest import libvirt from computes.models import Compute from django.contrib.auth import get_user_model +from django.contrib.messages import get_messages from django.test import TestCase from django.urls import reverse from vrtManager.connection import connection_manager @@ -121,6 +122,35 @@ class LiveDataLossTestCase(TestCase): self.assertTrue(livetest.volume_exists(self.conn, path), f"{path} was deleted") self.assertIn("s1.newer", dom.snapshotListNames(0)) + # R-13: internal snapshots of UEFI VMs + + def _uefi_internal_snapshot(self, running): + base = livetest.create_volume(self.conn, P + "r13") + dom, inst = self.vm("r13", [base], uefi=True) + if running: + dom.create() + os_before = self.os_xml(dom) + + response = self.post("snapshot", inst, {"name": "snap"}) + + # Taken where libvirt supports it, otherwise refused with a message; + # never a server error and never a rewritten loader or NVRAM. + self.assertEqual(response.status_code, 302) + if dom.snapshotNum(0) == 0: + self.assertTrue(list(get_messages(response.wsgi_request))) + self.assertEqual(self.os_xml(dom), os_before) + + def os_xml(self, dom): + from lxml import etree + + return etree.tostring(etree.fromstring(dom.XMLDesc(libvirt.VIR_DOMAIN_XML_INACTIVE)).find("os")) + + def test_r13_uefi_internal_snapshot_of_a_stopped_vm(self): + self._uefi_internal_snapshot(running=False) + + def test_r13_uefi_internal_snapshot_of_a_running_vm(self): + self._uefi_internal_snapshot(running=True) + # R-14: destroy def test_r14_destroy_keeps_a_volume_another_vm_uses(self): diff --git a/instances/views.py b/instances/views.py index bcc34b5d..24ab3c1d 100755 --- a/instances/views.py +++ b/instances/views.py @@ -1227,9 +1227,14 @@ def snapshot(request, pk): ): name = request.POST.get("name", "") desc = request.POST.get("description", "") - instance.proxy.create_snapshot(name, desc) - msg = _("Create snapshot: %(snap)s") % {"snap": name} - addlogmsg(request.user.username, instance.compute.name, instance.name, msg) + try: + instance.proxy.create_snapshot(name, desc) + except libvirtError as err: + # e.g. UEFI firmware whose NVRAM format does not allow internal snapshots + messages.error(request, _("Snapshot was not created: %(err)s") % {"err": err}) + else: + msg = _("Create snapshot: %(snap)s") % {"snap": name} + addlogmsg(request.user.username, instance.compute.name, instance.name, msg) return redirect(request.META.get("HTTP_REFERER") + "#managesnapshot") @@ -1261,7 +1266,11 @@ def revert_snapshot(request, pk): "instances.snapshot_instances" ): snap_name = request.POST.get("name", "") - instance.proxy.snapshot_revert(snap_name) + try: + instance.proxy.snapshot_revert(snap_name) + except libvirtError as err: + messages.error(request, _("Snapshot was not reverted: %(err)s") % {"err": err}) + return redirect(request.META.get("HTTP_REFERER") + "#managesnapshot") msg = _("Successful revert snapshot: ") msg += snap_name messages.success(request, msg) diff --git a/vrtManager/instance.py b/vrtManager/instance.py index 38531516..14a446a2 100644 --- a/vrtManager/instance.py +++ b/vrtManager/instance.py @@ -1367,15 +1367,12 @@ class wvmInstance(wvmConnect): state, time.time(), ) - changed = self.change_snapshot_xml() - try: - xml += self._XMLDesc(VIR_DOMAIN_XML_SECURE) - xml += """0 - """ - self._snapshotCreateXML(xml, 0) - finally: - if changed: - self.recover_snapshot_xml() + # libvirt decides whether this firmware/NVRAM supports internal + # snapshots; the domain definition is never rewritten for it (R-13). + xml += self._XMLDesc(VIR_DOMAIN_XML_SECURE) + xml += """0 + """ + self._snapshotCreateXML(xml, 0) def change_snapshot_xml(self): try: @@ -1508,13 +1505,8 @@ class wvmInstance(wvmConnect): snap.delete(0) def snapshot_revert(self, snapshot): - changed = self.change_snapshot_xml() - try: - snap = self.instance.snapshotLookupByName(snapshot, 0) - self.instance.revertToSnapshot(snap, 0) - finally: - if changed: - self.recover_snapshot_xml() + snap = self.instance.snapshotLookupByName(snapshot, 0) + self.instance.revertToSnapshot(snap, 0) def get_managed_save_image(self): return self.instance.hasManagedSaveImage(0) diff --git a/vrtManager/test_xml_safety.py b/vrtManager/test_xml_safety.py index df0bab7d..d7d61bc3 100644 --- a/vrtManager/test_xml_safety.py +++ b/vrtManager/test_xml_safety.py @@ -68,8 +68,8 @@ class TestXmlSafety(unittest.TestCase): with self.assertRaises(RuntimeError): inst.create_snapshot(bad_name, bad_desc) - inst.change_snapshot_xml.assert_called_once() - inst.recover_snapshot_xml.assert_called_once() + inst.change_snapshot_xml.assert_not_called() + inst.recover_snapshot_xml.assert_not_called() def test_snapshot_xml_payload_is_escaped(self): inst = wvmInstance.__new__(wvmInstance) @@ -138,12 +138,8 @@ class TestXmlSafety(unittest.TestCase): self.assertFalse(inst.change_snapshot_xml()) inst._defineXML.assert_not_called() - # Create snapshot should not invoke recover_snapshot_xml if not changed - inst.change_snapshot_xml = MagicMock(return_value=False) - inst.recover_snapshot_xml = MagicMock() inst.create_snapshot("snap_rom", "testing rom preservation") - inst.change_snapshot_xml.assert_called_once() - inst.recover_snapshot_xml.assert_not_called() + inst._defineXML.assert_not_called() def test_malformed_xml_raises_in_change(self): inst = wvmInstance.__new__(wvmInstance) @@ -165,15 +161,28 @@ class TestXmlSafety(unittest.TestCase): mock_snap = MagicMock() inst.instance.snapshotLookupByName = MagicMock(return_value=mock_snap) inst.instance.revertToSnapshot = MagicMock() - inst.change_snapshot_xml = MagicMock(return_value=False) - inst.recover_snapshot_xml = MagicMock() + inst._defineXML = MagicMock() inst.snapshot_revert("snap_rom") - inst.change_snapshot_xml.assert_called_once() inst.instance.snapshotLookupByName.assert_called_once_with("snap_rom", 0) inst.instance.revertToSnapshot.assert_called_once_with(mock_snap, 0) - inst.recover_snapshot_xml.assert_not_called() + inst._defineXML.assert_not_called() + + def test_internal_snapshot_never_rewrites_a_pflash_loader(self): + # R-13: libvirt decides; the definition is not switched to rom. + inst = wvmInstance.__new__(wvmInstance) + inst.get_status = MagicMock(return_value=5) + inst._XMLDesc = MagicMock( + return_value="/usr/share/OVMF/OVMF.fd" + ) + inst._defineXML = MagicMock() + inst._snapshotCreateXML = MagicMock() + + inst.create_snapshot("snap", "uefi") + + inst._defineXML.assert_not_called() + self.assertIn("type='pflash'", inst._snapshotCreateXML.call_args[0][0]) if __name__ == "__main__":