webvirtcloud/instances/tests/test_views.py

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 operations.models import Operation

from . import livetest
from .test_create_instance import run
from instances.models import Flavor, Instance
from instances.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",
                "qemu_ga": False,
                "virtio": True,
                "add_cdrom": "None",
                "add_input": "None",
                "create": True,
                "submission_key": "key-create-destroy",
            },
        )
        self.assertEqual(response.status_code, 302)
        run(Operation.objects.get(kind="instance.create"))

        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)

        self.instance.proxy.create_snapshot("test-snapshot")

        for cached in ("snapshots", "internal_snapshots", "external_snapshots"):
            self.instance.__dict__.pop(cached, None)
        self.assertEqual(len(self.instance.snapshots), 1)

        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)

        for cached in ("snapshots", "internal_snapshots", "external_snapshots"):
            self.instance.__dict__.pop(cached, None)
        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": "",
                "submission_key": "key-clone",
            },
            HTTP_REFERER=reverse("index"),
        )
        self.assertEqual(response.status_code, 302)
        run(Operation.objects.get(kind="instance.clone"))
        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")

        clone.proxy.create_snapshot("test")

        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.config_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.config_qos
        self.assertEqual(len(self.instance.config_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.config_qos
        self.assertEqual(len(self.instance.config_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]