shaba/openuds
1
0
Fork 1
mirror of https://github.com/dkmstr/openuds.git synced 2025-01-10 01:17:59 +03:00
Code Issues Releases Wiki Activity

Refactorized name and get_type from module to mod_name and mod_type, much more convenient and may avoid future name clashs

Browse Source
This commit is contained in:
Adolfo Gómez García 2024-03-31 15:21:40 +02:00
parent d598dd5d0d
commit fc230d7205
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
32 changed files with 87 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/uds/REST/methods/authenticators.py
View File

@ -174,8 +174,8 @@ class Authenticators(ModelHandler):
'mfa_id': item.mfa.uuid if item.mfa else '',
'small_name': item.small_name,
'users_count': item.users.count(),
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'type_info': self.type_as_dict(type_),
'permission': permissions.effective_permissions(self._user, item),
}

4
server/src/uds/REST/methods/mfas.py
View File

@ -114,7 +114,7 @@ class MFA(ModelHandler):
'validity': item.validity,
'tags': [tag.tag for tag in item.tags.all()],
'comments': item.comments,
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'permission': permissions.effective_permissions(self._user, item),
}

4
server/src/uds/REST/methods/notifiers.py
View File

@ -122,7 +122,7 @@ class Notifiers(ModelHandler):
'enabled': item.enabled,
'tags': [tag.tag for tag in item.tags.all()],
'comments': item.comments,
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'permission': permissions.effective_permissions(self._user, item),
}

4
server/src/uds/REST/methods/osmanagers.py
View File

@ -70,8 +70,8 @@ class OsManagers(ModelHandler):
'name': osm.name,
'tags': [tag.tag for tag in osm.tags.all()],
'deployed_count': osm.deployedServices.count(),
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'servicesTypes': [
type_.servicesType
], # A list for backward compatibility. TODO: To be removed when admin interface is changed

8
server/src/uds/REST/methods/providers.py
View File

@ -88,8 +88,8 @@ class Providers(ModelHandler):
# Icon can have a lot of data (1-2 Kbytes), but it's not expected to have a lot of services providers, and even so, this will work fine
offers = [
{
'name': gettext(t.name()),
'type': t.get_type(),
'name': gettext(t.mod_name()),
'type': t.mod_type(),
'description': gettext(t.description()),
'icon': t.icon64().replace('\n', ''),
}
@ -106,8 +106,8 @@ class Providers(ModelHandler):
.count(),
'maintenance_mode': item.maintenance_mode,
'offers': offers,
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'comments': item.comments,
'permission': permissions.effective_permissions(self._user, item),
}

12
server/src/uds/REST/methods/services.py
View File

@ -101,7 +101,7 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
'tags': [tag.tag for tag in item.tags.all()],
'comments': item.comments,
'type': item.data_type,
'type_name': _(itemType.name()),
'type_name': _(itemType.mod_name()),
'deployed_services_count': item.deployedServices.count(),
'user_services_count': models.UserService.objects.filter(deployed_service__service=item)
.exclude(state__in=State.INFO_STATES)
@ -254,8 +254,8 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
if for_type is None:
offers = [
{
'name': _(t.name()),
'type': t.get_type(),
'name': _(t.mod_name()),
'type': t.mod_type(),
'description': _(t.description()),
'icon': t.icon64().replace('\n', ''),
}
@ -263,11 +263,11 @@ class Services(DetailHandler): # pylint: disable=too-many-public-methods
]
else:
for t in parent.get_type().get_provided_services():
if for_type == t.get_type():
if for_type == t.mod_type():
offers = [
{
'name': _(t.name()),
'type': t.get_type(),
'name': _(t.mod_name()),
'type': t.mod_type(),
'description': _(t.description()),
'icon': t.icon64().replace('\n', ''),
}

4
server/src/uds/REST/methods/transports.py
View File

@ -165,8 +165,8 @@ class Transports(ModelHandler):
'pools': pools,
'pools_count': len(pools),
'deployed_count': item.deployedServices.count(),
'type': type_.get_type(),
'type_name': type_.name(),
'type': type_.mod_type(),
'type_name': type_.mod_name(),
'protocol': type_.protocol,
'permission': permissions.effective_permissions(self._user, item),
}

2
server/src/uds/REST/methods/user_services.py
View File

@ -414,7 +414,7 @@ class Transports(DetailHandler):
'type': get_type(i),
'comments': i.comments,
'priority': i.priority,
'trans_type': _(i.get_type().name()),
'trans_type': _(i.get_type().mod_name()),
}
for i in parent.transports.all()
if get_type(i)

4
server/src/uds/REST/model/base.py
View File

@ -267,8 +267,8 @@ class BaseModelHandler(Handler):
Returns a dictionary describing the type (the name, the icon, description, etc...)
"""
res = types.rest.TypeInfo(
name=_(type_.name()),
type=type_.get_type(),
name=_(type_.mod_name()),
type=type_.mod_type(),
description=_(type_.description()),
icon=type_.icon64().replace('\n', ''),
extra=self.type_info(type_),

4
server/src/uds/core/module.py
View File

@ -214,7 +214,7 @@ class Module(UserInterface, Environmentable, Serializable, abc.ABC):
"""
@classmethod
def name(cls: type['Module']) -> str:
def mod_name(cls: type['Module']) -> str:
"""
Returns "translated" type_name, using gettext for transforming
cls.type_name
@ -228,7 +228,7 @@ class Module(UserInterface, Environmentable, Serializable, abc.ABC):
return _(cls.type_name)
@classmethod
def get_type(cls: type['Module']) -> str:
def mod_type(cls: type['Module']) -> str:
"""
Returns type_type

17
server/src/uds/core/services/generics/fixed/service.py
View File

@ -132,14 +132,14 @@ class FixedService(services.Service, abc.ABC): # pylint: disable=too-many-publi
raise exceptions.ui.ValidationError(gettext('We need at least a machine'))
# Remove machines not in values from "assigned" set
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
assigned_vms &= set(self.machines.as_list())
self.token.value = self.token.value.strip()
# Recover userservice
self.userservices_limit = len(self.machines.as_list())
@contextlib.contextmanager
def _assigned_machines_access(self) -> collections.abc.Generator[set[str], None, None]:
def _assigned_access(self) -> collections.abc.Generator[set[str], None, None]:
with self.storage.as_dict(atomic=True) as d:
machines: set[str] = d.get('vms', set())
initial_machines = machines.copy() # for comparison later
@ -163,24 +163,27 @@ class FixedService(services.Service, abc.ABC): # pylint: disable=too-many-publi
return
@abc.abstractmethod
def get_machine_name(self, vmid: str) -> str:
def get_name(self, vmid: str) -> str:
"""
Returns the machine name for the given vmid
"""
raise NotImplementedError()
@abc.abstractmethod
def get_and_assign_machine(self) -> str:
def get_and_assign(self) -> str:
"""
Gets automatically an assigns a machine
Returns the id of the assigned machine
Returns the id of the assigned machine, or raises an exception if no machine is available
Note:
This is used when deploying a machine for an user
"""
raise NotImplementedError()
# default implementation, should be sufficient for most cases
def remove_and_free_machine(self, vmid: str) -> str:
def remove_and_free(self, vmid: str) -> str:
try:
with self._assigned_machines_access() as assigned:
with self._assigned_access() as assigned:
assigned.remove(vmid)
return types.states.State.FINISHED
except Exception as e:

8
server/src/uds/core/services/generics/fixed/userservice.py
View File

@ -109,7 +109,7 @@ class FixedUserService(services.UserService, autoserializable.AutoSerializable,
if self._vmid:
try:
self.service().remove_and_free_machine(self._vmid)
self.service().remove_and_free(self._vmid)
self.service().process_snapshot(remove=True, userservice_instance=self)
self._vmid = ''
except Exception as e:
@ -151,7 +151,7 @@ class FixedUserService(services.UserService, autoserializable.AutoSerializable,
Deploys an service instance for an user.
"""
logger.debug('Deploying for user')
self._vmid = self.service().get_and_assign_machine()
self._vmid = self.service().get_and_assign()
# copy is needed to avoid modifying class var, and access using instance allowing to get, if provided, overriden queue
self._queue = self._create_queue.copy()
return self._execute_queue()
@ -224,7 +224,7 @@ class FixedUserService(services.UserService, autoserializable.AutoSerializable,
Deploys a machine from template for user/cache
"""
self._mac = self.service().get_first_network_mac(self._vmid) or ''
self._name = self.service().get_machine_name(self._vmid) or f'VM-{self._vmid}'
self._name = self.service().get_name(self._vmid) or f'VM-{self._vmid}'
@typing.final
def _snapshot_create(self) -> None:
@ -253,7 +253,7 @@ class FixedUserService(services.UserService, autoserializable.AutoSerializable,
"""
Removes the snapshot if needed and releases the machine again
"""
self.service().remove_and_free_machine(self._vmid)
self.service().remove_and_free(self._vmid)
# Check methods
def create_checker(self) -> types.states.TaskState:

2
server/src/uds/core/services/provider.py
View File

@ -144,7 +144,7 @@ class ServiceProvider(module.Module):
the type_type that Service has.
"""
for _type in cls.offers:
if _type.get_type() == type_name:
if _type.mod_type() == type_name:
return _type
return None

2
server/src/uds/core/services/provider_factory.py
View File

@ -60,7 +60,7 @@ class ServiceProviderFactory(factory.ModuleFactory[ServiceProvider]):
# We will check that if service provided by "provider" needs
# cache, but service do not provides publication_type,
# that service will not be registered and it will be informed
type_name = type_.get_type().lower()
type_name = type_.mod_type().lower()
if type_name in self.providers():
logger.debug('%s already registered as Service Provider', type_)

2
server/src/uds/core/services/publication.py
View File

@ -320,4 +320,4 @@ class Publication(Environmentable, Serializable):
"""
String method, mainly used for debugging purposes
"""
return f'{self.__class__.__name__}({self._service.name})'
return f'{self.__class__.__name__}({self._service.mod_name})'

2
server/src/uds/core/util/factory.py
View File

@ -93,4 +93,4 @@ class ModuleFactory(Factory[T]):
Inserts an object into the factory.
'''
# logger.debug('Adding %s as %s', type_.type(), type_.__module__)
super().register(type_.get_type().lower(), type_)
super().register(type_.mod_type().lower(), type_)

2
server/src/uds/models/service.py
View File

@ -134,7 +134,7 @@ class Service(ManagedObjectModel, TaggingMixin):
obj = sType(self.get_environment(), prov, values, uuid=self.uuid)
self.deserialize(obj, values)
else:
raise Exception(f'Service type of {self.data_type} is not recognized by provider {prov.name}')
raise Exception(f'Service type of {self.data_type} is not recognized by provider {prov.mod_name}')
self._cached_instance = obj

2
server/src/uds/models/service_pool.py
View File

@ -550,7 +550,7 @@ class ServicePool(UUIDModel, TaggingMixin):
"""
from uds.core import services # pylint: disable=import-outside-toplevel
servicesNotNeedingPub = [t.get_type() for t in services.factory().services_not_needing_publication()]
servicesNotNeedingPub = [t.mod_type() for t in services.factory().services_not_needing_publication()]
# Get services that HAS publications
query = (
ServicePool.objects.filter(

14
server/src/uds/services/OpenStack/service_fixed.py
View File

@ -149,7 +149,7 @@ class OpenStackServiceFixed(FixedService): # pylint: disable=too-many-public-me
server.id: server.name for server in self.api.list_servers() if not server.name.startswith('UDS-')
}
with self._assigned_machines_access() as assigned_servers:
with self._assigned_access() as assigned_servers:
return [
gui.choice_item(k, servers[k])
for k in self.machines.as_list()
@ -161,7 +161,7 @@ class OpenStackServiceFixed(FixedService): # pylint: disable=too-many-public-me
self, assignable_id: str, user: 'models.User', userservice_instance: 'services.UserService'
) -> types.states.TaskState:
openstack_userservice_instance = typing.cast(OpenStackUserServiceFixed, userservice_instance)
with self._assigned_machines_access() as assigned:
with self._assigned_access() as assigned:
if assignable_id not in assigned:
assigned.add(assignable_id)
return openstack_userservice_instance.assign(assignable_id)
@ -171,10 +171,10 @@ class OpenStackServiceFixed(FixedService): # pylint: disable=too-many-public-me
def process_snapshot(self, remove: bool, userservice_instance: FixedUserService) -> None:
return # No snapshots support
def get_and_assign_machine(self) -> str:
def get_and_assign(self) -> str:
found_vmid: typing.Optional[str] = None
try:
with self._assigned_machines_access() as assigned:
with self._assigned_access() as assigned:
for checking_vmid in self.machines.as_list():
if checking_vmid not in assigned: # Not already assigned
try:
@ -209,12 +209,12 @@ class OpenStackServiceFixed(FixedService): # pylint: disable=too-many-public-me
def get_guest_ip_address(self, vmid: str) -> str:
return self.api.get_server(vmid).addresses[0].ip
def get_machine_name(self, vmid: str) -> str:
def get_name(self, vmid: str) -> str:
return self.api.get_server(vmid).name
def remove_and_free_machine(self, vmid: str) -> str:
def remove_and_free(self, vmid: str) -> str:
try:
with self._assigned_machines_access() as assigned:
with self._assigned_access() as assigned:
assigned.remove(vmid)
return types.states.State.FINISHED
except Exception as e:

10
server/src/uds/services/Proxmox/service_fixed.py
View File

@ -128,7 +128,7 @@ class ProxmoxServiceFixed(FixedService): # pylint: disable=too-many-public-meth
for member in self.provider().get_pool_info(self.pool.value.strip(), retrieve_vm_names=True).members:
vms[member.vmid] = member.vmname
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
return [
gui.choice_item(k, vms[int(k)])
for k in self.machines.as_list()
@ -140,7 +140,7 @@ class ProxmoxServiceFixed(FixedService): # pylint: disable=too-many-public-meth
self, assignable_id: str, user: 'models.User', userservice_instance: 'services.UserService'
) -> types.states.TaskState:
proxmox_service_instance = typing.cast(ProxmoxUserServiceFixed, userservice_instance)
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
if assignable_id not in assigned_vms:
assigned_vms.add(assignable_id)
return proxmox_service_instance.assign(assignable_id)
@ -176,10 +176,10 @@ class ProxmoxServiceFixed(FixedService): # pylint: disable=too-many-public-meth
except Exception as e:
self.do_log(log.LogLevel.WARNING, 'Could not create SNAPSHOT for this VM. ({})'.format(e))
def get_and_assign_machine(self) -> str:
def get_and_assign(self) -> str:
found_vmid: typing.Optional[str] = None
try:
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
for checking_vmid in self.machines.as_list():
if checking_vmid not in assigned_vms: # Not already assigned
try:
@ -214,5 +214,5 @@ class ProxmoxServiceFixed(FixedService): # pylint: disable=too-many-public-meth
def get_guest_ip_address(self, vmid: str) -> str:
return self.provider().get_guest_ip_address(int(vmid))
def get_machine_name(self, vmid: str) -> str:
def get_name(self, vmid: str) -> str:
return self.provider().get_machine_info(int(vmid)).name or ''

14
server/src/uds/services/Xen/service_fixed.py
View File

@ -174,7 +174,7 @@ class XenFixedService(FixedService): # pylint: disable=too-many-public-methods
for machine in self.provider().get_machines_from_folder(self.folder.value, retrieve_names=True)
}
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
return [
gui.choice_item(k, vms[k])
for k in self.machines.as_list()
@ -186,7 +186,7 @@ class XenFixedService(FixedService): # pylint: disable=too-many-public-methods
self, assignable_id: str, user: 'models.User', userservice_instance: 'services.UserService'
) -> types.states.TaskState:
xen_userservice_instance = typing.cast(XenFixedUserService, userservice_instance)
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
if assignable_id not in assigned_vms:
assigned_vms.add(assignable_id)
return xen_userservice_instance.assign(assignable_id)
@ -222,9 +222,9 @@ class XenFixedService(FixedService): # pylint: disable=too-many-public-methods
except Exception as e:
self.do_log(log.LogLevel.WARNING, 'Could not create SNAPSHOT for this VM. ({})'.format(e))
def get_and_assign_machine(self) -> str:
def get_and_assign(self) -> str:
found_vmid: typing.Optional[str] = None
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
try:
for checking_vmid in self.machines.as_list():
if checking_vmid not in assigned_vms: # Not assigned
@ -258,12 +258,12 @@ class XenFixedService(FixedService): # pylint: disable=too-many-public-methods
def get_guest_ip_address(self, vmid: str) -> str:
return self.provider().get_first_ip(vmid)
def get_machine_name(self, vmid: str) -> str:
def get_name(self, vmid: str) -> str:
return self.provider().get_machine_name(vmid)
def remove_and_free_machine(self, vmid: str) -> str:
def remove_and_free(self, vmid: str) -> str:
try:
with self._assigned_machines_access() as assigned_vms:
with self._assigned_access() as assigned_vms:
assigned_vms.remove(vmid)
return types.states.State.FINISHED
except Exception as e:

1
server/tests/REST/actor/test_initialize.py
View File

@ -29,7 +29,6 @@
@author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
import collections.abc
import functools
import logging

3
server/tests/REST/auths/test_users_groups_users.py
View File

@ -28,7 +28,6 @@
"""
@author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import time
import typing
import collections.abc
import functools
@ -83,7 +82,7 @@ class UsersTest(rest.test.RESTActorTestCase):
fields: list[collections.abc.Mapping[str, typing.Any]] = tableinfo['fields']
self.assertTrue(
functools.reduce(
lambda x, y: x and y,
lambda x, y: x and y, # pyright: ignore
map(
lambda f: next(iter(f.keys()))
in (

7
server/tests/REST/servers/test_events_log.py
View File

@ -28,20 +28,15 @@
"""
@author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
import collections.abc
import logging
from unittest import mock
from uds.core.util import log
from ...utils import rest, random_ip_v4, random_ip_v6, random_mac
from ...utils import rest
from ...fixtures import servers as servers_fixtures
if typing.TYPE_CHECKING:
from ...utils.test import UDSHttpResponse
logger = logging.getLogger(__name__)

4
server/tests/REST/test_system.py
View File

@ -28,13 +28,9 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
import collections.abc
import datetime
import logging
from uds.core.consts.actor import UNMANAGED
from uds.core.consts.actor import MANAGED
from uds.core.util.stats import counters
from ..utils import rest

6
server/tests/core/services/generics/fixtures.py
View File

@ -96,11 +96,11 @@ class FixedTestingService(fixed_service.FixedService):
userservice_instance._queue.insert(0, types.services.FixedOperation.NOP)
self.first_process_called = True
def get_machine_name(self, vmid: str) -> str:
def get_name(self, vmid: str) -> str:
self.mock.get_machine_name(vmid)
return f'Machine {vmid}'
def get_and_assign_machine(self) -> str:
def get_and_assign(self) -> str:
self.mock.get_and_assign_machine()
if self.available_machines_number <= 0:
raise Exception('No machine available')
@ -108,7 +108,7 @@ class FixedTestingService(fixed_service.FixedService):
self.assigned_machine = 'assigned'
return self.assigned_machine
def remove_and_free_machine(self, vmid: str) -> str:
def remove_and_free(self, vmid: str) -> str:
self.mock.remove_and_free_machine(vmid)
self.assigned_machine = ''
return types.states.TaskState.FINISHED

6
server/tests/services/openstack/test_service_fixed.py
View File

@ -40,7 +40,7 @@ from . import fixtures
from ...utils.test import UDSTransactionTestCase
class TestProxmovFixedService(UDSTransactionTestCase):
class TestOpenstackFixedService(UDSTransactionTestCase):
def test_service_fixed(self) -> None:
"""
@ -101,7 +101,7 @@ class TestProxmovFixedService(UDSTransactionTestCase):
api.get_server.reset_mock()
# Now get_and_assign_machine as much as remaining machines
for _ in range(remaining):
vm = service.get_and_assign_machine()
vm = service.get_and_assign()
self.assertIn(vm, assignables)
# enumarate_assignables should return an empty list now
@ -111,4 +111,4 @@ class TestProxmovFixedService(UDSTransactionTestCase):
self.assertEqual(api.get_server.call_count, remaining)
# And a new try, should raise an exception
self.assertRaises(Exception, service.get_and_assign_machine)
self.assertRaises(Exception, service.get_and_assign)

6
server/tests/services/openstack/test_userservice_fixed.py
View File

@ -44,7 +44,7 @@ from uds.services.OpenStack.openstack import types as openstack_types
# We use transactions on some related methods (storage access, etc...)
class TestOpenstackFixedService(UDSTransactionTestCase):
class TestOpenstackFixedUserService(UDSTransactionTestCase):
def test_userservice_fixed_user(self) -> None:
"""
@ -76,7 +76,7 @@ class TestOpenstackFixedService(UDSTransactionTestCase):
userservice_db.set_in_use.assert_called_once_with(True)
# vmid should have been assigned, so it must be in the assigned machines
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual({userservice._vmid}, assigned_machines)
# Now, let's release the service
@ -90,7 +90,7 @@ class TestOpenstackFixedService(UDSTransactionTestCase):
self.assertEqual(state, types.states.TaskState.FINISHED)
# must be empty now
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual(set(), assigned_machines)
# set_ready, machine is "stopped" in this test, so must return RUNNING

2
server/tests/services/proxmox/test_publication.py
View File

@ -43,7 +43,7 @@ from ...utils.generators import limited_iterator
# USe transactional, used by publication access to db on "removal"
class TestProxmovPublication(UDSTransactionTestCase):
class TestProxmoxPublication(UDSTransactionTestCase):
def test_publication(self) -> None:
with fixtures.patched_provider() as provider:

14
server/tests/services/proxmox/test_service_fixed.py
View File

@ -102,10 +102,10 @@ class TestProxmovFixedService(UDSTransactionTestCase):
# Get and assign machine
# will try to assign FIRST FREE machine, that is the second one
vmid2: str = typing.cast(list[str], fixtures.SERVICE_FIXED_VALUES_DICT['machines'])[1]
self.assertEqual(service.get_and_assign_machine(), vmid2)
self.assertEqual(service.get_and_assign(), vmid2)
# Now two machies should be assigned
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual(assigned_machines, set([vmid, vmid2]))
def test_service_methods_2(self) -> None:
@ -113,7 +113,7 @@ class TestProxmovFixedService(UDSTransactionTestCase):
service = fixtures.create_service_fixed(provider=provider)
# Get machine name
self.assertEqual(service.get_machine_name('1'), fixtures.VMS_INFO[0].name)
self.assertEqual(service.get_name('1'), fixtures.VMS_INFO[0].name)
# Get first network mac
self.assertEqual(
@ -125,13 +125,13 @@ class TestProxmovFixedService(UDSTransactionTestCase):
# Remove and free machine
# Fist, assign a machine
vmid = service.get_and_assign_machine()
with service._assigned_machines_access() as assigned_machines:
vmid = service.get_and_assign()
with service._assigned_access() as assigned_machines:
self.assertEqual(assigned_machines, set([vmid]))
# And now free it
self.assertEqual(service.remove_and_free_machine(vmid), types.states.State.FINISHED)
with service._assigned_machines_access() as assigned_machines:
self.assertEqual(service.remove_and_free(vmid), types.states.State.FINISHED)
with service._assigned_access() as assigned_machines:
self.assertEqual(assigned_machines, set())
def test_process_snapshot(self) -> None:

6
server/tests/services/proxmox/test_userservice_fixed.py
View File

@ -53,7 +53,7 @@ class TestProxmoxFixedUserService(UDSTransactionTestCase):
with fixtures.patched_provider() as provider:
service = fixtures.create_service_fixed(provider=provider)
userservice = fixtures.create_userservice_fixed(service=service)
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual(assigned_machines, set())
# patch userservice db_obj() method to return a mock
@ -78,7 +78,7 @@ class TestProxmoxFixedUserService(UDSTransactionTestCase):
userservice_db.set_in_use.assert_called_once_with(True)
# vmid should have been assigned, so it must be in the assigned machines
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual({userservice._vmid}, assigned_machines)
# Now, let's release the service
@ -92,7 +92,7 @@ class TestProxmoxFixedUserService(UDSTransactionTestCase):
self.assertEqual(state, types.states.TaskState.FINISHED)
# must be empty now
with service._assigned_machines_access() as assigned_machines:
with service._assigned_access() as assigned_machines:
self.assertEqual(assigned_machines, set())
# set_ready, machine is "started", set by mock fixture with the start invokation

2
server/tests/services/proxmox/test_userservice_linked.py
View File

@ -44,7 +44,7 @@ from ...utils.generators import limited_iterator
# We use transactions on some related methods (storage access, etc...)
class TestProxmovLinkedUserService(UDSTransactionTestCase):
class TestProxmoxLinkedUserService(UDSTransactionTestCase):
def setUp(self) -> None:
fixtures.set_all_vm_state('stopped')