webvirtcloud/instances/tests.py

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 instances.views import instance
from libvirt import VIR_DOMAIN_UNDEFINE_NVRAM
from vrtManager.create import wvmCreate
from vrtManager.util import randomUUID

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]