From c31ed280422f2639dcf69aebaf7d40d779d02ad1 Mon Sep 17 00:00:00 2001 From: catborise Date: Fri, 25 Sep 2026 11:26:53 +0300 Subject: [PATCH] test: expand unit test coverage across admin, accounts, vrtManager, logs, appsettings, and datasource --- accounts/test_utils.py | 69 +++++++++++++++++ admin/tests.py | 44 +++++------ appsettings/test_appsettings.py | 38 +++++++++ datasource/test_datasource.py | 57 ++++++++++++++ instances/test_root_pass.py | 55 +++++++++++++ logs/test_logs.py | 61 +++++++++++++++ vrtManager/test_network_ip.py | 10 ++- vrtManager/test_util.py | 133 ++++++++++++++++++++++++++++++++ 8 files changed, 442 insertions(+), 25 deletions(-) create mode 100644 accounts/test_utils.py create mode 100644 appsettings/test_appsettings.py create mode 100644 datasource/test_datasource.py create mode 100644 instances/test_root_pass.py create mode 100644 logs/test_logs.py create mode 100644 vrtManager/test_util.py 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()