j3d1/webvirtcloud
1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-10-18 00:50:07 +00:00
Code Issues Projects Releases Wiki Activity

fix(tests): decouple tests from local libvirt, support configurable test host, and fix URL/logging bugs

Browse source
This commit is contained in:
catborise 2026-09-25 11:51:45 +03:00
parent c31ed28042
commit 60565b05b3
6 changed files with 153 additions and 130 deletions
Download patch file Download diff file Expand all files Collapse all files

63
accounts/tests.py
View file

@ -6,9 +6,7 @@ from django.contrib.auth.models import Permission
from django.shortcuts import reverse from django.shortcuts import reverse
from django.test import Client, TestCase from django.test import Client, TestCase
from instances.models import Instance from instances.models import Instance
from instances.utils import refr
from libvirt import VIR_DOMAIN_UNDEFINE_NVRAM
from vrtManager.create import wvmCreate
from accounts.forms import UserInstanceForm, UserSSHKeyForm from accounts.forms import UserInstanceForm, UserSSHKeyForm
from accounts.models import UserInstance, UserSSHKey from accounts.models import UserInstance, UserSSHKey
@ -25,49 +23,26 @@ class AccountsTestCase(TestCase):
cls.admin_user = User.objects.get(pk=1) cls.admin_user = User.objects.get(pk=1)
cls.test_user = User.objects.create_user(username="test", password="test") cls.test_user = User.objects.create_user(username="test", password="test")
# Add localhost compute # Add compute
cls.compute = Compute( cls.compute = Compute.objects.create(
name="test-compute", name="accounts-test-compute",
hostname="localhost", hostname="localhost",
login="", login="",
password="", password="",
details="local", details="local",
type=4, type=4,
) )
cls.compute.save() cls.instance = Instance.objects.create(
compute=cls.compute,
cls.connection = wvmCreate( name="test-vm",
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.objects.get(pk=1)
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
# Destroy testing vm if getattr(cls, "instance", None):
cls.instance.proxy.delete_all_disks() cls.instance.delete()
cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM) if getattr(cls, "compute", None):
cls.compute.delete()
super().tearDownClass() super().tearDownClass()
def setUp(self): def setUp(self):
@ -99,7 +74,7 @@ class AccountsTestCase(TestCase):
response = client.post( response = client.post(
reverse("accounts:login"), {"username": "test", "password": "test"} reverse("accounts:login"), {"username": "test", "password": "test"}
) )
self.assertRedirects(response, reverse("accounts:profile")) self.assertRedirects(response, settings.LOGIN_REDIRECT_URL)
response = client.get(reverse("accounts:logout")) response = client.get(reverse("accounts:logout"))
self.assertRedirects(response, reverse("accounts:login")) self.assertRedirects(response, reverse("accounts:login"))
@ -156,7 +131,9 @@ class AccountsTestCase(TestCase):
response, reverse("accounts:account", args=[self.test_user.id]) response, reverse("accounts:account", args=[self.test_user.id])
) )
user_instance: UserInstance = UserInstance.objects.get(pk=1) user_instance: UserInstance = UserInstance.objects.get(
user=self.test_user, instance=self.instance
)
self.assertEqual(user_instance.user, self.test_user) self.assertEqual(user_instance.user, self.test_user)
self.assertEqual(user_instance.instance, self.instance) self.assertEqual(user_instance.instance, self.instance)
self.assertEqual(user_instance.is_change, False) self.assertEqual(user_instance.is_change, False)
@ -183,7 +160,9 @@ class AccountsTestCase(TestCase):
response, reverse("accounts:account", args=[self.test_user.id]) response, reverse("accounts:account", args=[self.test_user.id])
) )
user_instance: UserInstance = UserInstance.objects.get(pk=1) user_instance = UserInstance.objects.get(
user=self.test_user, instance=self.instance
)
self.assertEqual(user_instance.user, self.test_user) self.assertEqual(user_instance.user, self.test_user)
self.assertEqual(user_instance.instance, self.instance) self.assertEqual(user_instance.instance, self.instance)
self.assertEqual(user_instance.is_change, True) self.assertEqual(user_instance.is_change, True)
@ -250,14 +229,14 @@ class AccountsTestCase(TestCase):
) )
self.assertRedirects(response, reverse("accounts:profile")) self.assertRedirects(response, reverse("accounts:profile"))
key = UserSSHKey.objects.get(pk=1) key = UserSSHKey.objects.get(keyname="keyname")
self.assertEqual(key.keyname, "keyname") self.assertEqual(key.keyname, "keyname")
self.assertEqual(key.keypublic, self.rsa_key) self.assertEqual(key.keypublic, self.rsa_key)
response = self.client.get(reverse("accounts:ssh_key_delete", args=[1])) response = self.client.get(reverse("accounts:ssh_key_delete", args=[key.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
response = self.client.post(reverse("accounts:ssh_key_delete", args=[1])) response = self.client.post(reverse("accounts:ssh_key_delete", args=[key.id]))
self.assertRedirects(response, reverse("accounts:profile")) self.assertRedirects(response, reverse("accounts:profile"))
def test_validate_ssh_key(self): def test_validate_ssh_key(self):

12
accounts/urls.py
View file

@ -30,6 +30,12 @@ urlpatterns = [
), ),
path("ssh_key/create/", views.ssh_key_create, name="ssh_key_create"), path("ssh_key/create/", views.ssh_key_create, name="ssh_key_create"),
path("ssh_key/<int:pk>/delete/", views.ssh_key_delete, name="ssh_key_delete"), path("ssh_key/<int:pk>/delete/", views.ssh_key_delete, name="ssh_key_delete"),
path("email_otp/", views.email_otp, name="email_otp"),
path(
"admin_email_otp/<int:user_id>/",
views.admin_email_otp,
name="admin_email_otp",
),
] ]
if settings.OTP_ENABLED: if settings.OTP_ENABLED:
@ -42,12 +48,6 @@ if settings.OTP_ENABLED:
), ),
name="login", name="login",
), ),
path("email_otp/", views.email_otp, name="email_otp"),
path(
"admin_email_otp/<int:user_id>/",
views.admin_email_otp,
name="admin_email_otp",
),
] ]
else: else:
urlpatterns += ( urlpatterns += (

112
computes/tests.py
View file

@ -1,21 +1,47 @@
import os
from django.core.exceptions import ObjectDoesNotExist from django.core.exceptions import ObjectDoesNotExist
from django.shortcuts import reverse from django.shortcuts import reverse
from django.test import TestCase from django.test import TestCase
from vrtManager.connection import CONN_SOCKET, connection_manager
from .models import Compute from .models import Compute
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", CONN_SOCKET))
class ComputesTestCase(TestCase): class ComputesTestCase(TestCase):
def setUp(self): def setUp(self):
self.client.login(username="admin", password="admin") self.client.login(username="admin", password="admin")
Compute( self.compute, _ = Compute.objects.get_or_create(
name="local", name="computes-test-compute",
hostname="localhost", defaults={
login="", "hostname": TEST_COMPUTE_HOST,
password="", "login": TEST_COMPUTE_LOGIN,
details="local", "password": TEST_COMPUTE_PASSWORD,
type=4, "details": "test",
).save() "type": TEST_COMPUTE_TYPE,
},
)
def tearDown(self):
Compute.objects.filter(name="computes-test-compute").delete()
super().tearDown()
def _require_live_compute(self):
try:
conn = connection_manager.get_connection(
self.compute.hostname,
self.compute.login,
self.compute.password,
self.compute.type,
)
if not (conn and conn.isAlive()):
self.skipTest("Live libvirt compute host not available")
except Exception:
self.skipTest("Live libvirt compute host not available")
def test_index(self): def test_index(self):
response = self.client.get(reverse("computes")) response = self.client.get(reverse("computes"))
@ -36,15 +62,16 @@ class ComputesTestCase(TestCase):
) )
self.assertRedirects(response, reverse("computes")) self.assertRedirects(response, reverse("computes"))
compute = Compute.objects.get(pk=2) compute = Compute.objects.get(name="l1")
self.assertEqual(compute.name, "l1") self.assertEqual(compute.name, "l1")
self.assertEqual(compute.details, "Created") self.assertEqual(compute.details, "Created")
created_id = compute.id
response = self.client.get(reverse("compute_update", args=[2])) response = self.client.get(reverse("compute_update", args=[created_id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
response = self.client.post( response = self.client.post(
reverse("compute_update", args=[2]), reverse("compute_update", args=[created_id]),
{ {
"name": "l2", "name": "l2",
"details": "Updated", "details": "Updated",
@ -54,94 +81,94 @@ class ComputesTestCase(TestCase):
) )
self.assertRedirects(response, reverse("computes")) self.assertRedirects(response, reverse("computes"))
compute = Compute.objects.get(pk=2) compute = Compute.objects.get(id=created_id)
self.assertEqual(compute.name, "l2") self.assertEqual(compute.name, "l2")
self.assertEqual(compute.details, "Updated") self.assertEqual(compute.details, "Updated")
response = self.client.get(reverse("compute_delete", args=[2])) response = self.client.get(reverse("compute_delete", args=[created_id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
response = self.client.post(reverse("compute_delete", args=[2])) response = self.client.post(reverse("compute_delete", args=[created_id]))
self.assertRedirects(response, reverse("computes")) self.assertRedirects(response, reverse("computes"))
with self.assertRaises(ObjectDoesNotExist): with self.assertRaises(ObjectDoesNotExist):
Compute.objects.get(id=2) Compute.objects.get(id=created_id)
def test_overview(self): def test_overview(self):
response = self.client.get(reverse("overview", args=[1])) self._require_live_compute()
response = self.client.get(reverse("overview", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_graph(self): def test_graph(self):
response = self.client.get(reverse("compute_graph", args=[1])) self._require_live_compute()
response = self.client.get(reverse("compute_graph", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_instances(self): def test_instances(self):
response = self.client.get(reverse("instances", args=[1])) self._require_live_compute()
response = self.client.get(reverse("instances", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_storages(self): def test_storages(self):
response = self.client.get(reverse("storages", args=[1])) self._require_live_compute()
response = self.client.get(reverse("storages", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_storage(self):
pass
def test_default_storage_volumes(self): def test_default_storage_volumes(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse("volumes", kwargs={"compute_id": 1, "pool": "default"}) reverse("volumes", kwargs={"compute_id": self.compute.id, "pool": "default"})
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_default_storage(self): def test_default_storage(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse("storage", kwargs={"compute_id": 1, "pool": "default"}) reverse("storage", kwargs={"compute_id": self.compute.id, "pool": "default"})
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_networks(self): def test_networks(self):
response = self.client.get(reverse("networks", args=[1])) self._require_live_compute()
response = self.client.get(reverse("networks", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_default_network(self): def test_default_network(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse("network", kwargs={"compute_id": 1, "pool": "default"}) reverse("network", kwargs={"compute_id": self.compute.id, "pool": "default"})
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_interfaces(self): def test_interfaces(self):
response = self.client.get(reverse("interfaces", args=[1])) self._require_live_compute()
response = self.client.get(reverse("interfaces", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
# TODO: add test for single interface
def test_nwfilters(self): def test_nwfilters(self):
response = self.client.get(reverse("nwfilters", args=[1])) self._require_live_compute()
response = self.client.get(reverse("nwfilters", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
# TODO: add test for single nwfilter
def test_secrets(self): def test_secrets(self):
response = self.client.get(reverse("virtsecrets", args=[1])) self._require_live_compute()
response = self.client.get(reverse("virtsecrets", args=[self.compute.id]))
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
# def test_create_instance_select_type(self):
# response = self.client.get(reverse('create_instance_select_type', args=[1]))
# self.assertEqual(response.status_code, 200)
# TODO: create_instance
def test_machines(self): def test_machines(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse("machines", kwargs={"compute_id": 1, "arch": "x86_64"}) reverse("machines", kwargs={"compute_id": self.compute.id, "arch": "x86_64"})
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_compute_disk_buses(self): def test_compute_disk_buses(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse( reverse(
"buses", "buses",
kwargs={ kwargs={
"compute_id": 1, "compute_id": self.compute.id,
"arch": "x86_64", "arch": "x86_64",
"machine": "pc", "machine": "pc",
"disk": "disk", "disk": "disk",
@ -151,9 +178,10 @@ class ComputesTestCase(TestCase):
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)
def test_dom_capabilities(self): def test_dom_capabilities(self):
self._require_live_compute()
response = self.client.get( response = self.client.get(
reverse( reverse(
"domcaps", kwargs={"compute_id": 1, "arch": "x86_64", "machine": "pc"} "domcaps", kwargs={"compute_id": self.compute.id, "arch": "x86_64", "machine": "pc"}
) )
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)

90
instances/tests.py
View file

@ -1,4 +1,6 @@
import os
import re import re
import unittest
from accounts.models import UserAttributes, UserInstance, UserSSHKey from accounts.models import UserAttributes, UserInstance, UserSSHKey
from appsettings.models import AppSettings from appsettings.models import AppSettings
@ -16,6 +18,11 @@ from vrtManager.util import randomUUID
from .models import Flavor, Instance from .models import Flavor, Instance
from .utils import refr from .utils import refr
TEST_COMPUTE_HOST = os.environ.get("TEST_LIBVIRT_HOST", "localhost")
TEST_COMPUTE_LOGIN = os.environ.get("TEST_LIBVIRT_LOGIN", "")
TEST_COMPUTE_PASSWORD = os.environ.get("TEST_LIBVIRT_PASSWORD", "")
TEST_COMPUTE_TYPE = int(os.environ.get("TEST_LIBVIRT_TYPE", 4))
class InstancesTestCase(TestCase): class InstancesTestCase(TestCase):
@classmethod @classmethod
@ -24,7 +31,7 @@ class InstancesTestCase(TestCase):
# Add users for testing purposes # Add users for testing purposes
User = get_user_model() User = get_user_model()
cls.admin_user = User.objects.get(pk=1) 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") cls.test_user = User.objects.create(username="test-user")
UserAttributes.objects.create( UserAttributes.objects.create(
user=cls.test_user, user=cls.test_user,
@ -36,48 +43,57 @@ class InstancesTestCase(TestCase):
permission = Permission.objects.get(codename="clone_instances") permission = Permission.objects.get(codename="clone_instances")
cls.test_user.user_permissions.add(permission) cls.test_user.user_permissions.add(permission)
# Add localhost compute # Add compute
cls.compute = Compute( cls.compute = Compute.objects.create(
name="test-compute", name="instances-test-compute",
hostname="localhost", hostname=TEST_COMPUTE_HOST,
login="", login=TEST_COMPUTE_LOGIN,
password="", password=TEST_COMPUTE_PASSWORD,
details="local", details="test",
type=4, type=TEST_COMPUTE_TYPE,
)
cls.compute.save()
cls.connection = wvmCreate(
cls.compute.hostname,
cls.compute.login,
cls.compute.password,
cls.compute.type,
) )
# Add disks for testing try:
cls.connection.create_volume( cls.connection = wvmCreate(
"default", cls.compute.hostname,
"test-volume", cls.compute.login,
1, cls.compute.password,
"qcow2", cls.compute.type,
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 # Add disks for testing
cls.connection._defineXML(cls.xml) cls.connection.create_volume(
refr(cls.compute) "default",
cls.instance: Instance = Instance.objects.get(pk=1) "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.filter(compute=cls.compute).first()
except Exception as e:
cls.instance = None
raise unittest.SkipTest(f"Live libvirt host not available for instances tests: {e}")
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
# Destroy testing vm # Destroy testing vm
cls.instance.proxy.delete_all_disks() if getattr(cls, "instance", None):
cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM) try:
cls.instance.proxy.delete_all_disks()
cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM)
except Exception:
pass
if getattr(cls, "compute", None):
cls.compute.delete()
super().tearDownClass() super().tearDownClass()
def setUp(self): def setUp(self):
@ -94,7 +110,7 @@ class InstancesTestCase(TestCase):
def test_create_select_type(self): def test_create_select_type(self):
response = self.client.get( response = self.client.get(
reverse("instances:create_instance_select_type", args=[1]) reverse("instances:create_instance_select_type", args=[self.compute.id])
) )
self.assertEqual(response.status_code, 200) self.assertEqual(response.status_code, 200)

4
logs/views.py
View file

@ -16,10 +16,10 @@ def addlogmsg(user, host, instance, message):
:param message: :param message:
:return: :return:
""" """
add_log_msg = Logs(user=user, host=host, instance=instance, message=message) add_log_msg = Logs(user=user, host=host, instance=instance, message=str(message))
add_log_msg.save() add_log_msg.save()
logger = logging.getLogger("") logger = logging.getLogger("")
logger.info(user + " " + host+ " " + instance+ " " + message) logger.info(f"{user} {host} {instance} {message}")
@superuser_only @superuser_only

2
networks/views.py
View file

@ -140,7 +140,7 @@ def network(request, compute_id, pool):
xml = conn._XMLDesc(0) xml = conn._XMLDesc(0)
except libvirtError as lib_err: except libvirtError as lib_err:
messages.error(request, lib_err) messages.error(request, lib_err)
return HttpResponseRedirect(reverse("networks", args=compute_id)) return HttpResponseRedirect(reverse("networks", args=[compute_id]))
if request.method == "POST": if request.method == "POST":
if "start" in request.POST: if "start" in request.POST: