mirror of
https://github.com/retspen/webvirtcloud
synced 2026-11-08 11:16:47 +00:00
102 lines
5 KiB
Python
102 lines
5 KiB
Python
from contextlib import contextmanager
|
|
from unittest.mock import patch
|
|
|
|
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 instances.models import Instance
|
|
from libvirt import VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, VIR_DOMAIN_UNDEFINE_NVRAM, libvirtError
|
|
from vrtManager.util import OperationError
|
|
|
|
|
|
class DestroyOrderTestCase(TestCase):
|
|
def setUp(self):
|
|
admin = get_user_model().objects.create_superuser("destroy_admin", "d@example.com", "pw")
|
|
self.client.force_login(admin)
|
|
self.client.raise_request_exception = False
|
|
compute = Compute.objects.create(name="destroy-compute", hostname="127.0.0.1", login="", password="", type=4)
|
|
self.instance = Instance.objects.create(
|
|
compute=compute, name="destroy-vm", uuid="11111111-2222-3333-4444-555555555555"
|
|
)
|
|
|
|
def destroy(self, undefine_error=None, data=None):
|
|
with patch("instances.models.wvmInstance") as wvm:
|
|
proxy = wvm.return_value
|
|
proxy.instance.isActive.return_value = False
|
|
proxy.split_disk_paths_by_use.return_value = (["/pool/own.qcow2"], ["/pool/shared.qcow2"])
|
|
proxy.delete.side_effect = undefine_error
|
|
self.client.post(reverse("instances:destroy", args=[self.instance.id]), data or {"delete_disk": "1"})
|
|
return proxy
|
|
|
|
def test_the_nvram_goes_only_when_asked(self):
|
|
for data, wanted, unwanted in (
|
|
({"delete_nvram": "1"}, VIR_DOMAIN_UNDEFINE_NVRAM, VIR_DOMAIN_UNDEFINE_KEEP_NVRAM),
|
|
({}, VIR_DOMAIN_UNDEFINE_KEEP_NVRAM, VIR_DOMAIN_UNDEFINE_NVRAM),
|
|
):
|
|
with self.subTest(data=data):
|
|
Instance.objects.get_or_create(
|
|
compute=self.instance.compute, name="destroy-vm", uuid="11111111-2222-3333-4444-555555555555"
|
|
)
|
|
self.instance = Instance.objects.get(uuid="11111111-2222-3333-4444-555555555555")
|
|
flags = self.destroy(data=data).delete.call_args[0][0]
|
|
self.assertTrue(flags & wanted)
|
|
self.assertFalse(flags & unwanted)
|
|
|
|
def test_disks_are_deleted_after_the_undefine(self):
|
|
proxy = self.destroy()
|
|
proxy.get_volume_by_path.assert_called_once_with("/pool/own.qcow2")
|
|
self.assertFalse(Instance.objects.filter(pk=self.instance.pk).exists())
|
|
|
|
def test_failed_undefine_deletes_no_disk(self):
|
|
proxy = self.destroy(undefine_error=libvirtError("refused"))
|
|
proxy.get_volume_by_path.assert_not_called()
|
|
self.assertTrue(Instance.objects.filter(pk=self.instance.pk).exists())
|
|
|
|
def test_destroy_and_delete_vol_hold_the_compute_exclusively(self):
|
|
calls = []
|
|
|
|
@contextmanager
|
|
def record(compute, *, shared=False):
|
|
calls.append(shared)
|
|
yield
|
|
|
|
with patch("instances.views.libvirt_compute_lock", record):
|
|
with patch("instances.models.wvmInstance"):
|
|
self.client.post(reverse("instances:delete_vol", args=[self.instance.id]), {"dev": "vda"})
|
|
self.destroy()
|
|
self.assertEqual(calls, [False, False])
|
|
|
|
def test_a_volume_the_vms_snapshots_need_is_kept_after_the_detach(self):
|
|
disk = {"dev": "vdb", "path": "/pool/data.qcow2", "storage": "default", "image": "data.qcow2"}
|
|
with patch("instances.models.wvmInstance") as wvm, patch("instances.views.wvmStorage") as storage:
|
|
proxy = wvm.return_value
|
|
proxy.get_disk_devices.return_value = [disk]
|
|
proxy.paths_used_by_other_domains.return_value = set()
|
|
proxy.wait_disk_detached.return_value = True
|
|
storage.return_value.del_volume.side_effect = OperationError("Volume data.qcow2 is in use")
|
|
response = self.client.post(reverse("instances:delete_vol", args=[self.instance.id]), {"dev": "vdb"})
|
|
self.assertEqual(response.status_code, 302)
|
|
proxy.detach_disk.assert_called_once_with("vdb")
|
|
messages = [str(m) for m in get_messages(response.wsgi_request)]
|
|
self.assertTrue(any("data.qcow2 is in use" in m and "detached" in m for m in messages), messages)
|
|
|
|
def test_confirmation_page_takes_no_lock(self):
|
|
calls = []
|
|
|
|
@contextmanager
|
|
def record(compute, *, shared=False):
|
|
calls.append(shared)
|
|
yield
|
|
|
|
with patch("instances.views.libvirt_compute_lock", record), patch("instances.models.wvmInstance"):
|
|
response = self.client.get(reverse("instances:destroy", args=[self.instance.id]))
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(calls, [])
|
|
|
|
def test_the_confirmation_page_of_a_paused_vm_offers_the_form(self):
|
|
with patch("instances.models.wvmInstance") as wvm:
|
|
wvm.return_value.get_status.return_value = 3 # paused
|
|
response = self.client.get(reverse("instances:destroy", args=[self.instance.id]))
|
|
self.assertIn('id="delete_form"', response.content.decode())
|