diff --git a/accounts/tests/test_security_fixes.py b/accounts/tests/test_security_fixes.py deleted file mode 100644 index 1c77d3a4..00000000 --- a/accounts/tests/test_security_fixes.py +++ /dev/null @@ -1,64 +0,0 @@ -from django.contrib.auth import get_user_model -from django.test import TestCase -from django.urls import reverse - -from accounts.models import UserInstance, UserSSHKey -from accounts.utils import validate_ssh_key -from computes.models import Compute -from instances.models import Instance - - -class AccountsSecurityFixesTestCase(TestCase): - def setUp(self): - User = get_user_model() - self.admin = User.objects.create_superuser( - username="sec_acc_admin", email="admin@example.com", password="password" - ) - self.user = User.objects.create_user( - username="sec_acc_user", password="password" - ) - self.compute = Compute.objects.create( - name="acc-comp", hostname="127.0.0.1", login="root", type=1 - ) - self.instance = Instance.objects.create( - compute=self.compute, name="acc-vm", uuid="99999999-9999-9999-9999-999999999999" - ) - self.user_inst = UserInstance.objects.create( - instance=self.instance, user=self.user - ) - - def test_open_redirect_prevented_in_user_instance_delete(self): - self.client.force_login(self.admin) - url = reverse("accounts:user_instance_delete", args=[self.user_inst.id]) - - # Attempt external malicious redirect - res = self.client.post(f"{url}?next=https://malicious-attacker.com") - self.assertEqual(res.status_code, 302) - # Should redirect to safe internal URL (user account page), NOT external site - self.assertNotIn("malicious-attacker.com", res.url) - self.assertEqual(res.url, reverse("accounts:account", args=[self.user.id])) - - def test_ssh_key_cascade_deletion(self): - key = UserSSHKey.objects.create( - user=self.user, - keyname="my-key", - keypublic="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGabcdef123456 test@example.com", - ) - self.assertTrue(UserSSHKey.objects.filter(id=key.id).exists()) - - # When user is deleted, key should be deleted via CASCADE (not left orphan) - self.user.delete() - self.assertFalse(UserSSHKey.objects.filter(id=key.id).exists()) - - def test_validate_ssh_key_supports_keys_with_and_without_comment(self): - rsa_base = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw==" - # Key with comment - key_with_comment = f"{rsa_base} test@test" - self.assertTrue(validate_ssh_key(key_with_comment)) - - # Key without comment - key_without_comment = rsa_base - self.assertTrue(validate_ssh_key(key_without_comment)) - - # Invalid key - self.assertFalse(validate_ssh_key("not a valid key string")) diff --git a/accounts/tests/test_ssh_keys.py b/accounts/tests/test_ssh_keys.py new file mode 100644 index 00000000..02cd9495 --- /dev/null +++ b/accounts/tests/test_ssh_keys.py @@ -0,0 +1,35 @@ +"""A user's SSH keys: they go with the user, and a key is valid with or without a comment.""" + +from django.contrib.auth import get_user_model +from django.test import TestCase + +from accounts.models import UserSSHKey +from accounts.utils import validate_ssh_key + + +class SshKeyTestCase(TestCase): + def test_ssh_key_cascade_deletion(self): + user = get_user_model().objects.create_user(username="sec_acc_user", password="password") + key = UserSSHKey.objects.create( + user=user, + keyname="my-key", + keypublic="ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGabcdef123456 test@example.com", + ) + self.assertTrue(UserSSHKey.objects.filter(id=key.id).exists()) + + # When user is deleted, key should be deleted via CASCADE (not left orphan) + user.delete() + self.assertFalse(UserSSHKey.objects.filter(id=key.id).exists()) + + def test_validate_ssh_key_supports_keys_with_and_without_comment(self): + rsa_base = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAgQC6OOdbfv27QVnSC6sKxGaHb6YFc+3gxCkyVR3cTSXE/n5BEGf8aOgBpepULWa1RZfxYHY14PlKULDygdXSdrrR2kNSwoKz/Oo4d+3EE92L7ocl1+djZbptzgWgtw1OseLwbFik+iKlIdqPsH+IUQvX7yV545ZQtAP8Qj1R+uCqkw==" + # Key with comment + key_with_comment = f"{rsa_base} test@test" + self.assertTrue(validate_ssh_key(key_with_comment)) + + # Key without comment + key_without_comment = rsa_base + self.assertTrue(validate_ssh_key(key_without_comment)) + + # Invalid key + self.assertFalse(validate_ssh_key("not a valid key string")) diff --git a/accounts/tests/test_views.py b/accounts/tests/test_views.py index 29d7b0d7..11795dd1 100644 --- a/accounts/tests/test_views.py +++ b/accounts/tests/test_views.py @@ -335,3 +335,22 @@ class OwnerRightsReturnTestCase(TestCase): with self.subTest(next_url): response = self.client.post(f"{update}?next={next_url}", {"is_change": True}) self.assertRedirects(response, back, fetch_redirect_response=False) + + +class UserInstanceDeleteTestCase(TestCase): + def test_open_redirect_prevented_in_user_instance_delete(self): + User = get_user_model() + admin = User.objects.create_superuser(username="sec_acc_admin", email="admin@example.com", password="password") + user = User.objects.create_user(username="sec_acc_user", password="password") + compute = Compute.objects.create(name="acc-comp", hostname="127.0.0.1", login="root", type=1) + instance = Instance.objects.create(compute=compute, name="acc-vm", uuid="99999999-9999-9999-9999-999999999999") + user_inst = UserInstance.objects.create(instance=instance, user=user) + self.client.force_login(admin) + url = reverse("accounts:user_instance_delete", args=[user_inst.id]) + + # Attempt external malicious redirect + res = self.client.post(f"{url}?next=https://malicious-attacker.com") + self.assertEqual(res.status_code, 302) + # Should redirect to safe internal URL (user account page), NOT external site + self.assertNotIn("malicious-attacker.com", res.url) + self.assertEqual(res.url, reverse("accounts:account", args=[user.id])) diff --git a/instances/tests/test_console_rule.py b/instances/tests/test_console_access.py similarity index 100% rename from instances/tests/test_console_rule.py rename to instances/tests/test_console_access.py diff --git a/instances/tests/test_console_listener.py b/instances/tests/test_console_listener.py deleted file mode 100644 index 5ba83ca1..00000000 --- a/instances/tests/test_console_listener.py +++ /dev/null @@ -1,84 +0,0 @@ -"""Only administrators choose where a VM's console listens. Other users who may -manage the console cannot make it listen beyond localhost, and cannot remove -the password of a console that does; such a refusal saves nothing.""" - -from unittest.mock import PropertyMock, patch - -from accounts.models import UserInstance -from computes.models import Compute -from django.contrib.auth import get_user_model -from django.contrib.messages import get_messages -from django.test import TestCase, override_settings -from django.urls import reverse -from instances.models import Instance - - -class ConsoleListenerTests(TestCase): - def setUp(self): - User = get_user_model() - self.owner = User.objects.create_user("console-owner", password="pw") - self.admin = User.objects.create_superuser("console-admin", password="pw", email="ca@example.com") - compute = Compute.objects.create(name="cl", hostname="10.0.0.1", login="u", password="p", type=1) - self.vm = Instance.objects.create(compute=compute, name="cl-vm", uuid="78787878-0000-0000-0000-000000000000") - UserInstance.objects.create(user=self.owner, instance=self.vm, is_change=True, is_vnc=True) - - def post(self, user, data, local=True): - self.client.force_login(user) - with patch("instances.models.wvmInstance") as wvm: - proxy = wvm.return_value - proxy.console_listens_locally.return_value = local - proxy.get_console_listener_addr.return_value = "127.0.0.1" if local else "0.0.0.0" - proxy.get_console_passwd.return_value = "Old-pass-1" - proxy.get_console_keymap.return_value = "" - proxy.set_console_passwd.return_value = True - response = self.client.post(reverse("instances:update_console", args=[self.vm.id]), data) - return proxy, [str(m) for m in get_messages(response.wsgi_request)] - - def test_owner_cannot_change_the_listen_address(self): - proxy, _ = self.post(self.owner, {"listen_on": "0.0.0.0", "keymap": "auto", "password_action": "keep"}) - proxy.set_console_listener_addr.assert_not_called() - - def test_owner_cannot_remove_the_password_of_a_console_listening_beyond_localhost(self): - proxy, messages = self.post(self.owner, {"password_action": "none", "keymap": "de"}, local=False) - proxy.set_console_passwd.assert_not_called() - proxy.set_console_keymap.assert_not_called() - self.assertTrue(any("password" in m.lower() for m in messages), messages) - - @override_settings(SERIAL_CONSOLE_ENABLED=True) - def test_owner_can_switch_a_console_listening_beyond_localhost_to_serial(self): - proxy, messages = self.post(self.owner, {"console_type": "pty", "password_action": "none", "keymap": "auto"}, - local=False) - self.assertEqual(messages, []) - proxy.set_console_type.assert_called_once_with("pty") - proxy.set_console_passwd.assert_not_called() - - def test_owner_can_remove_the_password_of_a_local_console(self): - proxy, _ = self.post(self.owner, {"password_action": "none", "keymap": "auto"}) - proxy.set_console_passwd.assert_called_once_with("") - - def test_owner_can_set_a_password_of_any_console(self): - proxy, _ = self.post(self.owner, {"password_action": "set", "password": "s3cret-1", "keymap": "auto"}, local=False) - proxy.set_console_passwd.assert_called_once_with("s3cret-1") - - def test_superuser_can_open_the_console_to_all_interfaces(self): - proxy, _ = self.post(self.admin, {"listen_on": "0.0.0.0", "password_action": "none", "keymap": "auto"}) - proxy.set_console_listener_addr.assert_called_once_with("0.0.0.0") - proxy.set_console_passwd.assert_called_once_with("") - - def test_owner_sees_the_listen_address_but_cannot_edit_it(self): - self.client.force_login(self.owner) - with patch("instances.models.wvmInstance") as wvm, patch.object( - Compute, "status", new_callable=PropertyMock, return_value=True - ): - proxy = wvm.return_value - proxy.get_memory.return_value = 1024 - proxy.get_cur_memory.return_value = 1024 - proxy.get_max_memory.return_value = 4096 - proxy.get_console_listener_addr.return_value = "127.0.0.1" - for name in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices", "get_media_devices", "get_net_devices"): - getattr(proxy, name).return_value = [] - proxy.get_hostdevs.return_value = ([], [], []) - proxy.get_status.return_value = 5 - response = self.client.get(reverse("instances:instance", args=[self.vm.id])) - form = response.context["console_form"] - self.assertTrue(form.fields["listen_on"].disabled) diff --git a/instances/tests/test_console_settings.py b/instances/tests/test_console_settings.py new file mode 100644 index 00000000..98c8f7f5 --- /dev/null +++ b/instances/tests/test_console_settings.py @@ -0,0 +1,234 @@ +"""A VM's console settings: who sees the form; each change is saved to the +persistent definition, the password only by the action chosen for it, and +the VNC settings only while the saved console type is VNC; only +administrators choose where a console listens beyond localhost.""" + +from unittest.mock import patch + +from computes.models import Compute +from django.contrib.auth import get_user_model +from django.contrib.messages import get_messages +from django.test import TestCase, override_settings +from django.urls import reverse +from instances.forms import ConsoleForm +from instances.models import Instance +from unittest.mock import PropertyMock +from accounts.models import UserInstance +from django.contrib.auth.models import Permission + +VNC = {"console_type": "vnc", "listen_on": "127.0.0.1", "keymap": "auto", "password_action": "keep"} + + +class UpdateConsoleTestCase(TestCase): + def setUp(self): + admin = get_user_model().objects.create_superuser("console_admin", "c@example.com", "pw") + self.client.force_login(admin) + compute = Compute.objects.create(name="console-compute", hostname="127.0.0.1", login="", password="", type=4) + self.instance = Instance.objects.create( + compute=compute, name="console-vm", uuid="11111111-2222-3333-4444-555555555555" + ) + + def post(self, data, console_type="vnc"): + """console_type: the persistent definition's; the live one is VNC.""" + with patch("instances.models.wvmInstance") as wvm: + proxy = wvm.return_value + proxy.get_console_type.side_effect = lambda config=False: console_type if config else "vnc" + proxy.get_console_listener_addr.return_value = "127.0.0.1" + proxy.get_console_keymap.return_value = "" + proxy.set_console_passwd.return_value = True + response = self.client.post(reverse("instances:update_console", args=[self.instance.id]), data) + return proxy, [str(m) for m in get_messages(response.wsgi_request)] + + def test_console_settings_are_applied(self): + proxy, _ = self.post({**VNC, "listen_on": "0.0.0.0", "keymap": "de"}) + proxy.set_console_keymap.assert_called_once_with("de") + proxy.set_console_listener_addr.assert_called_once_with("0.0.0.0") + proxy.set_console_passwd.assert_not_called() + + def test_the_password_changes_by_the_action_chosen(self): + for action, password in (("set", " a b c "), ("set", "éééé"), ("none", "")): + with self.subTest(action, password=password): + proxy, _ = self.post({**VNC, "password_action": action, "password": password}) + proxy.set_console_passwd.assert_called_once_with(password) + proxy, _ = self.post({**VNC, "password_action": "generate"}) + (generated,), _ = proxy.set_console_passwd.call_args + self.assertEqual(len(generated), 8) + + def test_a_password_vnc_cannot_take_is_refused_and_nothing_is_saved(self): + for password in ("", "123456789", "ééééé"): # libvirt counts bytes: é is two + with self.subTest(password=password): + proxy, messages = self.post({**VNC, "keymap": "de", "password_action": "set", "password": password}) + proxy.set_console_passwd.assert_not_called() + proxy.set_console_keymap.assert_not_called() + self.assertTrue(messages[0].startswith("Console settings were not saved"), messages) + + @override_settings(SERIAL_CONSOLE_ENABLED=True) + def test_a_new_serial_console_ignores_the_vnc_settings(self): + # even ones that would be refused: an empty new password, an unknown keymap + for vnc in ({"keymap": "de", "password_action": "none"}, + {"keymap": "xx", "password_action": "set", "password": ""}): + with self.subTest(vnc): + proxy, messages = self.post({**VNC, **vnc, "console_type": "pty"}) + self.assertEqual(messages, []) + proxy.set_console_type.assert_called_once_with("pty") + proxy.set_console_keymap.assert_not_called() + proxy.set_console_passwd.assert_not_called() + + @override_settings(SERIAL_CONSOLE_ENABLED=True) + def test_a_saved_serial_console_shows_and_takes_no_vnc_settings(self): + # the running VM still has VNC: the type to compare with is the saved one + proxy, messages = self.post({"console_type": "pty"}, console_type="pty") + self.assertEqual(messages, []) + proxy.set_console_type.assert_not_called() + proxy, _ = self.post({"console_type": "vnc", "keymap": "de"}, console_type="pty") + proxy.set_console_type.assert_called_once_with("vnc") + proxy.set_console_keymap.assert_not_called() + with patch("instances.models.wvmInstance") as wvm: + wvm.return_value.get_console_type.side_effect = lambda config=False: "pty" if config else "vnc" + form = ConsoleForm(instance=Instance.objects.get(pk=self.instance.pk), is_superuser=True) + self.assertEqual((form["console_type"].value(), sorted(form.fields)), ("pty", ["console_type"])) + + @override_settings(SERIAL_CONSOLE_ENABLED=False) + def test_serial_is_not_offered_while_the_serial_console_is_off(self): + proxy, _ = self.post({**VNC, "console_type": "pty"}) + proxy.set_console_type.assert_not_called() + + +class ConsoleListenerTests(TestCase): + def setUp(self): + User = get_user_model() + self.owner = User.objects.create_user("console-owner", password="pw") + self.admin = User.objects.create_superuser("console-admin", password="pw", email="ca@example.com") + compute = Compute.objects.create(name="cl", hostname="10.0.0.1", login="u", password="p", type=1) + self.vm = Instance.objects.create(compute=compute, name="cl-vm", uuid="78787878-0000-0000-0000-000000000000") + UserInstance.objects.create(user=self.owner, instance=self.vm, is_change=True, is_vnc=True) + + def post(self, user, data, local=True): + self.client.force_login(user) + with patch("instances.models.wvmInstance") as wvm: + proxy = wvm.return_value + proxy.console_listens_locally.return_value = local + proxy.get_console_listener_addr.return_value = "127.0.0.1" if local else "0.0.0.0" + proxy.get_console_passwd.return_value = "Old-pass-1" + proxy.get_console_keymap.return_value = "" + proxy.set_console_passwd.return_value = True + response = self.client.post(reverse("instances:update_console", args=[self.vm.id]), data) + return proxy, [str(m) for m in get_messages(response.wsgi_request)] + + def test_owner_cannot_change_the_listen_address(self): + proxy, _ = self.post(self.owner, {"listen_on": "0.0.0.0", "keymap": "auto", "password_action": "keep"}) + proxy.set_console_listener_addr.assert_not_called() + + def test_owner_cannot_remove_the_password_of_a_console_listening_beyond_localhost(self): + proxy, messages = self.post(self.owner, {"password_action": "none", "keymap": "de"}, local=False) + proxy.set_console_passwd.assert_not_called() + proxy.set_console_keymap.assert_not_called() + self.assertTrue(any("password" in m.lower() for m in messages), messages) + + @override_settings(SERIAL_CONSOLE_ENABLED=True) + def test_owner_can_switch_a_console_listening_beyond_localhost_to_serial(self): + proxy, messages = self.post(self.owner, {"console_type": "pty", "password_action": "none", "keymap": "auto"}, + local=False) + self.assertEqual(messages, []) + proxy.set_console_type.assert_called_once_with("pty") + proxy.set_console_passwd.assert_not_called() + + def test_owner_can_remove_the_password_of_a_local_console(self): + proxy, _ = self.post(self.owner, {"password_action": "none", "keymap": "auto"}) + proxy.set_console_passwd.assert_called_once_with("") + + def test_owner_can_set_a_password_of_any_console(self): + proxy, _ = self.post(self.owner, {"password_action": "set", "password": "s3cret-1", "keymap": "auto"}, local=False) + proxy.set_console_passwd.assert_called_once_with("s3cret-1") + + def test_superuser_can_open_the_console_to_all_interfaces(self): + proxy, _ = self.post(self.admin, {"listen_on": "0.0.0.0", "password_action": "none", "keymap": "auto"}) + proxy.set_console_listener_addr.assert_called_once_with("0.0.0.0") + proxy.set_console_passwd.assert_called_once_with("") + + def test_owner_sees_the_listen_address_but_cannot_edit_it(self): + self.client.force_login(self.owner) + with patch("instances.models.wvmInstance") as wvm, patch.object( + Compute, "status", new_callable=PropertyMock, return_value=True + ): + proxy = wvm.return_value + proxy.get_memory.return_value = 1024 + proxy.get_cur_memory.return_value = 1024 + proxy.get_max_memory.return_value = 4096 + proxy.get_console_listener_addr.return_value = "127.0.0.1" + for name in ("get_networks", "get_ifaces", "get_storages", "get_disk_devices", "get_media_devices", "get_net_devices"): + getattr(proxy, name).return_value = [] + proxy.get_hostdevs.return_value = ([], [], []) + proxy.get_status.return_value = 5 + response = self.client.get(reverse("instances:instance", args=[self.vm.id])) + form = response.context["console_form"] + self.assertTrue(form.fields["listen_on"].disabled) + + +VNC_PASSWORD = "s3cretVNCpass" + + +class ConsoleSettingsAccessTestCase(TestCase): + def setUp(self): + User = get_user_model() + view_perm = Permission.objects.get(codename="view_instances") + self.superuser = User.objects.create_superuser( + username="vncpw_super", email="vncpw_super@example.com", password="x" + ) + self.staff_viewer = User.objects.create_user( + username="vncpw_staff_viewer", password="x", is_staff=True + ) + self.staff_viewer.user_permissions.add(view_perm) + self.staff_owner = User.objects.create_user( + username="vncpw_staff_owner", password="x", is_staff=True + ) + self.vnc_owner = User.objects.create_user(username="vncpw_vnc_owner", password="x") + + self.compute = Compute.objects.create( + name="vncpw-compute", hostname="127.0.0.1:1", login="root", password="", type=1 + ) + self.instance = Instance.objects.create( + compute=self.compute, + name="vncpw-vm", + uuid="cccccccc-dddd-eeee-ffff-000000000000", + ) + UserInstance.objects.create( + instance=self.instance, user=self.staff_owner, is_change=True, is_vnc=False + ) + UserInstance.objects.create( + instance=self.instance, user=self.vnc_owner, is_change=True, is_vnc=True + ) + + def _detail_page(self, user): + self.client.force_login(user) + with patch("instances.models.wvmInstance") as mock_wvm, patch.object( + Compute, "status", new_callable=PropertyMock, return_value=True + ): + proxy = mock_wvm.return_value + proxy.get_memory.return_value = 1024 + proxy.get_cur_memory.return_value = 1024 + proxy.get_max_memory.return_value = 4096 + proxy.get_networks.return_value = [] + proxy.get_ifaces.return_value = [] + proxy.get_hostdevs.return_value = ([], [], []) + proxy.get_storages.return_value = [] + proxy.get_console_passwd.return_value = VNC_PASSWORD + proxy.get_console_type.return_value = "vnc" + proxy.get_status.return_value = 5 + proxy.get_disk_devices.return_value = [] + proxy.get_media_devices.return_value = [] + proxy.get_net_devices.return_value = [] + response = self.client.get( + reverse("instances:instance", args=[self.instance.id]) + ) + self.assertEqual(response.status_code, 200) + return response.content.decode() + + def test_only_users_who_may_change_the_console_see_its_settings(self): + form = reverse("instances:update_console", args=[self.instance.id]) + for user, shown in ((self.staff_viewer, False), (self.staff_owner, False), + (self.vnc_owner, True), (self.superuser, True)): + with self.subTest(user.username): + page = self._detail_page(user) + self.assertEqual(form in page, shown) + self.assertNotIn(VNC_PASSWORD, page) diff --git a/instances/tests/test_console_settings_access.py b/instances/tests/test_console_settings_access.py deleted file mode 100644 index 885ac086..00000000 --- a/instances/tests/test_console_settings_access.py +++ /dev/null @@ -1,85 +0,0 @@ -""" -The console settings form is rendered only for users who may change console -settings, which is the rule update_console enforces: a superuser, or an owner -with both is_change and is_vnc; not for is_staff or the global view_instances -permission. The form never carries the VNC password: it says whether one is -set. -""" -from unittest.mock import PropertyMock, patch - -from django.contrib.auth import get_user_model -from django.contrib.auth.models import Permission -from django.test import TestCase -from django.urls import reverse - -from accounts.models import UserInstance -from computes.models import Compute -from instances.models import Instance - -VNC_PASSWORD = "s3cretVNCpass" - - -class ConsoleSettingsAccessTestCase(TestCase): - def setUp(self): - User = get_user_model() - view_perm = Permission.objects.get(codename="view_instances") - self.superuser = User.objects.create_superuser( - username="vncpw_super", email="vncpw_super@example.com", password="x" - ) - self.staff_viewer = User.objects.create_user( - username="vncpw_staff_viewer", password="x", is_staff=True - ) - self.staff_viewer.user_permissions.add(view_perm) - self.staff_owner = User.objects.create_user( - username="vncpw_staff_owner", password="x", is_staff=True - ) - self.vnc_owner = User.objects.create_user(username="vncpw_vnc_owner", password="x") - - self.compute = Compute.objects.create( - name="vncpw-compute", hostname="127.0.0.1:1", login="root", password="", type=1 - ) - self.instance = Instance.objects.create( - compute=self.compute, - name="vncpw-vm", - uuid="cccccccc-dddd-eeee-ffff-000000000000", - ) - UserInstance.objects.create( - instance=self.instance, user=self.staff_owner, is_change=True, is_vnc=False - ) - UserInstance.objects.create( - instance=self.instance, user=self.vnc_owner, is_change=True, is_vnc=True - ) - - def _detail_page(self, user): - self.client.force_login(user) - with patch("instances.models.wvmInstance") as mock_wvm, patch.object( - Compute, "status", new_callable=PropertyMock, return_value=True - ): - proxy = mock_wvm.return_value - proxy.get_memory.return_value = 1024 - proxy.get_cur_memory.return_value = 1024 - proxy.get_max_memory.return_value = 4096 - proxy.get_networks.return_value = [] - proxy.get_ifaces.return_value = [] - proxy.get_hostdevs.return_value = ([], [], []) - proxy.get_storages.return_value = [] - proxy.get_console_passwd.return_value = VNC_PASSWORD - proxy.get_console_type.return_value = "vnc" - proxy.get_status.return_value = 5 - proxy.get_disk_devices.return_value = [] - proxy.get_media_devices.return_value = [] - proxy.get_net_devices.return_value = [] - response = self.client.get( - reverse("instances:instance", args=[self.instance.id]) - ) - self.assertEqual(response.status_code, 200) - return response.content.decode() - - def test_only_users_who_may_change_the_console_see_its_settings(self): - form = reverse("instances:update_console", args=[self.instance.id]) - for user, shown in ((self.staff_viewer, False), (self.staff_owner, False), - (self.vnc_owner, True), (self.superuser, True)): - with self.subTest(user.username): - page = self._detail_page(user) - self.assertEqual(form in page, shown) - self.assertNotIn(VNC_PASSWORD, page) diff --git a/instances/tests/test_security_mutations.py b/instances/tests/test_permissions.py similarity index 100% rename from instances/tests/test_security_mutations.py rename to instances/tests/test_permissions.py diff --git a/instances/tests/test_quota.py b/instances/tests/test_quota.py index c239ff2c..4bde91e4 100644 --- a/instances/tests/test_quota.py +++ b/instances/tests/test_quota.py @@ -18,6 +18,10 @@ from libvirt import VIR_DOMAIN_PAUSED, VIR_DOMAIN_RUNNING from operations.models import Operation from vrtManager.util import OperationError import socket +import threading +import time +from django.db import connection +from django.test import Client, TransactionTestCase, override_settings def vm_reader(vcpu=2, memory=1024, disk_gib=10, pending=None): @@ -292,3 +296,58 @@ class QuotaWithDownHostTestCase(TestCase): get_settings() # app settings are normally loaded by middleware self.assertEqual(check_user_quota(user, 1, 1, 128, 1), QUOTA_UNVERIFIED) + + +QUOTA_CPUS = 4 + + +# Sessions and messages in cookies: concurrent writes to the in-memory test +# database would fail for reasons unrelated to the quota race. +@override_settings( + SESSION_ENGINE="django.contrib.sessions.backends.signed_cookies", + MESSAGE_STORAGE="django.contrib.messages.storage.cookie.CookieStorage", +) + + +class QuotaRaceTestCase(TransactionTestCase): + """Two concurrent resizes of one user's VMs must not both pass the quota.""" + + def setUp(self): + self.user = get_user_model().objects.create_user("quota_user", password="pw") + compute = Compute.objects.create(name="quota-compute", hostname="127.0.0.1", login="", password="", type=4) + self.vms = [ + Instance.objects.create(compute=compute, name=f"quota-vm-{i}", uuid=f"1111111{i}-2222-3333-4444-555555555555") + for i in range(2) + ] + for vm in self.vms: + UserInstance.objects.create(user=self.user, instance=vm, is_change=True) + self.allocated = 2 # two VMs with one vCPU each + + def fake_quota(self, user, instance, cpu, memory, disk): + seen = self.allocated + time.sleep(0.3) # widen the window between the check and the change + return "cpu" if seen + cpu > QUOTA_CPUS else "" + + def fake_resize(self, cur_vcpu, vcpu): + self.allocated += int(vcpu) - 1 + + def resize(self, vm): + client = Client() + client.force_login(self.user) + client.post(reverse("instances:resizevm_cpu", args=[vm.id]), {"vcpu": 3, "cur_vcpu": 3}) + connection.close() + + def test_second_concurrent_resize_is_refused(self): + with patch("instances.views.utils.check_user_quota", self.fake_quota), patch( + "instances.models.wvmInstance" + ) as wvm: + wvm.return_value.get_vcpu.return_value = 1 + wvm.return_value.instance.isActive.return_value = False # vCPUs change on a shut-off VM + wvm.return_value.resize_cpu.side_effect = self.fake_resize + threads = [threading.Thread(target=self.resize, args=(vm,)) for vm in self.vms] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(wvm.return_value.resize_cpu.call_count, 1) + self.assertLessEqual(self.allocated, QUOTA_CPUS) diff --git a/instances/tests/test_quota_lock.py b/instances/tests/test_quota_lock.py deleted file mode 100644 index 5ac608f5..00000000 --- a/instances/tests/test_quota_lock.py +++ /dev/null @@ -1,63 +0,0 @@ -import threading -import time -from unittest.mock import patch - -from accounts.models import UserInstance -from computes.models import Compute -from django.contrib.auth import get_user_model -from django.db import connection -from django.test import Client, TransactionTestCase, override_settings -from django.urls import reverse -from instances.models import Instance - -QUOTA_CPUS = 4 - - -# Sessions and messages in cookies: concurrent writes to the in-memory test -# database would fail for reasons unrelated to the quota race. -@override_settings( - SESSION_ENGINE="django.contrib.sessions.backends.signed_cookies", - MESSAGE_STORAGE="django.contrib.messages.storage.cookie.CookieStorage", -) -class QuotaRaceTestCase(TransactionTestCase): - """Two concurrent resizes of one user's VMs must not both pass the quota.""" - - def setUp(self): - self.user = get_user_model().objects.create_user("quota_user", password="pw") - compute = Compute.objects.create(name="quota-compute", hostname="127.0.0.1", login="", password="", type=4) - self.vms = [ - Instance.objects.create(compute=compute, name=f"quota-vm-{i}", uuid=f"1111111{i}-2222-3333-4444-555555555555") - for i in range(2) - ] - for vm in self.vms: - UserInstance.objects.create(user=self.user, instance=vm, is_change=True) - self.allocated = 2 # two VMs with one vCPU each - - def fake_quota(self, user, instance, cpu, memory, disk): - seen = self.allocated - time.sleep(0.3) # widen the window between the check and the change - return "cpu" if seen + cpu > QUOTA_CPUS else "" - - def fake_resize(self, cur_vcpu, vcpu): - self.allocated += int(vcpu) - 1 - - def resize(self, vm): - client = Client() - client.force_login(self.user) - client.post(reverse("instances:resizevm_cpu", args=[vm.id]), {"vcpu": 3, "cur_vcpu": 3}) - connection.close() - - def test_second_concurrent_resize_is_refused(self): - with patch("instances.views.utils.check_user_quota", self.fake_quota), patch( - "instances.models.wvmInstance" - ) as wvm: - wvm.return_value.get_vcpu.return_value = 1 - wvm.return_value.instance.isActive.return_value = False # vCPUs change on a shut-off VM - wvm.return_value.resize_cpu.side_effect = self.fake_resize - threads = [threading.Thread(target=self.resize, args=(vm,)) for vm in self.vms] - for t in threads: - t.start() - for t in threads: - t.join() - self.assertEqual(wvm.return_value.resize_cpu.call_count, 1) - self.assertLessEqual(self.allocated, QUOTA_CPUS) diff --git a/instances/tests/test_template_flag.py b/instances/tests/test_template_flag.py deleted file mode 100644 index 5880e7ba..00000000 --- a/instances/tests/test_template_flag.py +++ /dev/null @@ -1,81 +0,0 @@ -""" -Only superusers and staff may mark or unmark a VM as a template, on its own -form: the title and description form leaves the flag alone. -""" -from unittest.mock import patch - -from django.contrib.auth import get_user_model -from django.test import TestCase -from django.urls import reverse - -from accounts.models import UserInstance -from computes.models import Compute -from instances.models import Instance -from logs.models import Logs - - -class TemplateFlagTestCase(TestCase): - def setUp(self): - User = get_user_model() - self.superuser = User.objects.create_superuser( - username="tpl_super", email="tpl_super@example.com", password="x" - ) - self.owner = User.objects.create_user(username="tpl_owner", password="x") - self.staff_owner = User.objects.create_user( - username="tpl_staff_owner", password="x", is_staff=True - ) - self.compute = Compute.objects.create( - name="tpl-compute", hostname="127.0.0.1:1", login="root", password="", type=1 - ) - self.instance = Instance.objects.create( - compute=self.compute, - name="tpl-vm", - uuid="dddddddd-eeee-ffff-0000-111111111111", - ) - for user in (self.owner, self.staff_owner): - UserInstance.objects.create(instance=self.instance, user=user, is_change=True) - - def _post(self, user, action, data, status=302): - self.client.force_login(user) - with patch("instances.models.wvmInstance") as wvm: - wvm.return_value.get_status.return_value = 5 - # unmarking a template checks the owner's quota with the VM's size - wvm.return_value.get_vcpu.return_value = 1 - wvm.return_value.get_memory.return_value = 1024 - wvm.return_value.get_disk_devices.return_value = [] - res = self.client.post(reverse(f"instances:{action}", args=[self.instance.id]), data) - self.assertEqual(res.status_code, status) - self.instance.refresh_from_db() - - def _make_template(self): - self.instance.is_template = True - self.instance.save(update_fields=["is_template"]) - - def test_owner_cannot_mark_vm_as_template(self): - self._post(self.owner, "set_template", {"is_template": "True"}, status=403) - self.assertFalse(self.instance.is_template) - - def test_owner_cannot_change_a_template(self): - # a template is changed only by superusers and staff owners - self._make_template() - self._post(self.owner, "change_options", {"title": "new title", "description": ""}, status=403) - self.assertTrue(self.instance.is_template) - - def test_staff_owner_can_mark_and_unmark_template(self): - self._post(self.staff_owner, "set_template", {"is_template": "True"}) - self.assertTrue(self.instance.is_template) - # an unchecked switch is not submitted: "not a template" - self._post(self.staff_owner, "set_template", {}) - self.assertFalse(self.instance.is_template) - self.assertEqual(list(Logs.objects.filter(message__startswith="VM").order_by("id").values_list("message", "success")), - [("VM tpl-vm: mark as template", True), ("VM tpl-vm: unmark as template", True)]) - - def test_superuser_can_unmark_template(self): - self._make_template() - self._post(self.superuser, "set_template", {}) - self.assertFalse(self.instance.is_template) - - def test_the_title_form_leaves_the_template_alone(self): - self._make_template() - self._post(self.superuser, "change_options", {"title": "t", "description": ""}) - self.assertTrue(self.instance.is_template) diff --git a/instances/tests/test_template_quota.py b/instances/tests/test_template_quota.py deleted file mode 100644 index f7ff67bd..00000000 --- a/instances/tests/test_template_quota.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Templates do not count towards the quota and cannot be started. Turning a -template back into a VM adds it to its owner's usage, so for users with a -quota it is checked like any other increase. Only a shut-off VM changes.""" - -from unittest.mock import patch - -from accounts.models import UserInstance -from computes.models import Compute -from django.contrib.auth import get_user_model -from django.contrib.messages import get_messages -from django.test import TestCase -from django.urls import reverse -from instances.models import Instance -from instances.utils import QUOTA_UNVERIFIED, QuotaExceeded - - -class TemplateFlagQuotaTests(TestCase): - def setUp(self): - User = get_user_model() - self.staff = User.objects.create_user("tq-staff", password="pw", is_staff=True) - compute = Compute.objects.create(name="tq", hostname="10.0.0.5", login="u", password="p", type=1) - self.vm = Instance.objects.create(compute=compute, name="tq-vm", uuid="12345678-0000-0000-0000-00000000aaaa") - UserInstance.objects.create(user=self.staff, instance=self.vm, is_change=True) - - def post(self, user, is_template, quota="", status=5): - self.client.force_login(user) - data = {"is_template": "on"} if is_template else {} - with patch("instances.models.wvmInstance") as wvm, patch( - "instances.views.utils.check_user_quota", return_value=quota - ) as check: - wvm.return_value.get_status.return_value = status - wvm.return_value.get_vcpu.return_value = 2 - wvm.return_value.get_memory.return_value = 2048 - wvm.return_value.get_disk_devices.return_value = [{"size": 10 << 30}] - response = self.client.post(reverse("instances:set_template", args=[self.vm.id]), data) - self.vm.refresh_from_db() - return check, [str(m) for m in get_messages(response.wsgi_request)] - - def test_making_a_template_needs_no_quota(self): - check, _ = self.post(self.staff, is_template=True) - self.assertTrue(self.vm.is_template) - check.assert_not_called() - - def test_turning_a_template_into_a_vm_is_checked(self): - Instance.objects.filter(pk=self.vm.pk).update(is_template=True) - check, _ = self.post(self.staff, is_template=False) - self.assertFalse(self.vm.is_template) - check.assert_called_once_with(self.staff, 1, 2, 2048, 10) - - def test_over_quota_the_template_stays_a_template(self): - Instance.objects.filter(pk=self.vm.pk).update(is_template=True) - _, messages = self.post(self.staff, is_template=False, quota=QuotaExceeded("cpu", 9, 8)) - self.assertTrue(self.vm.is_template) - self.assertTrue(any("quota" in m for m in messages), messages) - - def test_an_unverifiable_quota_keeps_it_a_template(self): - Instance.objects.filter(pk=self.vm.pk).update(is_template=True) - _, messages = self.post(self.staff, is_template=False, quota=QUOTA_UNVERIFIED) - self.assertTrue(self.vm.is_template) - self.assertTrue(any("cannot be reached" in m for m in messages), messages) - - def test_a_vm_that_is_not_shut_off_stays_as_it_is(self): - # a running VM made a template would leave the quota while it runs - for status in (1, 3): - with self.subTest(status=status): - check, messages = self.post(self.staff, is_template=True, status=status) - self.assertFalse(self.vm.is_template) - self.assertIn("Shut the VM down to change whether it is a template.", messages) - - def test_superusers_have_no_quota(self): - Instance.objects.filter(pk=self.vm.pk).update(is_template=True) - admin = get_user_model().objects.create_superuser("tq-admin", "tq@example.com", "pw") - check, _ = self.post(admin, is_template=False, quota=QuotaExceeded("cpu", 9, 8)) - self.assertFalse(self.vm.is_template) - check.assert_not_called() diff --git a/instances/tests/test_template_rules.py b/instances/tests/test_template_rules.py deleted file mode 100644 index fb352086..00000000 --- a/instances/tests/test_template_rules.py +++ /dev/null @@ -1,64 +0,0 @@ -""" -A template is not started (power on or power cycle, web or API) and is -changed or deleted only by superusers and staff owners. -""" -from unittest.mock import patch - -from django.contrib.auth import get_user_model -from django.test import TestCase -from django.urls import reverse -from rest_framework.test import APIClient - -from accounts.models import UserInstance -from computes.models import Compute -from instances.models import Instance -from operations.models import Operation - - -class TemplateRulesTestCase(TestCase): - def setUp(self): - compute = Compute.objects.create(name="tpl-c", hostname="192.0.2.5", login="root", password="", type=1) - self.template = Instance.objects.create( - compute=compute, name="tpl-vm", uuid="ffffffff-0000-1111-2222-333333333333", is_template=True - ) - self.owner = get_user_model().objects.create_user("tpl_owner", password="x") - UserInstance.objects.create(instance=self.template, user=self.owner, is_delete=True) - self.staff = get_user_model().objects.create_user("tpl_staff", password="x", is_staff=True) - UserInstance.objects.create(instance=self.template, user=self.staff, is_delete=True) - - def test_power_cycle_does_not_start_a_template(self): - api = APIClient() - api.force_login(self.owner) - self.client.force_login(self.owner) - for post in ( - lambda: self.client.post(reverse("instances:powercycle", args=[self.template.id])), - lambda: api.post(reverse("compute-instance-powercycle", args=[self.template.compute_id, self.template.id])), - ): - with patch("instances.models.wvmInstance") as wvm: - post() - self.assertFalse(wvm.return_value.start.called) - self.assertFalse(wvm.return_value.force_shutdown.called) - - def test_only_staff_owners_delete_a_template(self): - self.client.force_login(self.owner) - with patch("instances.models.wvmInstance") as wvm: - response = self.client.post(reverse("instances:destroy", args=[self.template.id])) - self.assertEqual(response.status_code, 403) - self.assertFalse(wvm.return_value.delete.called) - - self.client.force_login(self.staff) - with patch("instances.models.wvmInstance"): - self.client.post(reverse("instances:destroy", args=[self.template.id]), {"submission_key": "key-tpldel01"}) - self.assertTrue(Operation.objects.filter(kind="instance.delete").exists()) # queued - - def test_the_destroy_confirmation_page_offers_the_form_to_who_may_delete(self): - url = reverse("instances:destroy", args=[self.template.id]) - superuser = get_user_model().objects.create_superuser("tpl_root", "r@example.com", "x") - for user, status, form in ((superuser, 200, True), (self.staff, 200, True), (self.owner, 403, False)): - with self.subTest(user=user.username): - self.client.force_login(user) - with patch("instances.models.wvmInstance") as wvm: - wvm.return_value.get_status.return_value = 5 - response = self.client.get(url) - self.assertEqual(response.status_code, status) - self.assertEqual('id="delete_form"' in response.content.decode(), form) diff --git a/instances/tests/test_templates.py b/instances/tests/test_templates.py new file mode 100644 index 00000000..1f77556b --- /dev/null +++ b/instances/tests/test_templates.py @@ -0,0 +1,194 @@ +"""Templates: only superusers and staff mark or unmark one; a template is +not started, does not count towards the quota (turning it back into a VM +is checked), and is changed or deleted only by superusers and staff owners.""" +from unittest.mock import patch + +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.urls import reverse +from rest_framework.test import APIClient + +from accounts.models import UserInstance +from computes.models import Compute +from instances.models import Instance +from operations.models import Operation +from logs.models import Logs +from django.contrib.messages import get_messages +from instances.utils import QUOTA_UNVERIFIED, QuotaExceeded + + +class TemplateRulesTestCase(TestCase): + def setUp(self): + compute = Compute.objects.create(name="tpl-c", hostname="192.0.2.5", login="root", password="", type=1) + self.template = Instance.objects.create( + compute=compute, name="tpl-vm", uuid="ffffffff-0000-1111-2222-333333333333", is_template=True + ) + self.owner = get_user_model().objects.create_user("tpl_owner", password="x") + UserInstance.objects.create(instance=self.template, user=self.owner, is_delete=True) + self.staff = get_user_model().objects.create_user("tpl_staff", password="x", is_staff=True) + UserInstance.objects.create(instance=self.template, user=self.staff, is_delete=True) + + def test_power_cycle_does_not_start_a_template(self): + api = APIClient() + api.force_login(self.owner) + self.client.force_login(self.owner) + for post in ( + lambda: self.client.post(reverse("instances:powercycle", args=[self.template.id])), + lambda: api.post(reverse("compute-instance-powercycle", args=[self.template.compute_id, self.template.id])), + ): + with patch("instances.models.wvmInstance") as wvm: + post() + self.assertFalse(wvm.return_value.start.called) + self.assertFalse(wvm.return_value.force_shutdown.called) + + def test_only_staff_owners_delete_a_template(self): + self.client.force_login(self.owner) + with patch("instances.models.wvmInstance") as wvm: + response = self.client.post(reverse("instances:destroy", args=[self.template.id])) + self.assertEqual(response.status_code, 403) + self.assertFalse(wvm.return_value.delete.called) + + self.client.force_login(self.staff) + with patch("instances.models.wvmInstance"): + self.client.post(reverse("instances:destroy", args=[self.template.id]), {"submission_key": "key-tpldel01"}) + self.assertTrue(Operation.objects.filter(kind="instance.delete").exists()) # queued + + def test_the_destroy_confirmation_page_offers_the_form_to_who_may_delete(self): + url = reverse("instances:destroy", args=[self.template.id]) + superuser = get_user_model().objects.create_superuser("tpl_root", "r@example.com", "x") + for user, status, form in ((superuser, 200, True), (self.staff, 200, True), (self.owner, 403, False)): + with self.subTest(user=user.username): + self.client.force_login(user) + with patch("instances.models.wvmInstance") as wvm: + wvm.return_value.get_status.return_value = 5 + response = self.client.get(url) + self.assertEqual(response.status_code, status) + self.assertEqual('id="delete_form"' in response.content.decode(), form) + + +class TemplateFlagTestCase(TestCase): + def setUp(self): + User = get_user_model() + self.superuser = User.objects.create_superuser( + username="tpl_super", email="tpl_super@example.com", password="x" + ) + self.owner = User.objects.create_user(username="tpl_owner", password="x") + self.staff_owner = User.objects.create_user( + username="tpl_staff_owner", password="x", is_staff=True + ) + self.compute = Compute.objects.create( + name="tpl-compute", hostname="127.0.0.1:1", login="root", password="", type=1 + ) + self.instance = Instance.objects.create( + compute=self.compute, + name="tpl-vm", + uuid="dddddddd-eeee-ffff-0000-111111111111", + ) + for user in (self.owner, self.staff_owner): + UserInstance.objects.create(instance=self.instance, user=user, is_change=True) + + def _post(self, user, action, data, status=302): + self.client.force_login(user) + with patch("instances.models.wvmInstance") as wvm: + wvm.return_value.get_status.return_value = 5 + # unmarking a template checks the owner's quota with the VM's size + wvm.return_value.get_vcpu.return_value = 1 + wvm.return_value.get_memory.return_value = 1024 + wvm.return_value.get_disk_devices.return_value = [] + res = self.client.post(reverse(f"instances:{action}", args=[self.instance.id]), data) + self.assertEqual(res.status_code, status) + self.instance.refresh_from_db() + + def _make_template(self): + self.instance.is_template = True + self.instance.save(update_fields=["is_template"]) + + def test_owner_cannot_mark_vm_as_template(self): + self._post(self.owner, "set_template", {"is_template": "True"}, status=403) + self.assertFalse(self.instance.is_template) + + def test_owner_cannot_change_a_template(self): + # a template is changed only by superusers and staff owners + self._make_template() + self._post(self.owner, "change_options", {"title": "new title", "description": ""}, status=403) + self.assertTrue(self.instance.is_template) + + def test_staff_owner_can_mark_and_unmark_template(self): + self._post(self.staff_owner, "set_template", {"is_template": "True"}) + self.assertTrue(self.instance.is_template) + # an unchecked switch is not submitted: "not a template" + self._post(self.staff_owner, "set_template", {}) + self.assertFalse(self.instance.is_template) + self.assertEqual(list(Logs.objects.filter(message__startswith="VM").order_by("id").values_list("message", "success")), + [("VM tpl-vm: mark as template", True), ("VM tpl-vm: unmark as template", True)]) + + def test_superuser_can_unmark_template(self): + self._make_template() + self._post(self.superuser, "set_template", {}) + self.assertFalse(self.instance.is_template) + + def test_the_title_form_leaves_the_template_alone(self): + self._make_template() + self._post(self.superuser, "change_options", {"title": "t", "description": ""}) + self.assertTrue(self.instance.is_template) + + +class TemplateFlagQuotaTests(TestCase): + def setUp(self): + User = get_user_model() + self.staff = User.objects.create_user("tq-staff", password="pw", is_staff=True) + compute = Compute.objects.create(name="tq", hostname="10.0.0.5", login="u", password="p", type=1) + self.vm = Instance.objects.create(compute=compute, name="tq-vm", uuid="12345678-0000-0000-0000-00000000aaaa") + UserInstance.objects.create(user=self.staff, instance=self.vm, is_change=True) + + def post(self, user, is_template, quota="", status=5): + self.client.force_login(user) + data = {"is_template": "on"} if is_template else {} + with patch("instances.models.wvmInstance") as wvm, patch( + "instances.views.utils.check_user_quota", return_value=quota + ) as check: + wvm.return_value.get_status.return_value = status + wvm.return_value.get_vcpu.return_value = 2 + wvm.return_value.get_memory.return_value = 2048 + wvm.return_value.get_disk_devices.return_value = [{"size": 10 << 30}] + response = self.client.post(reverse("instances:set_template", args=[self.vm.id]), data) + self.vm.refresh_from_db() + return check, [str(m) for m in get_messages(response.wsgi_request)] + + def test_making_a_template_needs_no_quota(self): + check, _ = self.post(self.staff, is_template=True) + self.assertTrue(self.vm.is_template) + check.assert_not_called() + + def test_turning_a_template_into_a_vm_is_checked(self): + Instance.objects.filter(pk=self.vm.pk).update(is_template=True) + check, _ = self.post(self.staff, is_template=False) + self.assertFalse(self.vm.is_template) + check.assert_called_once_with(self.staff, 1, 2, 2048, 10) + + def test_over_quota_the_template_stays_a_template(self): + Instance.objects.filter(pk=self.vm.pk).update(is_template=True) + _, messages = self.post(self.staff, is_template=False, quota=QuotaExceeded("cpu", 9, 8)) + self.assertTrue(self.vm.is_template) + self.assertTrue(any("quota" in m for m in messages), messages) + + def test_an_unverifiable_quota_keeps_it_a_template(self): + Instance.objects.filter(pk=self.vm.pk).update(is_template=True) + _, messages = self.post(self.staff, is_template=False, quota=QUOTA_UNVERIFIED) + self.assertTrue(self.vm.is_template) + self.assertTrue(any("cannot be reached" in m for m in messages), messages) + + def test_a_vm_that_is_not_shut_off_stays_as_it_is(self): + # a running VM made a template would leave the quota while it runs + for status in (1, 3): + with self.subTest(status=status): + check, messages = self.post(self.staff, is_template=True, status=status) + self.assertFalse(self.vm.is_template) + self.assertIn("Shut the VM down to change whether it is a template.", messages) + + def test_superusers_have_no_quota(self): + Instance.objects.filter(pk=self.vm.pk).update(is_template=True) + admin = get_user_model().objects.create_superuser("tq-admin", "tq@example.com", "pw") + check, _ = self.post(admin, is_template=False, quota=QuotaExceeded("cpu", 9, 8)) + self.assertFalse(self.vm.is_template) + check.assert_not_called() diff --git a/instances/tests/test_update_console.py b/instances/tests/test_update_console.py deleted file mode 100644 index ea58dfdf..00000000 --- a/instances/tests/test_update_console.py +++ /dev/null @@ -1,90 +0,0 @@ -"""Console settings: each change is saved to the persistent definition, the -password only by the action chosen for it, and the VNC settings only while -the saved console type is VNC.""" - -from unittest.mock import patch - -from computes.models import Compute -from django.contrib.auth import get_user_model -from django.contrib.messages import get_messages -from django.test import TestCase, override_settings -from django.urls import reverse -from instances.forms import ConsoleForm -from instances.models import Instance - -VNC = {"console_type": "vnc", "listen_on": "127.0.0.1", "keymap": "auto", "password_action": "keep"} - - -class UpdateConsoleTestCase(TestCase): - def setUp(self): - admin = get_user_model().objects.create_superuser("console_admin", "c@example.com", "pw") - self.client.force_login(admin) - compute = Compute.objects.create(name="console-compute", hostname="127.0.0.1", login="", password="", type=4) - self.instance = Instance.objects.create( - compute=compute, name="console-vm", uuid="11111111-2222-3333-4444-555555555555" - ) - - def post(self, data, console_type="vnc"): - """console_type: the persistent definition's; the live one is VNC.""" - with patch("instances.models.wvmInstance") as wvm: - proxy = wvm.return_value - proxy.get_console_type.side_effect = lambda config=False: console_type if config else "vnc" - proxy.get_console_listener_addr.return_value = "127.0.0.1" - proxy.get_console_keymap.return_value = "" - proxy.set_console_passwd.return_value = True - response = self.client.post(reverse("instances:update_console", args=[self.instance.id]), data) - return proxy, [str(m) for m in get_messages(response.wsgi_request)] - - def test_console_settings_are_applied(self): - proxy, _ = self.post({**VNC, "listen_on": "0.0.0.0", "keymap": "de"}) - proxy.set_console_keymap.assert_called_once_with("de") - proxy.set_console_listener_addr.assert_called_once_with("0.0.0.0") - proxy.set_console_passwd.assert_not_called() - - def test_the_password_changes_by_the_action_chosen(self): - for action, password in (("set", " a b c "), ("set", "éééé"), ("none", "")): - with self.subTest(action, password=password): - proxy, _ = self.post({**VNC, "password_action": action, "password": password}) - proxy.set_console_passwd.assert_called_once_with(password) - proxy, _ = self.post({**VNC, "password_action": "generate"}) - (generated,), _ = proxy.set_console_passwd.call_args - self.assertEqual(len(generated), 8) - - def test_a_password_vnc_cannot_take_is_refused_and_nothing_is_saved(self): - for password in ("", "123456789", "ééééé"): # libvirt counts bytes: é is two - with self.subTest(password=password): - proxy, messages = self.post({**VNC, "keymap": "de", "password_action": "set", "password": password}) - proxy.set_console_passwd.assert_not_called() - proxy.set_console_keymap.assert_not_called() - self.assertTrue(messages[0].startswith("Console settings were not saved"), messages) - - @override_settings(SERIAL_CONSOLE_ENABLED=True) - def test_a_new_serial_console_ignores_the_vnc_settings(self): - # even ones that would be refused: an empty new password, an unknown keymap - for vnc in ({"keymap": "de", "password_action": "none"}, - {"keymap": "xx", "password_action": "set", "password": ""}): - with self.subTest(vnc): - proxy, messages = self.post({**VNC, **vnc, "console_type": "pty"}) - self.assertEqual(messages, []) - proxy.set_console_type.assert_called_once_with("pty") - proxy.set_console_keymap.assert_not_called() - proxy.set_console_passwd.assert_not_called() - - @override_settings(SERIAL_CONSOLE_ENABLED=True) - def test_a_saved_serial_console_shows_and_takes_no_vnc_settings(self): - # the running VM still has VNC: the type to compare with is the saved one - proxy, messages = self.post({"console_type": "pty"}, console_type="pty") - self.assertEqual(messages, []) - proxy.set_console_type.assert_not_called() - proxy, _ = self.post({"console_type": "vnc", "keymap": "de"}, console_type="pty") - proxy.set_console_type.assert_called_once_with("vnc") - proxy.set_console_keymap.assert_not_called() - with patch("instances.models.wvmInstance") as wvm: - wvm.return_value.get_console_type.side_effect = lambda config=False: "pty" if config else "vnc" - form = ConsoleForm(instance=Instance.objects.get(pk=self.instance.pk), is_superuser=True) - self.assertEqual((form["console_type"].value(), sorted(form.fields)), ("pty", ["console_type"])) - - @override_settings(SERIAL_CONSOLE_ENABLED=False) - def test_serial_is_not_offered_while_the_serial_console_is_off(self): - proxy, _ = self.post({**VNC, "console_type": "pty"}) - proxy.set_console_type.assert_not_called()