j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-11-08 19:26:48 +00:00
Code Issues Projects Releases Wiki Activity
webvirtcloud/instances/tests.py
catborise c2fea2aa39 feat(accounts): failed logins lock a user for the address they come from 
Nothing limited failed logins: a password could be guessed at the speed
the server answers. django-axes (8.3.1, Django 5.2, Python 3.10-3.13)
now counts them.

- A user is locked for the address the failed logins come from
  (username and IP together), so an attacker cannot lock someone out
  from elsewhere and users behind one NAT address do not share a lock.
  A successful login resets the count. axes sits in front of every
  authentication: the login page with or without OTP, LDAP and the API.
- The number of failed logins (default 5) and the lock time in minutes
  (default 15) are on the Settings page (LOGIN_FAILURE_LIMIT,
  LOGIN_LOCKOUT_MINUTES, added by migration appsettings.0014) and apply
  at once; an invalid value falls back to the default. Unlock with
  "manage.py axes_reset_username <username>".
- Behind the bundled nginx every request comes from 127.0.0.1; the
  client address is read from X-Real-IP, which nginx sets, but only for
  requests from a loopback address or from LOGIN_TRUSTED_PROXIES, so a
  forged header does not escape a lock.

settings.py.template gets the app, middleware, backend and AXES_*
settings; README tells existing installations what to copy. Tests that
call client.login now pass a request, which the axes backend needs.
The new tests fail with axes disabled.
2026-10-06 11:05:03 +03:00

1172 lines
40 KiB
Python
Raw Blame History

import os
import re
import unittest
from accounts.models import UserAttributes, UserInstance, UserSSHKey
from appsettings.models import AppSettings
from computes.models import Compute
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Permission
from django.http.response import Http404
from django.shortcuts import reverse
from django.test import RequestFactory, TestCase
from vrtManager.create import wvmCreate
from vrtManager.util import randomUUID
from . import livetest
from .models import Flavor, Instance
from .utils import refr
TEST_COMPUTE_HOST = os.environ.get("TEST_LIBVIRT_HOST", "localhost")
TEST_COMPUTE_LOGIN = os.environ.get("TEST_LIBVIRT_LOGIN", "")
TEST_COMPUTE_PASSWORD = os.environ.get("TEST_LIBVIRT_PASSWORD", "")
TEST_COMPUTE_TYPE = int(os.environ.get("TEST_LIBVIRT_TYPE", 4))
class InstancesTestCase(TestCase):
@classmethod
def setUpClass(cls):
if not livetest.enabled():
raise unittest.SkipTest("Set TEST_LIBVIRT_HOST to run the live libvirt tests")
super().setUpClass()
# Add users for testing purposes
User = get_user_model()
cls.admin_user = User.objects.filter(is_superuser=True).first() or User.objects.get(username="admin")
cls.test_user = User.objects.create(username="test-user")
UserAttributes.objects.create(
user=cls.test_user,
max_instances=1,
max_cpus=1,
max_memory=128,
max_disk_size=1,
)
permission = Permission.objects.get(codename="clone_instances")
cls.test_user.user_permissions.add(permission)
# Add compute
cls.compute = Compute.objects.create(
name="instances-test-compute",
hostname=TEST_COMPUTE_HOST,
login=TEST_COMPUTE_LOGIN,
password=TEST_COMPUTE_PASSWORD,
details="test",
type=TEST_COMPUTE_TYPE,
)
try:
cls.connection = wvmCreate(
cls.compute.hostname,
cls.compute.login,
cls.compute.password,
cls.compute.type,
)
# Remove leftovers of an interrupted run, then record what
# else is on the host so tearDownClass can prove it is untouched.
livetest.cleanup(cls.connection.wvm)
livetest.ensure_pool(cls.connection.wvm)
cls.inventory = livetest.inventory(cls.connection.wvm)
# Add disks for testing
cls.connection.create_volume(
livetest.POOL,
"wvc-test-volume",
1,
"qcow2",
False,
0,
0,
)
# XML for testing vm
with open("conf/test-vm.xml", "r") as f:
cls.xml = f.read()
# Create testing vm from XML
cls.connection._defineXML(cls.xml)
refr(cls.compute)
# The compute may hold other VMs; never pick one of them.
cls.instance: Instance = Instance.objects.get(compute=cls.compute, name="wvc-test-vm")
except Exception as e:
cls.instance = None
if getattr(cls, "connection", None):
try:
livetest.remove_pool(cls.connection.wvm)
except Exception:
pass
if getattr(cls, "compute", None):
try:
cls.compute.delete()
except Exception:
pass
try:
super().tearDownClass()
except Exception:
pass
raise unittest.SkipTest(f"Live libvirt host not available for instances tests: {e}")
@classmethod
def tearDownClass(cls):
conn = cls.connection.wvm
try:
livetest.remove_pool(conn)
finally:
cls.compute.delete()
super().tearDownClass()
domains, volumes = livetest.inventory(conn)
if (domains, volumes) != cls.inventory:
raise AssertionError(
"Live tests changed objects outside the test namespace: "
f"domains {cls.inventory[0] ^ domains}, volumes {cls.inventory[1] ^ volumes}"
)
def setUp(self):
self.client.login(request=RequestFactory().get("/"), username="admin", password="admin")
self.rsa_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw== test@test"
def test_index(self):
response = self.client.get(reverse("instances:index"))
self.assertEqual(response.status_code, 200)
self.client.force_login(self.test_user)
response = self.client.get(reverse("instances:index"))
self.assertEqual(response.status_code, 200)
def test_create_select_type(self):
response = self.client.get(
reverse("instances:create_instance_select_type", args=[self.compute.id])
)
self.assertEqual(response.status_code, 200)
def test_instance_page(self):
response = self.client.get(
reverse("instances:instance", args=[self.instance.id])
)
self.assertEqual(response.status_code, 200)
self.client.force_login(self.test_user)
response = self.client.get(
reverse("instances:instance", args=[self.instance.id])
)
self.assertRaises(Http404)
# def test_create_volume(self):
# response = self.client.post(
# reverse('create_volume', args=[self.compute.id, 'default']),
# {
# 'name': 'test',
# 'format': 'qcow2',
# 'size': '1',
# 'meta_prealloc': False,
# },
# )
# self.assertRedirects(response, reverse('storage', args=[self.compute.id, 'default']))
def test_create_destroy_instance(self):
# Create
response = self.client.get(
reverse(
"instances:create_instance", args=[self.compute.id, "x86_64", "q35"]
)
)
self.assertEqual(response.status_code, 200)
response = self.client.post(
reverse(
"instances:create_instance", args=[self.compute.id, "x86_64", "q35"]
),
{
"name": "wvc-test-new",
"firmware": "BIOS",
"vcpu": 1,
"vcpu_mode": "host-model",
"memory": 128,
"device0": "disk",
"bus0": "virtio",
"images": "wvc-test-volume.qcow2",
"storage-control": livetest.POOL,
"image-control": "wvc-test.qcow2",
"networks": "default",
"network-control": "default",
"cache_mode": "directsync",
"nwfilter": "",
"graphics": "vnc",
"video": "vga",
"listener_addr": "0.0.0.0",
"console_pass": "",
"qemu_ga": False,
"virtio": True,
"add_cdrom": "None",
"add_input": "None",
"create": True,
},
)
self.assertEqual(response.status_code, 302)
instance_qs: Instance = Instance.objects.filter(name="wvc-test-new")
self.assertEqual(len(instance_qs), 1)
instance = instance_qs[0]
# Destroy
response = self.client.get(reverse("instances:destroy", args=[instance.id]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
reverse("instances:destroy", args=[instance.id]),
{}, # do not destroy disk image
HTTP_REFERER=reverse("index"),
)
self.assertRedirects(response, reverse("instances:index"))
def test_create_from_xml(self):
uuid = randomUUID()
xml = self.xml.replace("wvc-test-vm", "wvc-test-vm-xml")
xml = re.sub(r"\s?<uuid>.*?</uuid>", f"<uuid>{uuid}</uuid>", xml)
response = self.client.post(
reverse("instances:create_instance_select_type", args=[self.compute.id]),
{
"create_xml": True,
"dom_xml": xml,
},
)
self.assertEqual(response.status_code, 302)
xml_instance_qs: Instance = Instance.objects.filter(name="wvc-test-vm-xml")
self.assertEqual(len(xml_instance_qs), 1)
xml_instance = xml_instance_qs[0]
# destroy started instance to maximize coverage
xml_instance.proxy.start()
response = self.client.post(
reverse("instances:destroy", args=[xml_instance.id]),
{}, # do not delete disk image
HTTP_REFERER=reverse("index"),
)
self.assertRedirects(response, reverse("instances:index"))
def test_resize_cpu(self):
self.assertEqual(self.instance.vcpu, 1)
self.assertEqual(self.instance.cur_vcpu, 1)
response = self.client.post(
reverse("instances:resizevm_cpu", args=[self.instance.id]),
{
"vcpu": 4,
"cur_vcpu": 2,
},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
# reset cached properties
del self.instance.vcpu
del self.instance.cur_vcpu
self.assertEqual(self.instance.vcpu, 4)
self.assertEqual(self.instance.cur_vcpu, 2)
def test_resize_cpu_with_quota(self):
# test for non admin user with quotas
vcpu = self.instance.vcpu
cur_vcpu = self.instance.cur_vcpu
UserInstance.objects.create(
user=self.test_user, instance=self.instance, is_change=True
)
self.client.force_login(self.test_user)
response = self.client.post(
reverse("instances:resizevm_cpu", args=[self.instance.id]),
{
"vcpu": 4,
"cur_vcpu": 2,
},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
del self.instance.vcpu
del self.instance.cur_vcpu
# no changes as user reached quota
self.assertEqual(self.instance.vcpu, vcpu)
self.assertEqual(self.instance.cur_vcpu, cur_vcpu)
def test_resize_memory(self):
self.assertEqual(self.instance.memory, 128)
self.assertEqual(self.instance.cur_memory, 128)
response = self.client.post(
reverse("instances:resize_memory", args=[self.instance.id]),
{"memory": 512, "cur_memory": 256},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
del self.instance.memory
del self.instance.cur_memory
self.assertEqual(self.instance.memory, 512)
self.assertEqual(self.instance.cur_memory, 256)
response = self.client.post(
reverse("instances:resize_memory", args=[self.instance.id]),
{"memory_custom": 500, "cur_memory_custom": 200},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
del self.instance.memory
del self.instance.cur_memory
self.assertEqual(self.instance.memory, 500)
self.assertEqual(self.instance.cur_memory, 200)
def test_resize_memory_with_quota(self):
# test for non admin user with quotas
memory = self.instance.memory
cur_memory = self.instance.cur_memory
UserInstance.objects.create(
user=self.test_user, instance=self.instance, is_change=True
)
self.client.force_login(self.test_user)
response = self.client.post(
reverse("instances:resize_memory", args=[self.instance.id]),
{"memory": 512, "cur_memory": 256},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
del self.instance.memory
del self.instance.cur_memory
# no changes as user reached quota
self.assertEqual(self.instance.memory, memory)
self.assertEqual(self.instance.cur_memory, cur_memory)
def test_resize_disk(self):
self.assertEqual(self.instance.disks[0]["size"], 1024**3)
response = self.client.post(
reverse("instances:resize_disk", args=[self.instance.id]),
{
"disk_size_vda": 2,
},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
del self.instance.disks
self.assertEqual(self.instance.disks[0]["size"], 2 * 1024**3)
def test_resize_disk_with_quota(self):
# test for non admin user with quotas
disk_size = self.instance.disks[0]["size"]
UserInstance.objects.create(
user=self.test_user, instance=self.instance, is_change=True
)
self.client.force_login(self.test_user)
response = self.client.post(
reverse("instances:resize_disk", args=[self.instance.id]),
{
"disk_size_vda": 3,
},
)
self.assertRedirects(
response, reverse("instances:instance", args=[self.instance.id]) + "#resize"
)
# no changes as user reached quota
del self.instance.disks
self.assertEqual(self.instance.disks[0]["size"], disk_size)
def test_add_delete_new_volume(self):
self.assertEqual(len(self.instance.disks), 1)
response = self.client.post(
reverse("instances:add_new_vol", args=[self.instance.id]),
{
"storage": livetest.POOL,
"name": "wvc-test-volume-2",
"size": 1,
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.disks
self.assertEqual(len(self.instance.disks), 2)
response = self.client.post(
reverse("instances:delete_vol", args=[self.instance.id]),
{
"storage": livetest.POOL,
"dev": "vdb",
"name": "wvc-test-volume-2.qcow2",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.disks
self.assertEqual(len(self.instance.disks), 1)
def test_detach_attach_volume(self):
# detach volume
response = self.client.post(
reverse("instances:detach_vol", args=[self.instance.id]),
{
"dev": "vda",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.disks
self.assertEqual(len(self.instance.disks), 0)
# reattach volume
response = self.client.post(
reverse("instances:add_existing_vol", args=[self.instance.id]),
{
"selected_storage": livetest.POOL,
"vols": "wvc-test-volume.qcow2",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.disks
self.assertEqual(len(self.instance.disks), 1)
def test_edit_volume(self):
response = self.client.post(
reverse("instances:edit_volume", args=[self.instance.id]),
{
"vol_path": f"{livetest.POOL_PATH}/wvc-test-volume.qcow2",
# 'vol_shareable': False,
# 'vol_readonly': False,
"vol_bus": "virtio",
"vol_bus_old": "virtio",
"vol_format": "qcow2",
"dev": "vda",
"edit_volume": True,
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
def test_attach_detach_cdrom(self):
self.assertEqual(len(self.instance.media), 1)
response = self.client.post(
reverse("instances:add_cdrom", args=[self.instance.id]),
{
"bus": "sata",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.media
self.assertEqual(len(self.instance.media), 2)
# create dummy iso
# with tempfile.NamedTemporaryFile() as f:
# f.write(b'\x00' * 1024**2)
# response = self.client.post(
# reverse('storage', args=[instance.compute.id, 'default']),
# {
# 'iso_upload': True,
# 'file': f
# },
# )
# detach CD-ROM drive
response = self.client.post(
reverse("instances:detach_cdrom", args=[self.instance.id, "sda"]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.media
self.assertEqual(len(self.instance.media), 1)
def test_snapshots(self):
self.assertEqual(len(self.instance.snapshots), 0)
response = self.client.post(
reverse("instances:snapshot", args=[self.instance.id]),
{
"name": "test-snapshot",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.snapshots
self.assertEqual(len(self.instance.snapshots), 1)
response = self.client.post(
reverse("instances:revert_snapshot", args=[self.instance.id]),
{
"name": "test-snapshot",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
response = self.client.post(
reverse("instances:delete_snapshot", args=[self.instance.id]),
{
"name": "test-snapshot",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.snapshots
self.assertEqual(len(self.instance.snapshots), 0)
def test_autostart(self):
self.assertEqual(self.instance.autostart, 0)
response = self.client.post(
reverse("instances:set_autostart", args=[self.instance.id]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.autostart
self.assertEqual(self.instance.autostart, 1)
response = self.client.post(
reverse("instances:unset_autostart", args=[self.instance.id]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.autostart
self.assertEqual(self.instance.autostart, 0)
def test_bootmenu(self):
self.assertEqual(self.instance.bootmenu, True)
response = self.client.post(
reverse("instances:unset_bootmenu", args=[self.instance.id]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.bootmenu
self.assertEqual(self.instance.bootmenu, False)
response = self.client.post(
reverse("instances:set_bootmenu", args=[self.instance.id]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.bootmenu
self.assertEqual(self.instance.bootmenu, True)
def test_guest_agent(self):
self.assertEqual(self.instance.guest_agent, False)
response = self.client.post(
reverse("instances:set_guest_agent", args=[self.instance.id]),
{"guest_agent": True},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.guest_agent
self.assertEqual(self.instance.guest_agent, True)
response = self.client.post(
reverse("instances:set_guest_agent", args=[self.instance.id]),
{"guest_agent": False},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.guest_agent
self.assertEqual(self.instance.guest_agent, False)
def test_video_model(self):
self.assertEqual(self.instance.video_model, "vga")
response = self.client.post(
reverse("instances:set_video_model", args=[self.instance.id]),
{"video_model": "virtio"},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.video_model
self.assertEqual(self.instance.video_model, "virtio")
def test_owner(self):
self.assertEqual(UserInstance.objects.count(), 0)
response = self.client.post(
reverse("instances:add_owner", args=[self.instance.id]),
{"user_id": self.admin_user.id},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(UserInstance.objects.count(), 1)
user_instance: UserInstance = UserInstance.objects.get(id=1)
self.assertEqual(user_instance.instance_id, self.instance.id)
self.assertEqual(user_instance.user_id, self.admin_user.id)
# test when no multiple owners allowed
setting = AppSettings.objects.get(key="ALLOW_INSTANCE_MULTIPLE_OWNER")
setting.value = "False"
setting.save()
response = self.client.post(
reverse("instances:add_owner", args=[self.instance.id]),
{"user_id": self.test_user.id},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(UserInstance.objects.count(), 1)
response = self.client.post(
reverse("instances:del_owner", args=[self.instance.id]),
{"userinstance": user_instance.id},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(UserInstance.objects.count(), 0)
def test_clone(self):
instance_count = Instance.objects.count()
response = self.client.post(
reverse("instances:clone", args=[self.instance.id]),
{
"name": "wvc-test-vm-clone",
"clone-net-mac-0": "de:ad:be:ef:de:ad",
"disk-vda": "wvc-test-clone.img",
"clone-title": "",
"clone-description": "",
"clone": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(Instance.objects.count(), instance_count + 1)
clone_qs = Instance.objects.filter(name="wvc-test-vm-clone")
self.assertEqual(len(clone_qs), 1)
clone = clone_qs[0]
self.assertEqual(clone.proxy.get_net_devices()[0]["mac"], "de:ad:be:ef:de:ad")
response = self.client.post(
reverse("instances:snapshot", args=[clone.id]),
{
"name": "test",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
response = self.client.post(
reverse("instances:destroy", args=[clone.id]),
{"delete_disk": True, "delete_nvram": True},
HTTP_REFERER=reverse("index"),
)
self.assertRedirects(response, reverse("instances:index"))
self.assertEqual(Instance.objects.count(), instance_count)
def test_clone_with_quota(self):
# test for non admin user with quotas
instance_count = Instance.objects.count()
# Cloning needs change permission on the source VM
UserInstance.objects.create(user=self.test_user, instance=self.instance, is_change=True)
self.client.force_login(self.test_user)
response = self.client.post(
reverse("instances:clone", args=[self.instance.id]),
{
"name": "wvc-test-vm-clone",
"clone-net-mac-0": "de:ad:be:ef:de:ad",
"disk-vda": "wvc-test-clone.img",
"clone-title": "",
"clone-description": "",
"clone": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
# no new instances created as user reached quota
self.assertEqual(Instance.objects.count(), instance_count)
def test_clone_errors(self):
instance_count = Instance.objects.count()
# duplicate name
response = self.client.post(
reverse("instances:clone", args=[self.instance.id]),
{
"name": "wvc-test-vm",
"clone-net-mac-0": "de:ad:be:ef:de:ad",
"disk-vda": "wvc-test.img",
"clone-title": "",
"clone-description": "",
"clone": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(Instance.objects.count(), instance_count)
# wrong name
response = self.client.post(
reverse("instances:clone", args=[self.instance.id]),
{
"name": "!@#$",
"clone-net-mac-0": "de:ad:be:ef:de:ad",
"disk-vda": "!@#$.img",
"clone-title": "",
"clone-description": "",
"clone": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(Instance.objects.count(), instance_count)
# wrong mac
response = self.client.post(
reverse("instances:clone", args=[self.instance.id]),
{
"name": "wvc-test-vm-clone",
"clone-net-mac-0": "gh:ad:be:ef:de:ad",
"disk-vda": "wvc-test-clone.img",
"clone-title": "",
"clone-description": "",
"clone": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
self.assertEqual(Instance.objects.count(), instance_count)
def test_console(self):
response = self.client.post(
reverse("instances:update_console", args=[self.instance.id]),
{"listen_on": "0.0.0.0", "password": "", "keymap": "auto"},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
def test_status(self):
response = self.client.get(reverse("instances:status", args=[self.instance.id]))
self.assertEqual(response.status_code, 200)
def test_stats(self):
response = self.client.get(reverse("instances:stats", args=[self.instance.id]))
self.assertEqual(response.status_code, 200)
def test_guess_mac_address(self):
response = self.client.get(
reverse("instances:guess_mac_address", args=[self.instance.name])
)
self.assertEqual(response.status_code, 200)
def test_random_mac_address(self):
response = self.client.get(reverse("instances:random_mac_address"))
self.assertEqual(response.status_code, 200)
def test_guess_clone_name(self):
response = self.client.get(reverse("instances:guess_clone_name"))
self.assertEqual(response.status_code, 200)
def test_sshkeys(self):
UserSSHKey.objects.create(
keyname="keyname", keypublic=self.rsa_key, user=self.test_user
)
UserInstance.objects.create(user=self.test_user, instance=self.instance)
response = self.client.get(
reverse("instances:sshkeys", args=[self.instance.id])
)
self.assertEqual(response.status_code, 200)
response = self.client.get(
reverse("instances:sshkeys", args=[self.instance.id]) + "?plain=true"
)
self.assertEqual(response.status_code, 200)
def test_start_template(self):
# starting templates must fail
self.assertEqual(self.instance.status, 5)
self.instance.is_template = True
self.instance.save()
response = self.client.post(
reverse("instances:poweron", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 5)
self.instance.is_template = False
self.instance.save()
def test_power(self):
# poweron
self.assertEqual(self.instance.status, 5)
response = self.client.post(
reverse("instances:poweron", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 1)
# suspend
response = self.client.post(
reverse("instances:suspend", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 3)
# resume
response = self.client.post(
reverse("instances:resume", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 1)
# poweroff
response = self.client.post(
reverse("instances:poweroff", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
# as no OS is installed ACPI won't work
del self.instance.status
self.assertEqual(self.instance.status, 1)
# powercycle
response = self.client.post(
reverse("instances:powercycle", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 1)
# force_off
response = self.client.post(
reverse("instances:force_off", args=[self.instance.id]),
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.status
self.assertEqual(self.instance.status, 5)
def test_vv_file(self):
response = self.client.get(
reverse("instances:getvvfile", args=[self.instance.id])
)
self.assertEqual(response.status_code, 200)
def test_vcpu_hotplug(self):
response = self.client.post(
reverse("instances:set_vcpu_hotplug", args=[self.instance.id]),
{"vcpu_hotplug": "True"},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
def test_change_network(self):
self.assertEqual(self.instance.networks[0]["mac"], "52:54:00:a2:3c:e7")
response = self.client.post(
reverse("instances:change_network", args=[self.instance.id]),
{
"net-old-mac-0": "52:54:00:a2:3c:e7",
"net-mac-0": "52:54:00:a2:3c:e8",
"net-source-0": "net:default",
"net-nwfilter-0": "",
"net-model-0": "virtio",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.networks
self.assertEqual(len(self.instance.networks), 1)
self.assertEqual(self.instance.networks[0]["mac"], "52:54:00:a2:3c:e8")
# The second NIC's form changes only the second NIC
self.client.post(
reverse("instances:add_network", args=[self.instance.id]),
{"add-net-mac": "52:54:00:a2:3c:e9", "add-net-network": "net:default"},
HTTP_REFERER=reverse("index"),
)
self.client.post(
reverse("instances:change_network", args=[self.instance.id]),
{
"net-old-mac-1": "52:54:00:a2:3c:e9",
"net-mac-1": "52:54:00:a2:3c:ea",
"net-source-1": "net:default",
"net-nwfilter-1": "",
"net-model-1": "e1000",
},
HTTP_REFERER=reverse("index"),
)
del self.instance.networks
self.assertEqual([n["mac"] for n in self.instance.networks], ["52:54:00:a2:3c:e8", "52:54:00:a2:3c:ea"])
self.assertEqual(self.instance.networks[1]["model"], "e1000")
# Restore the shared test VM
self.client.post(
reverse("instances:delete_network", args=[self.instance.id]),
{"delete_network": "52:54:00:a2:3c:ea"},
HTTP_REFERER=reverse("index"),
)
self.client.post(
reverse("instances:change_network", args=[self.instance.id]),
{
"net-old-mac-0": "52:54:00:a2:3c:e8",
"net-mac-0": "52:54:00:a2:3c:e7",
"net-source-0": "net:default",
"net-model-0": "virtio",
},
HTTP_REFERER=reverse("index"),
)
del self.instance.networks
self.assertEqual([n["mac"] for n in self.instance.networks], ["52:54:00:a2:3c:e7"])
def test_add_delete_network(self):
self.assertEqual(len(self.instance.networks), 1)
net_mac = self.instance.networks[0]["mac"]
response = self.client.post(
reverse("instances:add_network", args=[self.instance.id]),
{
"add-net-mac": "52:54:00:a2:3c:e9",
"add-net-network": "net:default",
"add_net-nwfilter": "",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.networks
self.assertEqual(len(self.instance.networks), 2)
self.assertEqual(self.instance.networks[1]["mac"], "52:54:00:a2:3c:e9")
response = self.client.post(
reverse("instances:delete_network", args=[self.instance.id]),
{
"delete_network": "52:54:00:a2:3c:e9",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.networks
self.assertEqual(len(self.instance.networks), 1)
self.assertEqual(self.instance.networks[0]["mac"], net_mac)
def test_set_link_state(self):
self.assertEqual(self.instance.networks[0]["state"], "up")
response = self.client.post(
reverse("instances:set_link_state", args=[self.instance.id]),
{
"mac": self.instance.networks[0]["mac"],
"set_link_state": "up",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.networks
self.assertEqual(self.instance.networks[0]["state"], "down")
def test_set_unset_qos(self):
self.assertEqual(len(self.instance.qos.keys()), 0)
net_mac = self.instance.networks[0]["mac"]
response = self.client.post(
reverse("instances:set_qos", args=[self.instance.id]),
{
"net-mac-0": net_mac,
"qos_direction": "inbound",
"qos_average": 1,
"qos_peak": 1,
"qos_burst": 1,
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.qos
self.assertEqual(len(self.instance.qos.keys()), 1)
response = self.client.post(
reverse("instances:unset_qos", args=[self.instance.id]),
{
"net-mac": net_mac,
"qos_direction": "inbound",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.qos
self.assertEqual(len(self.instance.qos.keys()), 0)
# test on running instance
# self.instance.proxy.start()
# response = self.client.post(
# reverse('instances:set_qos', args=[self.instance.id]),
# {
# 'net-mac-0': net_mac,
# 'qos_direction': 'inbound',
# 'qos_average': 1,
# 'qos_peak': 1,
# 'qos_burst': 1,
# },
# HTTP_REFERER=reverse('index'),
# )
# self.assertEqual(response.status_code, 302)
# self.instance.proxy.force_shutdown()
def test_change_options(self):
self.assertEqual(self.instance.title, "")
self.assertEqual(self.instance.description, "")
response = self.client.post(
reverse("instances:change_options", args=[self.instance.id]),
{
"title": "test-vm-title",
"description": "test-vm description",
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
del self.instance.title
del self.instance.description
self.assertEqual(self.instance.title, "test-vm-title")
self.assertEqual(self.instance.description, "test-vm description")
def test_flavors(self):
response = self.client.get(reverse("instances:flavor_create"))
self.assertEqual(response.status_code, 200)
response = self.client.post(
reverse("instances:flavor_create"),
{
"label": "test_flavor",
"memory": 256,
"vcpu": 1,
"disk": 10,
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
id = Flavor.objects.last().id
response = self.client.get(reverse("instances:flavor_update", args=[id]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
reverse("instances:flavor_update", args=[id]),
{
"label": "test_flavor_",
"memory": 256,
"vcpu": 1,
"disk": 10,
},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
response = self.client.get(reverse("instances:flavor_delete", args=[id]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
reverse("instances:flavor_delete", args=[id]),
{},
HTTP_REFERER=reverse("index"),
)
self.assertEqual(response.status_code, 302)
# def donot_test_instance(self):
# compute: Compute = Compute.objects.get(pk=1)
# user: User = User.objects.get(pk=1)
# # delete started instance with disks
# self.instance.proxy.start()
# del self.instance.status
# self.assertEqual(self.instance.status, 1)
# # create volume
# response = self.client.post(
# reverse('create_volume', args=[compute.id, 'default']),
# {
# 'name': 'test3',
# 'format': 'qcow2',
# 'size': '1',
# 'meta_prealloc': False,
# },
# )
# self.assertRedirects(response, reverse('storage', args=[compute.id, 'default']))
# # delete volume
# response = self.client.post(
# reverse('instances:delete_vol', args=[instance.id]),
# {
# 'storage': 'default',
# 'dev': 'vdb',
# 'name': 'test3.qcow2',
# },
# HTTP_REFERER=reverse('index'),
# )
# self.assertEqual(response.status_code, 302)
# , list(response.context['messages'])[0]
Reference in a new issue View git blame Copy permalink