1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
webvirtcloud/instances/test_create_instance.py
catborise 03ac93ea53 fix(instances): CD-ROM bus, USB controller and MACs of a new VM
Creating each option on a RHEL qemu-kvm host showed:

- CD-ROM: the create page offered every disk bus, but the empty CD-ROM
  of a new VM works only on ide, sata and scsi. fdc and virtio failed
  ("Invalid bus type", "does not support ejectable media"), usb failed
  with a device named 'Nonea' because vol_dev_type() had no prefix for
  usb, and after that libvirt refuses an empty usb disk. Only ide, sata
  and scsi are offered now, and another value is refused before any
  storage is allocated. vol_dev_type() names usb disks sdX, which also
  fixes data volumes on the usb bus.
- USB controller: any input device added a bare <controller
  type='usb'/>, which libvirt turns into piix3-uhci (USB 1.1) on q35
  instead of its default qemu-xhci. The bare controller is gone; libvirt
  adds the machine's default for USB inputs.
- MACs: with more networks than MAC addresses, create_instance indexed
  past the list and the request failed with a 500. NICs without a given
  MAC now get one from libvirt. MAC addresses are validated (format, not
  more than networks) by the page and the API before anything is
  created; they went into the XML unchecked.

Unit tests cover the refused values, the MAC rule and the usb prefix.
Live tests create a VM with every offered CD-ROM bus and start it,
check the USB controller with a USB input device and create a VM with
two networks and one MAC.
2026-10-04 18:33:28 +03:00

146 lines
6.9 KiB
Python

"""Regression tests for the instance creation request."""
from unittest.mock import patch
from computes.models import Compute
from django.contrib.auth import get_user_model
from django.test import TestCase
from django.urls import reverse
from instances.api.serializers import CreateInstanceSerializer
from instances.models import Instance
from instances.utils import nic_macs
class CreateInstanceViewTests(TestCase):
@patch("instances.views.wvmCreate")
@patch("instances.views.NewVMForm")
def test_successful_create_redirects_after_saving_instance(self, form_class, connection_class):
response, conn = self.create(form_class, connection_class)
instance = Instance.objects.get(name="new-vm")
self.assertRedirects(
response, reverse("instances:instance", args=[instance.id]),
fetch_redirect_response=False,
)
conn.create_instance.assert_called_once()
@patch("instances.views.wvmCreate")
@patch("instances.views.NewVMForm")
def test_secure_boot_loader_on_q35_requests_secure_boot(self, form_class, connection_class):
self.create(form_class, connection_class, firmware="UEFI x86_64: /usr/share/edk2/ovmf/OVMF_CODE.secboot.fd")
firmware = connection_class.return_value.create_instance.call_args.kwargs["firmware"]
self.assertEqual(firmware["secure"], "yes")
@patch("instances.views.wvmCreate")
@patch("instances.views.NewVMForm")
def test_invalid_devices_are_refused_before_anything_is_created(self, form_class, connection_class):
for data in (
{"add_cdrom": "virtio"}, # no ejectable media
{"add_cdrom": "fdc"},
{"add_cdrom": "usb"}, # libvirt refuses an empty usb disk
{"mac": "52:54:00:00:00:01,52:54:00:00:00:02"}, # one network
{"mac": "52:54:00:00:00:01'/><evil/>"},
):
with self.subTest(data):
Instance.objects.all().delete()
response, conn = self.create(form_class, connection_class, **data)
self.assertEqual(response.status_code, 200)
self.assertFalse(Instance.objects.exists())
conn.create_volume.assert_not_called()
conn.create_instance.assert_not_called()
connection_class.reset_mock()
def create(self, form_class, connection_class, **data):
user = get_user_model().objects.filter(username="create-test-admin").first() or (
get_user_model().objects.create_superuser(
username="create-test-admin", password="password", email="create@example.com"
)
)
self.client.force_login(user)
compute, _ = Compute.objects.get_or_create(
name="create-test-compute", hostname="localhost", type=4,
login="", password="",
)
conn = connection_class.return_value
conn.get_instances.return_value = []
conn.get_video_models.return_value = ["vga"]
conn.get_cache_modes.return_value = {"default": "Default"}
conn.get_disk_device_types.return_value = ["disk"]
conn.get_disk_bus_types.return_value = ["fdc", "virtio", "sata"]
conn.get_networks.return_value = ["default"]
conn.get_nwfilters.return_value = []
conn.get_network_models.return_value = ["virtio"]
conn.get_storages.return_value = ["default"]
conn.get_dom_capabilities.return_value = {"loader_enums": [], "loaders": []}
conn.get_capabilities.return_value = {}
conn.label_for_firmware_path.return_value = None
conn.find_uefi_path_for_arch.return_value = None
conn.get_volume_path.return_value = "/var/lib/libvirt/images/base.qcow2"
conn.get_volume_format_type.return_value = "qcow2"
form = form_class.return_value
form.is_valid.return_value = True
form.cleaned_data = {
"name": "new-vm", "meta_prealloc": False, "hdd_size": 0,
"template": "", "images": "base.qcow2", "cache_mode": "default",
"firmware": "BIOS", "net_model": "virtio", "memory": 1024,
"vcpu": 1, "vcpu_mode": "host-model", "networks": "default",
"virtio": True, "listener_addr": "0.0.0.0", "nwfilter": "",
"graphics": "vnc", "video": "vga", "console_pass": "",
"mac": "52:54:00:00:00:01", "qemu_ga": False,
"add_cdrom": "sata", "add_input": "default",
**data,
}
response = self.client.post(
reverse("instances:create_instance", args=[compute.id, "x86_64", "q35"]),
{"create": "1"},
)
return response, conn
@patch("instances.views.wvmCreate")
def test_firmware_list_uses_the_chosen_machine_type(self, connection_class):
user = get_user_model().objects.create_superuser(
username="create-fw-admin", password="password", email="fw@example.com"
)
self.client.force_login(user)
compute = Compute.objects.create(
name="create-fw-compute", hostname="localhost", type=4, login="", password="",
)
conn = connection_class.return_value
conn.get_instances.return_value = []
conn.get_storages.return_value = []
conn.get_networks.return_value = []
conn.get_nwfilters.return_value = []
conn.get_cache_modes.return_value = {}
# A host without loader support has no "loaders"/"loader_enums" keys
conn.get_dom_capabilities.return_value = {"loader_support": "no"}
conn.get_capabilities.return_value = {}
conn.label_for_firmware_path.return_value = None
conn.find_uefi_path_for_arch.return_value = None
response = self.client.get(reverse("instances:create_instance", args=[compute.id, "x86_64", "q35"]))
self.assertEqual(response.status_code, 200)
conn.find_uefi_path_for_arch.assert_called_once_with("x86_64", "q35")
class NicMacsTestCase(TestCase):
def test_macs_follow_the_networks(self):
self.assertEqual(nic_macs("52:54:00:00:00:01", "a,b"), ["52:54:00:00:00:01"])
self.assertEqual(nic_macs("", "a,b"), [])
def test_invalid_or_extra_macs_are_refused(self):
for mac, networks in (("52:54:00:00:00:01,52:54:00:00:00:02", "a"), ("not-a-mac", "a")):
with self.subTest(mac=mac), self.assertRaises(ValueError):
nic_macs(mac, networks)
def test_api_serializer_refuses_extra_macs(self):
serializer = CreateInstanceSerializer(data={
"name": "api-vm", "vcpu": 1, "vcpu_mode": "host-model", "memory": 512,
"networks": "default", "mac": "52:54:00:00:00:01,52:54:00:00:00:02", "nwfilter": "",
"storage": "default", "hdd_size": 1, "cache_mode": "none", "meta_prealloc": False,
"virtio": True, "qemu_ga": True, "console_pass": "", "video": "vga", "listener_addr": "0.0.0.0",
})
self.assertFalse(serializer.is_valid())
self.assertIn("mac", serializer.errors)