1
0
mirror of https://github.com/dkmstr/openuds.git synced 2025-01-03 01:17:56 +03:00

fixed specialization and specialization tests

This commit is contained in:
Adolfo Gómez García 2024-02-16 21:56:53 +01:00
parent 042a27737f
commit 8d26fff400
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
7 changed files with 36 additions and 16 deletions

View File

@ -122,6 +122,7 @@ class ServicesPools(ModelHandler):
('getFallbackAccess', True),
('actionsList', True),
('listAssignables', True),
('list_assignables', True),
('createFromAssignable', True),
]
@ -647,10 +648,14 @@ class ServicesPools(ModelHandler):
)
return validActions
# Deprecated, use list_assignables
def listAssignables(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, ServicePool)
service = item.service.get_instance() # type: ignore
return [gui.choice_item(i[0], i[1]) for i in service.enumerate_assignables()]
return list(service.enumerate_assignables())
def list_assignables(self, item: 'Model') -> typing.Any:
return self.listAssignables(item)
def createFromAssignable(self, item: 'Model') -> typing.Any:
item = ensure.is_instance(item, ServicePool)

View File

@ -328,13 +328,14 @@ class Service(Module):
"""
return typing.cast('UniqueNameGenerator', self.id_generator('name'))
def enumerate_assignables(self) -> collections.abc.Iterable[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
"""
If overrided, will provide list of assignables elements, so we can "add" an element manually to the list of assigned user services
If not overriden, means that it cannot assign manually
Returns:
list[tuple[str, str]] -- List of asignables services, first element is id, second is name of the element
collections.abc.Iterable[types.ui.ChoiceItem]: List of asignables services (as ChoiceItem)
"""
return []

View File

@ -186,7 +186,7 @@ class FixedService(services.Service, abc.ABC): # pylint: disable=too-many-publi
raise NotImplementedError()
@abc.abstractmethod
def enumerate_assignables(self) -> list[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
"""
Returns a list of tuples with the id and the name of the assignables
"""

View File

@ -318,9 +318,9 @@ class IPMachinesService(IPServiceBase):
except Exception:
logger.exception("Exception at getUnassignedMachine")
def enumerate_assignables(self) -> typing.List[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
return [
(f'{host.host}|{host.mac}', host.host)
gui.choice_item(f'{host.host}|{host.mac}', host.host)
for host in self.hosts
if self.storage.read_from_db(host.host) is None
]

View File

@ -28,6 +28,7 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import collections.abc
import logging
import typing
@ -150,7 +151,7 @@ class ProxmoxFixedService(FixedService): # pylint: disable=too-many-public-meth
def is_avaliable(self) -> bool:
return self.parent().is_available()
def enumerate_assignables(self) -> list[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
# Obtain machines names and ids for asignables
vms: dict[int, str] = {}
@ -158,7 +159,7 @@ class ProxmoxFixedService(FixedService): # pylint: disable=too-many-public-meth
vms[member.vmid] = member.vmname
assigned_vms = self._get_assigned_machines()
return [(k, vms.get(int(k), 'Unknown!')) for k in self.machines.as_list() if int(k) not in assigned_vms]
return [gui.choice_item(k, vms.get(int(k), 'Unknown!')) for k in self.machines.as_list() if int(k) not in assigned_vms]
def assign_from_assignables(
self, assignable_id: str, user: 'models.User', user_deployment: 'services.UserService'

View File

@ -30,6 +30,7 @@ Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import logging
import typing
import collections.abc
from django.utils.translation import gettext
from django.utils.translation import gettext_noop as _
@ -179,12 +180,12 @@ class XenFixedService(FixedService): # pylint: disable=too-many-public-methods
def is_avaliable(self) -> bool:
return self.parent().is_available()
def enumerate_assignables(self) -> list[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
# Obtain machines names and ids for asignables
vms: dict[int, str] = {}
assigned_vms = self._get_assigned_machines()
return [(k, vms.get(int(k), 'Unknown!')) for k in self.machines.as_list() if int(k) not in assigned_vms]
return [gui.choice_item(k, vms.get(int(k), 'Unknown!')) for k in self.machines.as_list() if int(k) not in assigned_vms]
def assign_from_assignables(
self, assignable_id: str, user: 'models.User', user_deployment: 'services.UserService'

View File

@ -31,16 +31,16 @@
"""
import dataclasses
import typing
import collections.abc
from unittest import mock
from regex import F
from uds import models
from uds.core import services, types
from uds.core.services.specializations.fixed_machine import (
fixed_service,
fixed_userservice,
)
from uds.core.ui.user_interface import gui
from ....utils.test import UDSTestCase
@ -226,6 +226,14 @@ class FixedServiceTest(UDSTestCase):
prov, service, userservice = self.create_elements()
self.check_iterations(service, userservice, EXPECTED_DEPLOY_ITERATIONS_INFO, removal=False)
def test_fixed_service_deploy_no_machine(self) -> None:
prov, service, userservice = self.create_elements()
service.available_machines_number = 2
self.deploy_service(service, userservice) # Should be deployed without issues
self.deploy_service(service, userservice) # Should be deployed without issues, 2nd time
# And now, should fail to deploy again
self.assertRaises(Exception, self.deploy_service, service, userservice)
def test_fixed_service_removal(self) -> None:
prov, service, userservice = self.create_elements()
@ -269,6 +277,7 @@ class FixedTestingService(fixed_service.FixedService):
user_service_type = FixedTestingUserService
first_process_called = False
available_machines_number = 1
mock: 'mock.Mock' = mock.MagicMock()
@ -289,6 +298,9 @@ class FixedTestingService(fixed_service.FixedService):
def get_and_assign_machine(self) -> str:
self.mock.get_and_assign_machine()
if self.available_machines_number <= 0:
raise Exception('No machine available')
self.available_machines_number -= 1
self.assigned_machine = 'assigned'
return self.assigned_machine
@ -305,15 +317,15 @@ class FixedTestingService(fixed_service.FixedService):
self.mock.get_guest_ip_address(vmid)
return '10.0.0.10'
def enumerate_assignables(self) -> list[tuple[str, str]]:
def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]:
"""
Returns a list of tuples with the id and the name of the assignables
"""
self.mock.enumerate_assignables()
return [
('1', 'Machine 1'),
('2', 'Machine 2'),
('3', 'Machine 3'),
gui.choice_item('1', 'Machine 1'),
gui.choice_item('2', 'Machine 2'),
gui.choice_item('3', 'Machine 3'),
]
def assign_from_assignables(