diff --git a/accounts/test_utils.py b/accounts/test_utils.py
new file mode 100644
index 0000000..124035d
--- /dev/null
+++ b/accounts/test_utils.py
@@ -0,0 +1,69 @@
+import base64
+import struct
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.core import mail
+from django.test import TestCase
+from django_otp.plugins.otp_totp.models import TOTPDevice
+
+from accounts.utils import get_user_totp_device, send_email_with_otp, validate_ssh_key
+
+
+class AccountsUtilsTestCase(TestCase):
+ def setUp(self):
+ User = get_user_model()
+ self.user = User.objects.create_user(
+ username="test_otp_user", email="user@example.com", password="password123"
+ )
+
+ def test_validate_ssh_key_valid(self):
+ blob = (
+ struct.pack(">I", 7)
+ + b"ssh-rsa"
+ + struct.pack(">I", 3)
+ + b"\x01\x00\x01"
+ + struct.pack(">I", 8)
+ + b"\x00" * 8
+ )
+ b64 = base64.b64encode(blob).decode()
+ valid_key = f"ssh-rsa {b64} user@example.com"
+ self.assertTrue(validate_ssh_key(valid_key))
+
+ def test_validate_ssh_key_invalid_parts(self):
+ self.assertFalse(validate_ssh_key("only_two parts"))
+ self.assertFalse(validate_ssh_key("one_part"))
+ self.assertFalse(validate_ssh_key("four parts key here now"))
+
+ def test_validate_ssh_key_invalid_base64(self):
+ self.assertFalse(validate_ssh_key("ssh-rsa not-valid-b64! user@example.com"))
+
+ def test_validate_ssh_key_struct_error(self):
+ # Base64 with less than 4 bytes cannot unpack length
+ short_b64 = base64.b64encode(b"ab").decode()
+ self.assertFalse(validate_ssh_key(f"ssh-rsa {short_b64} user@example.com"))
+
+ def test_validate_ssh_key_type_mismatch(self):
+ blob = struct.pack(">I", 7) + b"ssh-rsa" + b"\x00" * 8
+ b64 = base64.b64encode(blob).decode()
+ # header says ssh-ed25519 but payload has ssh-rsa
+ self.assertFalse(validate_ssh_key(f"ssh-ed25519 {b64} user@example.com"))
+
+ def test_get_user_totp_device_creates_new(self):
+ self.assertEqual(TOTPDevice.objects.filter(user=self.user).count(), 0)
+ device = get_user_totp_device(self.user)
+ self.assertIsInstance(device, TOTPDevice)
+ self.assertEqual(TOTPDevice.objects.filter(user=self.user).count(), 1)
+
+ def test_get_user_totp_device_retrieves_existing(self):
+ device1 = get_user_totp_device(self.user)
+ device2 = get_user_totp_device(self.user)
+ self.assertEqual(device1.id, device2.id)
+ self.assertEqual(TOTPDevice.objects.filter(user=self.user).count(), 1)
+
+ def test_send_email_with_otp(self):
+ device = get_user_totp_device(self.user)
+ send_email_with_otp(self.user, device)
+ self.assertEqual(len(mail.outbox), 1)
+ self.assertIn("OTP QR Code", mail.outbox[0].subject)
+ self.assertEqual(mail.outbox[0].to, ["user@example.com"])
diff --git a/admin/tests.py b/admin/tests.py
index 92d261c..488fd2b 100644
--- a/admin/tests.py
+++ b/admin/tests.py
@@ -24,27 +24,27 @@ class AdminTestCase(TestCase):
self.assertRedirects(response, reverse("admin:group_list"))
group = Group.objects.get(name="Test Group")
- self.assertEqual(group.id, 1)
+ group_id = group.id
- response = self.client.get(reverse("admin:group_update", args=[1]))
+ response = self.client.get(reverse("admin:group_update", args=[group_id]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
- reverse("admin:group_update", args=[1]), {"name": "Updated Group Test"}
+ reverse("admin:group_update", args=[group_id]), {"name": "Updated Group Test"}
)
self.assertRedirects(response, reverse("admin:group_list"))
- group = Group.objects.get(id=1)
+ group = Group.objects.get(id=group_id)
self.assertEqual(group.name, "Updated Group Test")
- response = self.client.get(reverse("admin:group_delete", args=[1]))
+ response = self.client.get(reverse("admin:group_delete", args=[group_id]))
self.assertEqual(response.status_code, 200)
- response = self.client.post(reverse("admin:group_delete", args=[1]))
+ response = self.client.post(reverse("admin:group_delete", args=[group_id]))
self.assertRedirects(response, reverse("admin:group_list"))
with self.assertRaises(ObjectDoesNotExist):
- Group.objects.get(id=1)
+ Group.objects.get(id=group_id)
def test_user_list(self):
response = self.client.get(reverse("admin:user_list"))
@@ -68,20 +68,20 @@ class AdminTestCase(TestCase):
self.assertRedirects(response, reverse("admin:user_list"))
user = User.objects.get(username="test")
- self.assertEqual(user.id, 2)
+ user_id = user.id
- ua: UserAttributes = UserAttributes.objects.get(id=2)
- self.assertEqual(ua.user_id, 2)
+ ua: UserAttributes = UserAttributes.objects.get(user_id=user_id)
+ self.assertEqual(ua.user_id, user_id)
self.assertEqual(ua.max_instances, 1)
self.assertEqual(ua.max_cpus, 1)
self.assertEqual(ua.max_memory, 1024)
self.assertEqual(ua.max_disk_size, 4)
- response = self.client.get(reverse("admin:user_update", args=[2]))
+ response = self.client.get(reverse("admin:user_update", args=[user_id]))
self.assertEqual(response.status_code, 200)
response = self.client.post(
- reverse("admin:user_update", args=[2]),
+ reverse("admin:user_update", args=[user_id]),
{
"username": "utest",
"max_instances": 2,
@@ -92,32 +92,32 @@ class AdminTestCase(TestCase):
)
self.assertRedirects(response, reverse("admin:user_list"))
- user = User.objects.get(id=2)
+ user = User.objects.get(id=user_id)
self.assertEqual(user.username, "utest")
- ua: UserAttributes = UserAttributes.objects.get(id=2)
- self.assertEqual(ua.user_id, 2)
+ ua = UserAttributes.objects.get(user_id=user_id)
+ self.assertEqual(ua.user_id, user_id)
self.assertEqual(ua.max_instances, 2)
self.assertEqual(ua.max_cpus, 2)
self.assertEqual(ua.max_memory, 2048)
self.assertEqual(ua.max_disk_size, 8)
- response = self.client.get(reverse("admin:user_block", args=[2]))
- user = User.objects.get(id=2)
+ response = self.client.get(reverse("admin:user_block", args=[user_id]))
+ user = User.objects.get(id=user_id)
self.assertFalse(user.is_active)
- response = self.client.get(reverse("admin:user_unblock", args=[2]))
- user = User.objects.get(id=2)
+ response = self.client.get(reverse("admin:user_unblock", args=[user_id]))
+ user = User.objects.get(id=user_id)
self.assertTrue(user.is_active)
- response = self.client.get(reverse("admin:user_delete", args=[2]))
+ response = self.client.get(reverse("admin:user_delete", args=[user_id]))
self.assertEqual(response.status_code, 200)
- response = self.client.post(reverse("admin:user_delete", args=[2]))
+ response = self.client.post(reverse("admin:user_delete", args=[user_id]))
self.assertRedirects(response, reverse("admin:user_list"))
with self.assertRaises(ObjectDoesNotExist):
- User.objects.get(id=2)
+ User.objects.get(id=user_id)
def test_logs(self):
response = self.client.get(reverse("admin:logs"))
diff --git a/appsettings/test_appsettings.py b/appsettings/test_appsettings.py
new file mode 100644
index 0000000..3ac82fc
--- /dev/null
+++ b/appsettings/test_appsettings.py
@@ -0,0 +1,38 @@
+from django.test import Client, TestCase
+from django.urls import reverse
+
+from appsettings.models import AppSettings
+from appsettings.settings import app_settings, get_settings
+
+
+class AppSettingsTestCase(TestCase):
+ def setUp(self):
+ self.client.login(username="admin", password="admin")
+
+ def test_get_settings_populates_singleton(self):
+ AppSettings.objects.update_or_create(
+ key="CUSTOM_TEST_SETTING",
+ defaults={"value": "test_value_123", "name": "Custom Test Setting"}
+ )
+ get_settings()
+ self.assertEqual(getattr(app_settings, "CUSTOM_TEST_SETTING", None), "test_value_123")
+
+ def test_appsettings_view_get(self):
+ response = self.client.get(reverse("appsettings"))
+ self.assertEqual(response.status_code, 200)
+
+ def test_appsettings_update_setting(self):
+ setting = AppSettings.objects.get(key="SHOW_ACCESS_ROOT_PASSWORD")
+ original_value = setting.value
+ new_val = "False" if original_value == "True" else "True"
+
+ response = self.client.post(
+ reverse("appsettings"),
+ {
+ "SHOW_ACCESS_ROOT_PASSWORD": new_val
+ }
+ )
+ self.assertRedirects(response, reverse("appsettings"))
+
+ setting.refresh_from_db()
+ self.assertEqual(setting.value, new_val)
diff --git a/datasource/test_datasource.py b/datasource/test_datasource.py
new file mode 100644
index 0000000..cbaf19f
--- /dev/null
+++ b/datasource/test_datasource.py
@@ -0,0 +1,57 @@
+import json
+from unittest.mock import MagicMock, patch
+
+from django.test import RequestFactory, TestCase
+from django.urls import reverse
+
+from computes.models import Compute
+from datasource.views import get_client_ip, get_hostname_by_ip, os_index, os_metadata_json, os_userdata
+
+
+class DataSourceTestCase(TestCase):
+ def setUp(self):
+ self.factory = RequestFactory()
+ self.client.login(username="admin", password="admin")
+ self.compute = Compute.objects.create(
+ name="test_compute",
+ hostname="localhost",
+ type=1,
+ )
+
+ def test_os_index(self):
+ response = self.client.get(reverse("ds_openstack_index"))
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("latest", response.content.decode("utf-8"))
+
+ def test_os_metadata_json_latest(self):
+ response = self.client.get(reverse("ds_openstack_metadata", args=["latest"]))
+ self.assertEqual(response.status_code, 200)
+ data = json.loads(response.content.decode("utf-8"))
+ self.assertEqual(data["uuid"], "iid-dswebvirtcloud")
+ self.assertIn("hostname", data)
+
+ def test_os_metadata_json_invalid_version(self):
+ response = self.client.get(reverse("ds_openstack_metadata", args=["v1"]))
+ self.assertEqual(response.status_code, 404)
+
+ def test_os_userdata_invalid_version(self):
+ response = self.client.get(reverse("ds_openstack_userdata", args=["v1"]))
+ self.assertEqual(response.status_code, 404)
+
+ def test_get_client_ip_direct(self):
+ request = self.factory.get("/")
+ request.META["REMOTE_ADDR"] = "192.168.1.10"
+ self.assertEqual(get_client_ip(request), "192.168.1.10")
+
+ def test_get_client_ip_forwarded(self):
+ request = self.factory.get("/")
+ request.META["HTTP_X_FORWARDED_FOR"] = "10.0.0.1, 192.168.1.50"
+ self.assertEqual(get_client_ip(request), "192.168.1.50")
+
+ def test_get_hostname_by_ip_fallback(self):
+ # Invalid IP will cause gethostbyaddr to fail and return the IP string itself
+ self.assertEqual(get_hostname_by_ip("256.256.256.256"), "256.256.256.256")
+
+ def test_vdi_url_not_found_compute(self):
+ response = self.client.get(reverse("vdi_url", args=[9999, "vm1"]))
+ self.assertEqual(response.status_code, 404)
diff --git a/instances/test_root_pass.py b/instances/test_root_pass.py
new file mode 100644
index 0000000..9003feb
--- /dev/null
+++ b/instances/test_root_pass.py
@@ -0,0 +1,55 @@
+import json
+from unittest.mock import MagicMock, patch
+
+from django.test import TestCase
+from django.urls import reverse
+
+from computes.models import Compute
+from instances.models import Instance
+
+
+class SetRootPassTestCase(TestCase):
+ def setUp(self):
+ self.client.login(username="admin", password="admin")
+ self.compute = Compute.objects.create(
+ name="test_compute_root_pass",
+ hostname="localhost",
+ type=1,
+ )
+ self.instance = Instance.objects.create(
+ compute=self.compute,
+ name="root-pass-vm",
+ uuid="87654321-4321-4321-4321-210987654321",
+ )
+
+ @patch("socket.socket")
+ @patch.object(Instance, "proxy")
+ def test_set_root_pass_post(self, mock_proxy, mock_socket_cls):
+ mock_proxy.get_status.return_value = 5 # status: running
+ mock_sock = MagicMock()
+ mock_socket_cls.return_value = mock_sock
+ mock_sock.recv.return_value = json.dumps({"return": "success"}).encode()
+
+ response = self.client.post(
+ reverse("instances:rootpasswd", args=[self.instance.id]),
+ {"passwd": "newSecretPassword123"}
+ )
+ # Redirects back to referer or instance detail
+ self.assertEqual(response.status_code, 302)
+
+ # Check socket sent the command with hashed password
+ self.assertTrue(mock_sock.send.called)
+ sent_bytes = mock_sock.send.call_args[0][0]
+ data = json.loads(sent_bytes.decode())
+ self.assertEqual(data["action"], "password")
+ self.assertEqual(data["vname"], "root-pass-vm")
+ # SHA-512 crypt starts with $6$kgPoiREy$
+ self.assertTrue(data["passwd"].startswith("$6$kgPoiREy$"))
+
+ def test_set_root_pass_get_redirects(self):
+ response = self.client.get(reverse("instances:rootpasswd", args=[self.instance.id]))
+ self.assertEqual(response.status_code, 302)
+
+ def test_set_root_pass_not_found(self):
+ response = self.client.get(reverse("instances:rootpasswd", args=[99999]))
+ self.assertEqual(response.status_code, 404)
diff --git a/logs/test_logs.py b/logs/test_logs.py
new file mode 100644
index 0000000..023a67c
--- /dev/null
+++ b/logs/test_logs.py
@@ -0,0 +1,61 @@
+import json
+from django.contrib.auth import get_user_model
+from django.test import Client, TestCase
+from django.urls import reverse
+
+from computes.models import Compute
+from instances.models import Instance
+from logs.models import Logs
+from logs.views import addlogmsg
+
+
+class LogsTestCase(TestCase):
+ def setUp(self):
+ User = get_user_model()
+ self.superuser = User.objects.create_superuser(
+ username="admin_logs", email="admin@example.com", password="adminpassword"
+ )
+ self.normal_user = User.objects.create_user(
+ username="normal_logs", email="user@example.com", password="userpassword"
+ )
+ self.compute = Compute.objects.create(
+ name="local_compute",
+ hostname="localhost",
+ type=1,
+ )
+ self.instance = Instance.objects.create(
+ compute=self.compute,
+ name="test-vm",
+ uuid="12345678-1234-1234-1234-123456789012",
+ )
+
+ def test_addlogmsg(self):
+ addlogmsg("admin", "localhost", "test-vm", "Started VM")
+ log = Logs.objects.filter(instance="test-vm").first()
+ self.assertIsNotNone(log)
+ self.assertEqual(log.user, "admin")
+ self.assertEqual(log.host, "localhost")
+ self.assertEqual(log.message, "Started VM")
+
+ def test_vm_logs_superuser_access(self):
+ addlogmsg("admin", "localhost", "test-vm", "Created VM")
+ client = Client()
+ client.login(username="admin_logs", password="adminpassword")
+
+ response = client.get(reverse("vm_logs", args=["test-vm"]))
+ self.assertEqual(response.status_code, 200)
+
+ data = json.loads(response.content.decode("utf-8"))
+ self.assertIsInstance(data, list)
+ self.assertEqual(len(data), 1)
+ self.assertEqual(data[0]["user"], "admin")
+ self.assertEqual(data[0]["message"], "Created VM")
+ self.assertEqual(data[0]["instance"], "test-vm")
+
+ def test_vm_logs_non_superuser_forbidden(self):
+ client = Client()
+ client.login(username="normal_logs", password="userpassword")
+
+ response = client.get(reverse("vm_logs", args=["test-vm"]))
+ # superuser_only decorator redirects non-superusers to index or login
+ self.assertNotEqual(response.status_code, 200)
diff --git a/vrtManager/test_network_ip.py b/vrtManager/test_network_ip.py
index 1a33eb0..61b3d66 100644
--- a/vrtManager/test_network_ip.py
+++ b/vrtManager/test_network_ip.py
@@ -2,11 +2,15 @@ import sys
import unittest
from unittest.mock import MagicMock
-# Mock django if not installed so tests can run in minimal/standalone environments
-if "django" not in sys.modules:
+try:
+ from django.conf import settings
+ if not settings.configured:
+ settings.configure(MAC_OUI="52:54:10")
+except ImportError:
django_mock = MagicMock()
+ django_mock.conf.settings.MAC_OUI = "52:54:10"
sys.modules["django"] = django_mock
- sys.modules["django.conf"] = django_mock
+ sys.modules["django.conf"] = django_mock.conf
from vrtManager.network import network_size, wvmNetwork
diff --git a/vrtManager/test_util.py b/vrtManager/test_util.py
new file mode 100644
index 0000000..dced762
--- /dev/null
+++ b/vrtManager/test_util.py
@@ -0,0 +1,133 @@
+import re
+import unittest
+
+from django.conf import settings
+
+if not settings.configured:
+ settings.configure(MAC_OUI="52:54:10")
+
+from vrtManager.util import (
+ compareMAC,
+ get_xml_path,
+ is_kvm_available,
+ pretty_bytes,
+ pretty_mem,
+ randomMAC,
+ randomPasswd,
+ randomUUID,
+ validate_macaddr,
+ validate_uuid,
+ vol_dev_type,
+ xml_escape,
+)
+
+
+class TestVrtManagerUtil(unittest.TestCase):
+ def test_random_uuid_format(self):
+ uuid_str = randomUUID()
+ # UUIDv4 format check
+ pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
+ self.assertRegex(uuid_str, pattern)
+ # Ensure successive UUIDs are distinct
+ self.assertNotEqual(uuid_str, randomUUID())
+
+ def test_validate_uuid(self):
+ valid_dashed = "c9a646d3-9c61-4cd8-8024-f6b72a6642d9"
+ self.assertEqual(validate_uuid(valid_dashed), valid_dashed)
+
+ valid_no_dash = "c9a646d39c614cd88024f6b72a6642d9"
+ self.assertEqual(validate_uuid(valid_no_dash), valid_dashed)
+
+ with self.assertRaises(ValueError):
+ validate_uuid("invalid-uuid-too-short")
+
+ with self.assertRaises(ValueError):
+ validate_uuid(12345)
+
+ def test_random_mac(self):
+ mac = randomMAC()
+ self.assertTrue(mac.startswith("52:54:10:"))
+ parts = mac.split(":")
+ self.assertEqual(len(parts), 6)
+ for part in parts:
+ self.assertEqual(len(part), 2)
+ int(part, 16) # valid hex
+
+ def test_compare_mac(self):
+ mac1 = "52:54:10:00:00:01"
+ mac2 = "52:54:10:00:00:02"
+ self.assertEqual(compareMAC(mac1, mac1), 0)
+ self.assertEqual(compareMAC(mac1, mac2), -1)
+ self.assertEqual(compareMAC(mac2, mac1), 1)
+ # Different lengths
+ self.assertEqual(compareMAC("52:54", "52:54:10"), -1)
+ self.assertEqual(compareMAC("52:54:10", "52:54"), 1)
+
+ def test_validate_macaddr(self):
+ self.assertIsNone(validate_macaddr(None))
+ self.assertIsNone(validate_macaddr("52:54:10:aa:bb:cc"))
+ self.assertIsNone(validate_macaddr("0:1:2:3:4:5"))
+
+ with self.assertRaises(ValueError):
+ validate_macaddr("invalid_mac")
+
+ with self.assertRaises(ValueError):
+ validate_macaddr(12345)
+
+ def test_random_passwd(self):
+ pwd = randomPasswd(16)
+ self.assertEqual(len(pwd), 16)
+ # Custom alphabet
+ numeric_pwd = randomPasswd(10, alphabet="0123456789")
+ self.assertEqual(len(numeric_pwd), 10)
+ self.assertTrue(numeric_pwd.isdigit())
+
+ def test_xml_escape(self):
+ self.assertIsNone(xml_escape(None))
+ escaped = xml_escape('')
+ self.assertEqual(escaped, "<foo name="bar" test='baz' &/>")
+
+ def test_get_xml_path(self):
+ xml = "testvm1048576"
+ self.assertEqual(get_xml_path(xml, "/domain/name"), "testvm")
+ self.assertEqual(get_xml_path(xml, "/domain/memory"), "1048576")
+ self.assertEqual(get_xml_path(xml, "/domain/@type"), "kvm")
+
+ # Function callback mode
+ def extract_name(doc):
+ return doc.xpath("/domain/name")[0].text
+
+ self.assertEqual(get_xml_path(xml, func=extract_name), "testvm")
+
+ # Missing both path and func
+ with self.assertRaises(ValueError):
+ get_xml_path(xml)
+
+ def test_is_kvm_available(self):
+ kvm_xml = ""
+ qemu_xml = ""
+ self.assertTrue(is_kvm_available(kvm_xml))
+ self.assertFalse(is_kvm_available(qemu_xml))
+
+ def test_pretty_mem(self):
+ # Value in KB
+ self.assertEqual(pretty_mem(1024), " 1 MB")
+ self.assertEqual(pretty_mem(1024 * 1024), "1024 MB")
+ self.assertEqual(pretty_mem(16 * 1024 * 1024), "16.00 GB")
+
+ def test_pretty_bytes(self):
+ # Value in Bytes
+ self.assertEqual(pretty_bytes(10 * 1024 * 1024), "10.00 MB")
+ self.assertEqual(pretty_bytes(2 * 1024 * 1024 * 1024), "2.00 GB")
+
+ def test_vol_dev_type(self):
+ self.assertEqual(vol_dev_type("ide"), "hd")
+ self.assertEqual(vol_dev_type("fdc"), "fd")
+ self.assertEqual(vol_dev_type("virtio"), "vd")
+ self.assertEqual(vol_dev_type("scsi"), "sd")
+ self.assertEqual(vol_dev_type("sata"), "sd")
+ self.assertIsNone(vol_dev_type("unknown_bus"))
+
+
+if __name__ == "__main__":
+ unittest.main()