1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2024-12-25 15:45:23 +00:00
webvirtcloud/instances/tests.py
2020-10-15 11:58:42 +06:00

1026 lines
35 KiB
Python

import re
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 TestCase
from libvirt import VIR_DOMAIN_UNDEFINE_NVRAM
from vrtManager.create import wvmCreate
from vrtManager.util import randomUUID
from instances.views import instance
from .models import Flavor, Instance
from .utils import refr
class InstancesTestCase(TestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
# Add users for testing purposes
User = get_user_model()
cls.admin_user = User.objects.get(pk=1)
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 localhost compute
cls.compute = Compute(
name='test-compute',
hostname='localhost',
login='',
password='',
details='local',
type=4,
)
cls.compute.save()
cls.connection = wvmCreate(
cls.compute.hostname,
cls.compute.login,
cls.compute.password,
cls.compute.type,
)
# Add disks for testing
cls.connection.create_volume(
'default',
'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)
cls.instance: Instance = Instance.objects.get(pk=1)
@classmethod
def tearDownClass(cls):
# Destroy testing vm
cls.instance.proxy.delete_all_disks()
cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM)
super().tearDownClass()
def setUp(self):
self.client.login(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=[1]))
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': 'test',
'firmware': 'BIOS',
'vcpu': 1,
'vcpu_mode': 'host-model',
'memory': 128,
'device0': 'disk',
'bus0': 'virtio',
'images': 'test-volume.qcow2',
'storage-control': 'default',
'image-control': 'test.qcow2',
'networks': 'default',
'network-control': 'default',
'cache_mode': 'directsync',
'nwfilter': '',
'graphics': 'spice',
'video': 'vga',
'listener_addr': '0.0.0.0',
'console_pass': '',
'qemu_ga': False,
'virtio': True,
'create': True,
},
)
self.assertEqual(response.status_code, 302)
instance_qs: Instance = Instance.objects.filter(name='test')
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('test-vm', 'test-vm-xml')
xml = re.sub('\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='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': 'default',
'name': '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': 'default',
'dev': 'vdb',
'name': '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': 'default',
'vols': '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': '/var/lib/libvirt/images/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': 'test-vm-clone',
'clone-net-mac-0': 'de:ad:be:ef:de:ad',
'disk-vda': '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='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()
UserInstance.objects.create(user=self.test_user, instance=self.instance)
self.client.force_login(self.test_user)
response = self.client.post(
reverse('instances:clone', args=[self.instance.id]),
{
'name': 'test-vm-clone',
'clone-net-mac-0': 'de:ad:be:ef:de:ad',
'disk-vda': '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': 'test-vm',
'clone-net-mac-0': 'de:ad:be:ef:de:ad',
'disk-vda': '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': 'test-vm-clone',
'clone-net-mac-0': 'gh:ad:be:ef:de:ad',
'disk-vda': '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]),
{
'type': 'spice',
'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_check_instance(self):
response = self.client.get(reverse('instances:check_instance', args=['test-vm']))
self.assertEqual(response.status_code, 200)
self.assertJSONEqual(response.content, {'vname': 'test-vm', 'exists': True})
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.get(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.get(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.get(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.get(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.get(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.get(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.get(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-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(self.instance.networks[0]['mac'], '52:54:00:a2:3c:e8')
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]