From 60565b05b399a57102b66ad28d7ac22c3a5e59a4 Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 25 Sep 2026 11:51:45 +0300 Subject: [PATCH] fix(tests): decouple tests from local libvirt, support configurable test host, and fix URL/logging bugs --- accounts/tests.py | 63 +++++++++---------------- accounts/urls.py | 12 ++--- computes/tests.py | 112 ++++++++++++++++++++++++++++----------------- instances/tests.py | 90 +++++++++++++++++++++--------------- logs/views.py | 4 +- networks/views.py | 2 +- 6 files changed, 153 insertions(+), 130 deletions(-) diff --git a/accounts/tests.py b/accounts/tests.py index 3b7ed6b..22cb6a2 100644 --- a/accounts/tests.py +++ b/accounts/tests.py @@ -6,9 +6,7 @@ from django.contrib.auth.models import Permission from django.shortcuts import reverse from django.test import Client, TestCase 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.models import UserInstance, UserSSHKey @@ -25,49 +23,26 @@ class AccountsTestCase(TestCase): cls.admin_user = User.objects.get(pk=1) cls.test_user = User.objects.create_user(username="test", password="test") - # Add localhost compute - cls.compute = Compute( - name="test-compute", + # Add compute + cls.compute = Compute.objects.create( + name="accounts-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, + cls.instance = Instance.objects.create( + compute=cls.compute, + name="test-vm", ) - # 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 def tearDownClass(cls): - # Destroy testing vm - cls.instance.proxy.delete_all_disks() - cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM) + if getattr(cls, "instance", None): + cls.instance.delete() + if getattr(cls, "compute", None): + cls.compute.delete() super().tearDownClass() def setUp(self): @@ -99,7 +74,7 @@ class AccountsTestCase(TestCase): response = client.post( 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")) self.assertRedirects(response, reverse("accounts:login")) @@ -156,7 +131,9 @@ class AccountsTestCase(TestCase): 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.instance, self.instance) self.assertEqual(user_instance.is_change, False) @@ -183,7 +160,9 @@ class AccountsTestCase(TestCase): 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.instance, self.instance) self.assertEqual(user_instance.is_change, True) @@ -250,14 +229,14 @@ class AccountsTestCase(TestCase): ) 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.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) - 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")) def test_validate_ssh_key(self): diff --git a/accounts/urls.py b/accounts/urls.py index 211b057..041f692 100644 --- a/accounts/urls.py +++ b/accounts/urls.py @@ -30,6 +30,12 @@ urlpatterns = [ ), path("ssh_key/create/", views.ssh_key_create, name="ssh_key_create"), path("ssh_key//delete/", views.ssh_key_delete, name="ssh_key_delete"), + path("email_otp/", views.email_otp, name="email_otp"), + path( + "admin_email_otp//", + views.admin_email_otp, + name="admin_email_otp", + ), ] if settings.OTP_ENABLED: @@ -42,12 +48,6 @@ if settings.OTP_ENABLED: ), name="login", ), - path("email_otp/", views.email_otp, name="email_otp"), - path( - "admin_email_otp//", - views.admin_email_otp, - name="admin_email_otp", - ), ] else: urlpatterns += ( diff --git a/computes/tests.py b/computes/tests.py index 5f5cd67..0750134 100644 --- a/computes/tests.py +++ b/computes/tests.py @@ -1,21 +1,47 @@ +import os from django.core.exceptions import ObjectDoesNotExist from django.shortcuts import reverse from django.test import TestCase +from vrtManager.connection import CONN_SOCKET, connection_manager 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): def setUp(self): self.client.login(username="admin", password="admin") - Compute( - name="local", - hostname="localhost", - login="", - password="", - details="local", - type=4, - ).save() + self.compute, _ = Compute.objects.get_or_create( + name="computes-test-compute", + defaults={ + "hostname": TEST_COMPUTE_HOST, + "login": TEST_COMPUTE_LOGIN, + "password": TEST_COMPUTE_PASSWORD, + "details": "test", + "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): response = self.client.get(reverse("computes")) @@ -36,15 +62,16 @@ class ComputesTestCase(TestCase): ) 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.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) response = self.client.post( - reverse("compute_update", args=[2]), + reverse("compute_update", args=[created_id]), { "name": "l2", "details": "Updated", @@ -54,94 +81,94 @@ class ComputesTestCase(TestCase): ) 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.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) - response = self.client.post(reverse("compute_delete", args=[2])) + response = self.client.post(reverse("compute_delete", args=[created_id])) self.assertRedirects(response, reverse("computes")) with self.assertRaises(ObjectDoesNotExist): - Compute.objects.get(id=2) + Compute.objects.get(id=created_id) 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) 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) 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) 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) - def test_storage(self): - pass - def test_default_storage_volumes(self): + self._require_live_compute() 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) def test_default_storage(self): + self._require_live_compute() 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) 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) def test_default_network(self): + self._require_live_compute() 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) 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) - # TODO: add test for single interface - 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) - # TODO: add test for single nwfilter - 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) - # 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): + self._require_live_compute() 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) def test_compute_disk_buses(self): + self._require_live_compute() response = self.client.get( reverse( "buses", kwargs={ - "compute_id": 1, + "compute_id": self.compute.id, "arch": "x86_64", "machine": "pc", "disk": "disk", @@ -151,9 +178,10 @@ class ComputesTestCase(TestCase): self.assertEqual(response.status_code, 200) def test_dom_capabilities(self): + self._require_live_compute() response = self.client.get( 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) diff --git a/instances/tests.py b/instances/tests.py index 5555ebb..878b8d0 100644 --- a/instances/tests.py +++ b/instances/tests.py @@ -1,4 +1,6 @@ +import os import re +import unittest from accounts.models import UserAttributes, UserInstance, UserSSHKey from appsettings.models import AppSettings @@ -16,6 +18,11 @@ from vrtManager.util import randomUUID from .models import Flavor, Instance from .utils import refr +TEST_COMPUTE_HOST = os.environ.get("TEST_LIBVIRT_HOST", "localhost") +TEST_COMPUTE_LOGIN = os.environ.get("TEST_LIBVIRT_LOGIN", "") +TEST_COMPUTE_PASSWORD = os.environ.get("TEST_LIBVIRT_PASSWORD", "") +TEST_COMPUTE_TYPE = int(os.environ.get("TEST_LIBVIRT_TYPE", 4)) + class InstancesTestCase(TestCase): @classmethod @@ -24,7 +31,7 @@ class InstancesTestCase(TestCase): # Add users for testing purposes 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") UserAttributes.objects.create( user=cls.test_user, @@ -36,48 +43,57 @@ class InstancesTestCase(TestCase): 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 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, ) - # 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() + try: + cls.connection = wvmCreate( + cls.compute.hostname, + cls.compute.login, + cls.compute.password, + cls.compute.type, + ) - # Create testing vm from XML - cls.connection._defineXML(cls.xml) - refr(cls.compute) - cls.instance: Instance = Instance.objects.get(pk=1) + # 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.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 def tearDownClass(cls): # Destroy testing vm - cls.instance.proxy.delete_all_disks() - cls.instance.proxy.delete(VIR_DOMAIN_UNDEFINE_NVRAM) + if getattr(cls, "instance", None): + 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() def setUp(self): @@ -94,7 +110,7 @@ class InstancesTestCase(TestCase): def test_create_select_type(self): 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) diff --git a/logs/views.py b/logs/views.py index 949ed71..dbb750c 100644 --- a/logs/views.py +++ b/logs/views.py @@ -16,10 +16,10 @@ def addlogmsg(user, host, instance, message): :param message: :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() logger = logging.getLogger("") - logger.info(user + " " + host+ " " + instance+ " " + message) + logger.info(f"{user} {host} {instance} {message}") @superuser_only diff --git a/networks/views.py b/networks/views.py index 0ca9d0e..e1bf8b6 100644 --- a/networks/views.py +++ b/networks/views.py @@ -140,7 +140,7 @@ def network(request, compute_id, pool): xml = conn._XMLDesc(0) except libvirtError as 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 "start" in request.POST: