test: expand unit test coverage across admin, accounts, vrtManager, logs, appsettings, and datasource

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"])

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)
+        ua: UserAttributes = UserAttributes.objects.get(user_id=user_id)
-        self.assertEqual(ua.user_id, 2)
+        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)
+        ua = UserAttributes.objects.get(user_id=user_id)
-        self.assertEqual(ua.user_id, 2)
+        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]))
+        response = self.client.get(reverse("admin:user_block", args=[user_id]))
-        user = User.objects.get(id=2)
+        user = User.objects.get(id=user_id)
         self.assertFalse(user.is_active)
 
-        response = self.client.get(reverse("admin:user_unblock", args=[2]))
+        response = self.client.get(reverse("admin:user_unblock", args=[user_id]))
-        user = User.objects.get(id=2)
+        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"))

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)

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)

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)

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)

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
+try:
-if "django" not in sys.modules:
+    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
 

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('<foo name="bar" test=\'baz\' &/>')
+        self.assertEqual(escaped, "&lt;foo name=&quot;bar&quot; test=&apos;baz&apos; &amp;/&gt;")
+
+    def test_get_xml_path(self):
+        xml = "<domain type='kvm'><name>testvm</name><memory unit='KiB'>1048576</memory></domain>"
+        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 = "<capabilities><guest><arch name='x86_64'><domain type='kvm'/></arch></guest></capabilities>"
+        qemu_xml = "<capabilities><guest><arch name='x86_64'><domain type='qemu'/></arch></guest></capabilities>"
+        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()