mirror of
https://github.com/dkmstr/openuds.git
synced 2025-03-12 04:58:34 +03:00
Added required methods to allow fixedservice and finished client mock and data
This commit is contained in:
Adolfo Gómez García
parent
cbacc20909
commit
ce1d1eba19
@ -38,6 +38,7 @@ DEFAULT_CACHE_TIMEOUT: typing.Final[int] = 60 * 3 # 3 minutes
|
||||
LONG_CACHE_TIMEOUT: typing.Final[int] = DEFAULT_CACHE_TIMEOUT * 20 # 1 hour
|
||||
EXTREME_CACHE_TIMEOUT: typing.Final[int] = LONG_CACHE_TIMEOUT * 24 # 1 day
|
||||
SHORT_CACHE_TIMEOUT: typing.Final[int] = DEFAULT_CACHE_TIMEOUT // 3 # 1 minute
|
||||
SMALLEST_CACHE_TIMEOUT: typing.Final[int] = 3 # 3 seconds
|
||||
|
||||
# Used to mark a cache as not found
|
||||
# use "cache.get(..., default=CACHE_NOT_FOUND)" to check if a cache is non existing instead of real None value
|
||||
|
||||
@ -162,7 +162,7 @@ class Client:
|
||||
for vm in typing.cast(list[typing.Any], self.api.system_service().vms_service().list())
|
||||
]
|
||||
|
||||
@decorators.cached(prefix='o-vm', timeout=3, key_helper=_key_helper)
|
||||
@decorators.cached(prefix='o-vm', timeout=consts.cache.SMALLEST_CACHE_TIMEOUT, key_helper=_key_helper)
|
||||
def get_machine_info(self, machine_id: str, **kwargs: typing.Any) -> ov_types.VMInfo:
|
||||
with _access_lock():
|
||||
try:
|
||||
@ -327,6 +327,7 @@ class Client:
|
||||
)
|
||||
)
|
||||
|
||||
@decorators.cached(prefix='o-templates', timeout=consts.cache.DEFAULT_CACHE_TIMEOUT, key_helper=_key_helper)
|
||||
def get_template_info(self, template_id: str) -> ov_types.TemplateInfo:
|
||||
"""
|
||||
Returns the template info for the given template id
|
||||
@ -399,9 +400,7 @@ class Client:
|
||||
usb=usb,
|
||||
) # display=display,
|
||||
|
||||
return ov_types.VMInfo.from_data(
|
||||
self.api.system_service().vms_service().add(par)
|
||||
)
|
||||
return ov_types.VMInfo.from_data(self.api.system_service().vms_service().add(par))
|
||||
|
||||
def remove_template(self, template_id: str) -> None:
|
||||
"""
|
||||
@ -413,6 +412,71 @@ class Client:
|
||||
self.api.system_service().templates_service().service(template_id).remove()
|
||||
# This returns nothing, if it fails it raises an exception
|
||||
|
||||
@decorators.cached(
|
||||
prefix='o-templates', timeout=consts.cache.SMALLEST_CACHE_TIMEOUT, key_helper=_key_helper
|
||||
)
|
||||
def list_snapshots(self, machine_id: str) -> list[ov_types.SnapshotInfo]:
|
||||
"""
|
||||
Lists the snapshots of the given machine
|
||||
"""
|
||||
with _access_lock():
|
||||
vm_service: typing.Any = self.api.system_service().vms_service().service(machine_id)
|
||||
|
||||
if vm_service.get() is None:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
return [
|
||||
ov_types.SnapshotInfo.from_data(s)
|
||||
for s in typing.cast(list[typing.Any], vm_service.snapshots_service().list())
|
||||
]
|
||||
|
||||
@decorators.cached(prefix='o-snapshot', timeout=consts.cache.SMALLEST_CACHE_TIMEOUT, key_helper=_key_helper)
|
||||
def get_snapshot_info(self, machine_id: str, snapshot_id: str) -> ov_types.SnapshotInfo:
|
||||
"""
|
||||
Returns the snapshot info for the given snapshot id
|
||||
"""
|
||||
with _access_lock():
|
||||
vm_service: typing.Any = self.api.system_service().vms_service().service(machine_id)
|
||||
|
||||
if vm_service.get() is None:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
return ov_types.SnapshotInfo.from_data(
|
||||
typing.cast(
|
||||
ovirtsdk4.types.Snapshot,
|
||||
vm_service.snapshots_service().service(snapshot_id).get(),
|
||||
)
|
||||
)
|
||||
|
||||
def create_snapshot(
|
||||
self, machine_id: str, snapshot_name: str, snapshot_description: str
|
||||
) -> ov_types.SnapshotInfo:
|
||||
"""
|
||||
Creates a snapshot of the machine with the given name and description
|
||||
"""
|
||||
with _access_lock():
|
||||
vm_service: typing.Any = self.api.system_service().vms_service().service(machine_id)
|
||||
|
||||
if vm_service.get() is None:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
snapshot = ovirtsdk4.types.Snapshot(
|
||||
name=snapshot_name, description=snapshot_description, persist_memorystate=True
|
||||
)
|
||||
return ov_types.SnapshotInfo.from_data(vm_service.snapshots_service().add(snapshot))
|
||||
|
||||
def remove_snapshot(self, machine_id: str, snapshot_id: str) -> None:
|
||||
"""
|
||||
Removes the snapshot with the given id
|
||||
"""
|
||||
with _access_lock():
|
||||
vm_service: typing.Any = self.api.system_service().vms_service().service(machine_id)
|
||||
|
||||
if vm_service.get() is None:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
vm_service.snapshots_service().service(snapshot_id).remove()
|
||||
|
||||
def start_machine(self, machine_id: str) -> None:
|
||||
"""
|
||||
Tries to start a machine. No check is done, it is simply requested to oVirt.
|
||||
@ -434,10 +498,10 @@ class Client:
|
||||
|
||||
def stop_machine(self, machine_id: str) -> None:
|
||||
"""
|
||||
Tries to start a machine. No check is done, it is simply requested to oVirt
|
||||
Tries to stop a machine. No check is done, it is simply requested to oVirt
|
||||
|
||||
Args:
|
||||
machineId: Id of the machine
|
||||
machine_id: Id of the machine
|
||||
|
||||
Returns:
|
||||
"""
|
||||
@ -448,13 +512,30 @@ class Client:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
vm_service.stop()
|
||||
|
||||
def shutdown_machine(self, machine_id: str) -> None:
|
||||
"""
|
||||
Tries to shutdown a machine. No check is done, it is simply requested to oVirt
|
||||
|
||||
Args:
|
||||
machine_id: Id of the machine
|
||||
|
||||
Returns:
|
||||
"""
|
||||
with _access_lock():
|
||||
vm_service: typing.Any = self.api.system_service().vms_service().service(machine_id)
|
||||
|
||||
if vm_service.get() is None:
|
||||
raise Exception('Machine not found')
|
||||
|
||||
vm_service.shutdown()
|
||||
|
||||
def suspend_machine(self, machine_id: str) -> None:
|
||||
"""
|
||||
Tries to start a machine. No check is done, it is simply requested to oVirt
|
||||
Tries to suspend a machine. No check is done, it is simply requested to oVirt
|
||||
|
||||
Args:
|
||||
machineId: Id of the machine
|
||||
machine_id: Id of the machine
|
||||
|
||||
Returns:
|
||||
"""
|
||||
@ -509,7 +590,9 @@ class Client:
|
||||
vmu = ovirtsdk4.types.Vm(usb=usb)
|
||||
vms.update(vmu)
|
||||
|
||||
def get_console_connection_info(self, machine_id: str) -> typing.Optional[types.services.ConsoleConnectionInfo]:
|
||||
def get_console_connection_info(
|
||||
self, machine_id: str
|
||||
) -> typing.Optional[types.services.ConsoleConnectionInfo]:
|
||||
"""
|
||||
Gets the connetion info for the specified machine
|
||||
"""
|
||||
|
||||
@ -88,6 +88,41 @@ class TemplateStatus(enum.StrEnum):
|
||||
return TemplateStatus.ILLEGAL
|
||||
|
||||
|
||||
class SnapshotStatus(enum.StrEnum):
|
||||
# Adapted from ovirtsdk4
|
||||
IN_PREVIEW = 'in_preview'
|
||||
LOCKED = 'locked'
|
||||
OK = 'ok'
|
||||
|
||||
# Custom value to represent the snapshot is missing
|
||||
UNKNOWN = 'unknown'
|
||||
|
||||
@staticmethod
|
||||
def from_str(status: str) -> 'SnapshotStatus':
|
||||
try:
|
||||
return SnapshotStatus(status)
|
||||
except ValueError:
|
||||
return SnapshotStatus.UNKNOWN
|
||||
|
||||
|
||||
class SnapshotType(enum.StrEnum):
|
||||
# Adapted from ovirtsdk4
|
||||
ACTIVE = 'active'
|
||||
PREVIEW = 'preview'
|
||||
REGULAR = 'regular'
|
||||
STATELESS = 'stateless'
|
||||
|
||||
# Custom value to represent the snapshot is missing
|
||||
UNKNOWN = 'unknown'
|
||||
|
||||
@staticmethod
|
||||
def from_str(status: str) -> 'SnapshotType':
|
||||
try:
|
||||
return SnapshotType(status)
|
||||
except ValueError:
|
||||
return SnapshotType.UNKNOWN
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class StorageInfo:
|
||||
id: str
|
||||
@ -197,3 +232,28 @@ class TemplateInfo:
|
||||
@staticmethod
|
||||
def missing() -> 'TemplateInfo':
|
||||
return TemplateInfo(id='', name='', description='', cluster_id='', status=TemplateStatus.UNKNOWN)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SnapshotInfo:
|
||||
id: str
|
||||
name: typing.Optional[str ]
|
||||
description: str
|
||||
status: SnapshotStatus
|
||||
type: SnapshotType
|
||||
|
||||
@staticmethod
|
||||
def from_data(snapshot: typing.Any) -> 'SnapshotInfo':
|
||||
return SnapshotInfo(
|
||||
id=snapshot.id,
|
||||
name=snapshot.name,
|
||||
description=snapshot.description,
|
||||
status=SnapshotStatus.from_str(snapshot.snapshot_status.value),
|
||||
type=SnapshotType.from_str(snapshot.snapshot_type.value),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def missing() -> 'SnapshotInfo':
|
||||
return SnapshotInfo(
|
||||
id='', name='', description='', status=SnapshotStatus.UNKNOWN, type=SnapshotType.UNKNOWN
|
||||
)
|
||||
|
||||
@ -121,7 +121,7 @@ VMS_INFO: list[ov_types.VMInfo] = [
|
||||
|
||||
TEMPLATES_INFO: list[ov_types.TemplateInfo] = [
|
||||
ov_types.TemplateInfo(
|
||||
id=f'tid-{i}',
|
||||
id=f'teid-{i}',
|
||||
name=f'template-{i}',
|
||||
description=f'Template {i} description',
|
||||
cluster_id=from_list(CLUSTERS_INFO, i // 8).id,
|
||||
@ -130,6 +130,17 @@ TEMPLATES_INFO: list[ov_types.TemplateInfo] = [
|
||||
for i in range(16)
|
||||
]
|
||||
|
||||
SNAPSHOTS_INFO: list[ov_types.SnapshotInfo] = [
|
||||
ov_types.SnapshotInfo(
|
||||
id=f'snid-{i}',
|
||||
name=f'snapshot-{i}',
|
||||
description='Active VM',
|
||||
status=ov_types.SnapshotStatus.OK,
|
||||
type=ov_types.SnapshotType.ACTIVE,
|
||||
)
|
||||
for i in range(8)
|
||||
]
|
||||
|
||||
CONSOLE_CONNECTION_INFO: types.services.ConsoleConnectionInfo = types.services.ConsoleConnectionInfo(
|
||||
type='spice',
|
||||
address=GUEST_IP_ADDRESS,
|
||||
@ -146,6 +157,7 @@ CONSOLE_CONNECTION_INFO: types.services.ConsoleConnectionInfo = types.services.C
|
||||
# Methods that returns None or "internal" methods are not tested
|
||||
# The idea behind this is to allow testing the provider, service and deployment classes
|
||||
# without the need of a real OpenStack environment
|
||||
# all methods that returns None are provided by the auto spec mock
|
||||
CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
AutoSpecMethodInfo(client.Client.list_machines, return_value=VMS_INFO),
|
||||
AutoSpecMethodInfo(
|
||||
@ -180,11 +192,19 @@ CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
client.Client.deploy_from_template,
|
||||
return_value=lambda *args, **kwargs: random.choice(VMS_INFO), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(client.Client.list_snapshots, return_value=SNAPSHOTS_INFO),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_snapshot_info,
|
||||
return_value=lambda snapshot_id, **kwargs: get_id(SNAPSHOTS_INFO, snapshot_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_console_connection_info,
|
||||
return_value=CONSOLE_CONNECTION_INFO,
|
||||
),
|
||||
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.create_snapshot,
|
||||
return_value=lambda *args, **kwargs: random.choice(SNAPSHOTS_INFO), # pyright: ignore
|
||||
),
|
||||
# connect returns None
|
||||
# Test method
|
||||
# AutoSpecMethodInfo(client.Client.list_projects, returns=True),
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user