# pylint: disable=no-member import importlib.machinery import importlib.util from pathlib import Path from unittest.mock import MagicMock, patch from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.test import TestCase, override_settings from django.urls import reverse from libvirt import libvirtError from accounts.models import UserInstance from computes.models import Compute from instances.models import Instance # Dynamically load console/novncd script for testing NOVNCD_PATH = str(Path(__file__).resolve().parent / "novncd") _loader = importlib.machinery.SourceFileLoader("novncd_mod", NOVNCD_PATH) _spec = importlib.util.spec_from_file_location( "novncd_mod", NOVNCD_PATH, loader=_loader ) novncd_mod = importlib.util.module_from_spec(_spec) _loader.exec_module(novncd_mod) class ConsoleViewsTestCase(TestCase): @classmethod def setUpTestData(cls): User = get_user_model() cls.admin_user = ( User.objects.filter(is_superuser=True).first() or User.objects.filter(username="admin").first() or User.objects.create_superuser( username="admin_console_test", email="admin@example.com", password="adminpassword", ) ) cls.alice = User.objects.create_user( username="alice", email="alice@example.com", password="alicepassword" ) cls.bob = User.objects.create_user( username="bob", email="bob@example.com", password="bobpassword" ) cls.compute = Compute.objects.create( name="test-compute", hostname="127.0.0.1", login="root", password="", type=1, ) cls.instance = Instance.objects.create( compute=cls.compute, name="vm-alpha", uuid="11111111-2222-3333-4444-555555555555", ) # Grant alice ownership of vm-alpha UserInstance.objects.create( user=cls.alice, instance=cls.instance, ) cls.token = f"{cls.compute.id}-{cls.instance.uuid}" @patch("console.views.wvmInstance") def test_console_superuser_vnc_lite(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_conn.get_console_websocket_port.return_value = None mock_conn.get_console_passwd.return_value = "vncsecret" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}&view=lite" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-vnc-lite.html") self.assertEqual(response.cookies["token"].value, self.token) self.assertEqual(response.context["ws_path"], "novncd/") self.assertEqual(response.context["ws_port"], 6080) self.assertEqual(response.context["console_passwd"], "vncsecret") @patch("console.views.wvmInstance") def test_console_superuser_vnc_full(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_conn.get_console_websocket_port.return_value = 5900 mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}&view=full" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-vnc-full.html") self.assertEqual(response.context["ws_port"], 5900) @patch("console.views.wvmInstance") def test_console_superuser_spice_lite(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "spice" mock_conn.get_console_websocket_port.return_value = None mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}&view=lite" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-spice-lite.html") @patch("console.views.wvmInstance") def test_console_superuser_spice_full(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "spice" mock_conn.get_console_websocket_port.return_value = None mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}&view=full" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-spice-full.html") @patch("console.views.wvmInstance") def test_console_superuser_pty(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "pty" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-xterm.html") @patch("console.views.wvmInstance") def test_console_user_assigned_instance_access(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_conn.get_console_websocket_port.return_value = None mock_wvm.return_value = mock_conn self.client.force_login(self.alice) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-vnc-lite.html") def test_console_unauthorized_user_forbidden(self): self.client.force_login(self.bob) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) # Non-permitted user attempting to view other user's instance self.assertEqual(response.status_code, 500) self.assertIn("permission", response.content.decode("utf-8").lower()) @patch("console.views.wvmInstance") def test_console_user_with_view_instances_perm(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_conn.get_console_websocket_port.return_value = None mock_wvm.return_value = mock_conn perm = Permission.objects.get(codename="view_instances") self.bob.user_permissions.add(perm) self.client.force_login(self.bob) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) @patch("console.views.wvmInstance") @override_settings(WS_PUBLIC_PATH="/novncd/") def test_console_ws_path_normalization_leading_slash(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["ws_path"], "novncd/") @patch("console.views.wvmInstance") @override_settings(WS_PUBLIC_PATH="novncd/") def test_console_ws_path_normalization_clean(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["ws_path"], "novncd/") @patch("console.views.wvmInstance") @override_settings(WS_PUBLIC_PATH="/") def test_console_ws_path_root_slash(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertEqual(response.context["ws_path"], "") @patch("console.views.wvmInstance") def test_console_ws_host_strip_port(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url, HTTP_HOST="panel.example.com:8080") self.assertEqual(response.status_code, 200) self.assertEqual(response.context["ws_host"], "panel.example.com") @patch("console.views.wvmInstance") def test_console_libvirt_error_fallback(self, mock_wvm): mock_wvm.side_effect = libvirtError("Connection to libvirtd failed") self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}" response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-vnc-lite.html") self.assertIsNotNone(response.context["console_error"]) def test_console_missing_token_safe_fallback(self): self.client.force_login(self.admin_user) url = reverse("console") response = self.client.get(url) self.assertEqual(response.status_code, 200) self.assertTemplateUsed(response, "console-vnc-lite.html") self.assertIsNotNone(response.context["console_error"]) @patch("console.views.wvmInstance") def test_console_send_keys_menu_elements(self, mock_wvm): mock_conn = MagicMock() mock_conn.get_console_type.return_value = "vnc" mock_wvm.return_value = mock_conn self.client.force_login(self.admin_user) url = reverse("console") + f"?token={self.token}&view=lite" response = self.client.get(url) self.assertEqual(response.status_code, 200) content = response.content.decode("utf-8") expected_ids = [ "ctrlaltdel", "ctrlaltbackspace", "alttab", "superkey", "sendesc", "sendtab", "ctrlc", "ctrlz", "ctrld", "ctrlbackslash", ] + [f"ctrlaltf{i}" for i in range(1, 13)] for key_id in expected_ids: self.assertIn( f'id="{key_id}"', content, f"Expected key element id '{key_id}' in console menu", ) class NovncdDaemonLogicTestCase(TestCase): def setUp(self): self.get_parser = novncd_mod.get_parser self.CompatibilityMixIn = novncd_mod.CompatibilityMixIn def test_novncd_parser_defaults(self): parser = self.get_parser() opts = parser.parse_args([]) self.assertFalse(opts.verbose) self.assertFalse(opts.debug) self.assertEqual(opts.port, 6080) def test_novncd_parser_custom_arguments(self): parser = self.get_parser() opts = parser.parse_args( ["-v", "-d", "-H", "192.168.1.100", "-p", "7070", "-c", "/path/to/cert.pem"] ) self.assertTrue(opts.verbose) self.assertTrue(opts.debug) self.assertEqual(opts.host, "192.168.1.100") self.assertEqual(opts.port, 7070) self.assertEqual(opts.cert, "/path/to/cert.pem") @patch.object(novncd_mod, "get_connection_infos") def test_compatibility_mixin_token_from_cookie(self, mock_get_info): mock_get_info.return_value = ( "host1", 22, "root", 1, "192.168.1.50", 5900, None, ) handler = self.CompatibilityMixIn() handler.headers = {"cookie": "other=123; token=1-uuid-from-cookie; test=456"} handler.path = "/novncd/" handler.msg = MagicMock() handler.do_proxy = MagicMock() handler.verbose = False socket_factory = MagicMock() handler._new_client(daemon=False, socket_factory=socket_factory) mock_get_info.assert_called_once_with("1-uuid-from-cookie") socket_factory.assert_called_once_with("host1", 5900, connect=True) @patch.object(novncd_mod, "get_connection_infos") def test_compatibility_mixin_token_from_query_params(self, mock_get_info): mock_get_info.return_value = ( "host2", 22, "root", 1, "192.168.1.50", 5901, None, ) handler = self.CompatibilityMixIn() handler.headers = {} handler.path = "/novncd/?token=1-uuid-from-query" handler.msg = MagicMock() handler.do_proxy = MagicMock() handler.verbose = False socket_factory = MagicMock() handler._new_client(daemon=False, socket_factory=socket_factory) mock_get_info.assert_called_once_with("1-uuid-from-query") socket_factory.assert_called_once_with("host2", 5901, connect=True) @patch.object(novncd_mod, "get_connection_infos") def test_compatibility_mixin_missing_token_safe_exit(self, mock_get_info): handler = self.CompatibilityMixIn() handler.headers = {} handler.path = "/novncd/" handler.msg = MagicMock() socket_factory = MagicMock() # Should cleanly exit and log message without raising UnboundLocalError handler._new_client(daemon=False, socket_factory=socket_factory) mock_get_info.assert_not_called() socket_factory.assert_not_called() handler.msg.assert_called_with( "No console token provided in cookie or query parameters" )