import random import string from django.conf import settings from django.test import TestCase, override_settings from unittest import skipIf from unittest.mock import MagicMock, patch from oca.pool import WrongIdError from .exceptions import TemplateNotFoundError from .models import OpenNebulaManager from .serializers import ( VirtualMachineSerializer, strip_template_prefix, template_name_from_vm_name ) from utils.models import CustomUser @skipIf( settings.OPENNEBULA_DOMAIN is None or settings.OPENNEBULA_DOMAIN == "test_domain", """OpenNebula details unavailable, so skipping OpenNebulaManagerTestCases""" ) class OpenNebulaManagerTestCases(TestCase): """This class defines the test suite for the opennebula manager model.""" def setUp(self): """Define the test client and other test variables.""" self.email = '{}@ungleich.ch'.format(''.join(random.choices(string.ascii_uppercase, k=10))) self.password = ''.join(random.choices(string.ascii_uppercase + string.digits, k=20)) self.user = CustomUser.objects.create(name='test', email=self.email, password=self.password) self.vm_specs = {} self.vm_specs['cpu'] = 1 self.vm_specs['memory'] = 2 self.vm_specs['disk_size'] = 10 self.manager = OpenNebulaManager() def test_connect_to_server(self): """Test the opennebula manager can connect to a server.""" try: ver = self.manager.oneadmin_client.version() except: ver = None self.assertTrue(ver is not None) def test_get_user(self): """Test the opennebula manager can get a existing user.""" self.manager.create_user(self.user) user = self.manager._get_user(self.user) name = user.name self.assertNotEqual(name, None) def test_create_and_delete_user(self): """Test the opennebula manager can create and delete a new user.""" old_count = len(self.manager._get_user_pool()) self.manager = OpenNebulaManager(email=self.email, password=self.password) user_pool = self.manager._get_user_pool() new_count = len(user_pool) # Remove the user afterwards user = user_pool.get_by_name(self.email) user.delete() self.assertNotEqual(old_count, new_count) def test_user_can_login(self): """ Test the manager can login to a new created user""" self.manager.create_user(self.user) user = self.manager._get_user(self.user) client = self.manager._get_client(self.user) version = client.version() # Cleanup user.delete() self.assertNotEqual(version, None) def test_add_public_key_to_user(self): """ Test the manager can add a new public key to an user """ self.manager.create_user(self.user) user = self.manager._get_user(self.user) public_key = 'test' self.manager.add_public_key(self.user, public_key) # Fetch new user information from opennebula user.info() user_public_key = user.template.ssh_public_key # Cleanup user.delete() self.assertEqual(user_public_key, public_key) def test_append_public_key_to_user(self): """ Test the manager can append a new public key to an user """ self.manager.create_user(self.user) user = self.manager._get_user(self.user) public_key = 'test' self.manager.add_public_key(self.user, public_key) # Fetch new user information from opennebula user.info() old_public_key = user.template.ssh_public_key self.manager.add_public_key(self.user, public_key, merge=True) user.info() new_public_key = user.template.ssh_public_key # Cleanup user.delete() self.assertEqual(new_public_key, '{}\n{}'.format(old_public_key, public_key)) def test_remove_public_key_to_user(self): """ Test the manager can remove a public key from an user """ self.manager.create_user(self.user) user = self.manager._get_user(self.user) public_key = 'test' self.manager.add_public_key(self.user, public_key) self.manager.add_public_key(self.user, public_key, merge=True) user.info() old_public_key = user.template.ssh_public_key self.manager.remove_public_key(self.user, public_key) user.info() new_public_key = user.template.ssh_public_key # Cleanup user.delete() self.assertEqual(new_public_key, old_public_key.replace('{}\n'.format(public_key), '', 1)) def test_requires_ssh_key_for_new_vm(self): """Test the opennebula manager requires the user to have a ssh key when creating a new vm""" @skipIf( settings.OPENNEBULA_DOMAIN is None or settings.OPENNEBULA_DOMAIN == "test_domain", """OpenNebula details unavailable, so skipping VirtualMachineSerializerTestCase""" ) class VirtualMachineSerializerTestCase(TestCase): def setUp(self): """Define the test client and other test variables.""" self.manager = OpenNebulaManager(email=None, password=None) def test_serializer_strips_of_public(self): """ Test the serialized virtual machine object contains no 'public-'.""" for vm in self.manager.get_vms(): serialized = VirtualMachineSerializer(vm) self.assertEqual( serialized.data.get('name'), vm.name.lstrip('public-') ) break class TemplateNameHelpersTestCase(TestCase): def test_strip_template_prefix(self): self.assertEqual(strip_template_prefix('public-Ubuntu 20.10'), 'Ubuntu 20.10') self.assertEqual(strip_template_prefix('ipv6only-Alpine 3.21'), 'Alpine 3.21') self.assertEqual(strip_template_prefix('Devuan Beowulf'), 'Devuan Beowulf') def test_strip_template_prefix_is_not_lstrip(self): # str.lstrip('public-') strips a character set and would give 'ntu' self.assertEqual(strip_template_prefix('public-ubuntu'), 'ubuntu') def test_template_name_from_vm_name(self): self.assertEqual( template_name_from_vm_name('public-Ubuntu 20.10-26623', 26623), 'public-Ubuntu 20.10' ) def test_template_name_from_vm_name_keeps_custom_names(self): # VMs created or renamed outside dynamicweb self.assertEqual(template_name_from_vm_name('gitea', 27001), 'gitea') self.assertEqual( template_name_from_vm_name('ipv6only-Alpine 3.15-27397-Jonas', 27397), 'ipv6only-Alpine 3.15-27397-Jonas' ) def _fake_vm(vm_id, name, template_id): vm = MagicMock() vm.id = vm_id vm.name = name vm.template.template_id = template_id return vm class VirtualMachineSerializerMissingTemplateTestCase(TestCase): """ A VM whose OpenNebula template has been deleted must not break the serialization of the whole VM list (the symptom was the generic "backend connection error" banner on /hosting/my-virtual-machines). """ def _configuration(self, vms, template_names): serializer = VirtualMachineSerializer(vms, many=True) with patch.object(OpenNebulaManager, '__init__', return_value=None), \ patch.object(OpenNebulaManager, 'get_template_names', return_value=template_names) as get_names: configurations = [ serializer.child.get_configuration(vm) for vm in vms ] return configurations, get_names @override_settings(UPDATED_TEMPLATES_DICT={}) def test_missing_template_falls_back_to_vm_name(self): vms = [ _fake_vm(28344, 'public-Alpine 3.21-28344', 310), _fake_vm(26623, 'public-Ubuntu 20.10-26623', 164), ] configurations, _ = self._configuration( vms, {310: 'public-Alpine 3.21'} ) self.assertEqual(configurations, ['Alpine 3.21', 'Ubuntu 20.10']) @override_settings(UPDATED_TEMPLATES_DICT={164: 192}) def test_updated_templates_mapping_is_applied(self): vms = [_fake_vm(26623, 'public-Ubuntu 20.10-26623', 164)] configurations, _ = self._configuration( vms, {192: 'public-Ubuntu 20.10 (new)'} ) self.assertEqual(configurations, ['Ubuntu 20.10 (new)']) @override_settings(UPDATED_TEMPLATES_DICT={}) def test_template_pool_is_fetched_once_per_list(self): vms = [ _fake_vm(1, 'public-Alpine 3.21-1', 310), _fake_vm(2, 'public-Alpine 3.21-2', 310), _fake_vm(3, 'public-Ubuntu 20.10-3', 164), ] _, get_names = self._configuration(vms, {310: 'public-Alpine 3.21'}) self.assertEqual(get_names.call_count, 1) @override_settings(UPDATED_TEMPLATES_DICT={}) def test_unreachable_template_pool_falls_back_to_vm_name(self): vms = [_fake_vm(26623, 'public-Ubuntu 20.10-26623', 164)] serializer = VirtualMachineSerializer(vms, many=True) with patch.object(OpenNebulaManager, '__init__', return_value=None), \ patch.object(OpenNebulaManager, 'get_template_names', side_effect=ConnectionRefusedError): configuration = serializer.child.get_configuration(vms[0]) self.assertEqual(configuration, 'Ubuntu 20.10') class OpenNebulaManagerGetTemplateTestCase(TestCase): @override_settings(UPDATED_TEMPLATES_DICT={164: 192}) def test_resolve_template_id(self): self.assertEqual(OpenNebulaManager.resolve_template_id(164), 192) self.assertEqual(OpenNebulaManager.resolve_template_id('164'), 192) self.assertEqual(OpenNebulaManager.resolve_template_id(310), 310) @override_settings(UPDATED_TEMPLATES_DICT={}) def test_missing_template_raises_template_not_found(self): pool = MagicMock() pool.get_by_id.side_effect = WrongIdError with patch.object(OpenNebulaManager, '__init__', return_value=None), \ patch.object(OpenNebulaManager, '_get_template_pool', return_value=pool): with self.assertRaises(TemplateNotFoundError) as ctx: OpenNebulaManager().get_template(164) self.assertEqual(ctx.exception.template_id, 164) @override_settings(UPDATED_TEMPLATES_DICT={}) def test_unreachable_pool_still_raises_connection_error(self): with patch.object(OpenNebulaManager, '__init__', return_value=None), \ patch.object(OpenNebulaManager, '_get_template_pool', side_effect=ConnectionRefusedError): with self.assertRaises(ConnectionRefusedError): OpenNebulaManager().get_template(164)