mirror of
https://github.com/dkmstr/openuds.git
synced 2025-02-02 09:47:13 +03:00
Done ovirt provider tests
This commit is contained in:
Adolfo Gómez García
parent
a0cb8d4778
commit
3e8405482e
@ -242,72 +242,72 @@ def get_id(iterable: typing.Iterable[T], id: str) -> T:
|
||||
# The idea behind this is to allow testing the provider, service and deployment classes
|
||||
# without the need of a real OpenStack environment
|
||||
CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_flavors, return_value=FLAVORS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_flavors, returns=FLAVORS_LIST),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.list_availability_zones, return_value=AVAILABILITY_ZONES_LIST
|
||||
openstack_client.OpenstackClient.list_availability_zones, returns=AVAILABILITY_ZONES_LIST
|
||||
),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_projects, return_value=PROJECTS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_regions, return_value=REGIONS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_servers, return_value=SERVERS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_images, return_value=IMAGES_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_volume_types, return_value=VOLUMES_TYPE_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_volumes, return_value=VOLUMES_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_projects, returns=PROJECTS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_regions, returns=REGIONS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_servers, returns=SERVERS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_images, returns=IMAGES_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_volume_types, returns=VOLUMES_TYPE_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_volumes, returns=VOLUMES_LIST),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.list_volume_snapshots, return_value=VOLUME_SNAPSHOTS_LIST
|
||||
openstack_client.OpenstackClient.list_volume_snapshots, returns=VOLUME_SNAPSHOTS_LIST
|
||||
),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_networks, return_value=NETWORKS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_ports, return_value=PORTS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_networks, returns=NETWORKS_LIST),
|
||||
AutoSpecMethodInfo(openstack_client.OpenstackClient.list_ports, returns=PORTS_LIST),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.list_security_groups, return_value=SECURITY_GROUPS_LIST
|
||||
openstack_client.OpenstackClient.list_security_groups, returns=SECURITY_GROUPS_LIST
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.get_server,
|
||||
return_value=lambda server_id: get_id(SERVERS_LIST, server_id), # pyright: ignore
|
||||
returns=lambda server_id: get_id(SERVERS_LIST, server_id), # pyright: ignore
|
||||
), # pyright: ignore
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.get_volume,
|
||||
return_value=lambda volume_id: get_id(VOLUMES_LIST, volume_id), # pyright: ignore
|
||||
returns=lambda volume_id: get_id(VOLUMES_LIST, volume_id), # pyright: ignore
|
||||
), # pyright: ignore
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.get_volume_snapshot,
|
||||
return_value=lambda snapshot_id: get_id(VOLUME_SNAPSHOTS_LIST, snapshot_id), # pyright: ignore
|
||||
returns=lambda snapshot_id: get_id(VOLUME_SNAPSHOTS_LIST, snapshot_id), # pyright: ignore
|
||||
), # pyright: ignore
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.update_snapshot,
|
||||
return_value=lambda snapshot_id, name, description: get_id( # pyright: ignore
|
||||
returns=lambda snapshot_id, name, description: get_id( # pyright: ignore
|
||||
VOLUME_SNAPSHOTS_LIST, snapshot_id # pyright: ignore
|
||||
),
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.create_volume_snapshot,
|
||||
return_value=lambda volume_id, name, description: random.choice( # pyright: ignore
|
||||
returns=lambda volume_id, name, description: random.choice( # pyright: ignore
|
||||
VOLUME_SNAPSHOTS_LIST,
|
||||
),
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.create_volume_from_snapshot,
|
||||
return_value=lambda snapshot_id, name, description: get_id( # pyright: ignore
|
||||
returns=lambda snapshot_id, name, description: get_id( # pyright: ignore
|
||||
VOLUMES_LIST, f'vid{len(VOLUMES_LIST) + 1}'
|
||||
),
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.create_server_from_snapshot,
|
||||
return_value=lambda *args, **kwargs: random.choice(SERVERS_LIST), # pyright: ignore
|
||||
returns=lambda *args, **kwargs: random.choice(SERVERS_LIST), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.test_connection,
|
||||
return_value=True,
|
||||
returns=True,
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
openstack_client.OpenstackClient.is_available,
|
||||
return_value=True,
|
||||
returns=True,
|
||||
),
|
||||
# connect returns None
|
||||
# Test method
|
||||
# AutoSpecMethodInfo(client.Client.list_projects, return_value=True),
|
||||
# AutoSpecMethodInfo(client.Client.list_projects, returns=True),
|
||||
# AutoSpecMethodInfo(
|
||||
# client.ProxmoxClient.get_node_stats,
|
||||
# return_value=lambda node, **kwargs: next(filter(lambda n: n.name == node, NODE_STATS)), # pyright: ignore
|
||||
# returns=lambda node, **kwargs: next(filter(lambda n: n.name == node, NODE_STATS)), # pyright: ignore
|
||||
# ),
|
||||
]
|
||||
|
||||
|
||||
@ -159,52 +159,52 @@ CONSOLE_CONNECTION_INFO: types.services.ConsoleConnectionInfo = types.services.C
|
||||
# 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.test, return_value=True),
|
||||
AutoSpecMethodInfo(client.Client.list_machines, return_value=VMS_INFO),
|
||||
AutoSpecMethodInfo(client.Client.test, returns=True),
|
||||
AutoSpecMethodInfo(client.Client.list_machines, returns=VMS_INFO),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_machine_info,
|
||||
lambda machine_id, **kwargs: get_id(VMS_INFO, machine_id), # pyright: ignore
|
||||
returns=lambda machine_id, **kwargs: get_id(VMS_INFO, machine_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(client.Client.list_clusters, return_value=CLUSTERS_INFO),
|
||||
AutoSpecMethodInfo(client.Client.list_clusters, returns=CLUSTERS_INFO),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_cluster_info,
|
||||
lambda cluster_id, **kwargs: get_id(CLUSTERS_INFO, cluster_id), # pyright: ignore
|
||||
returns=lambda cluster_id, **kwargs: get_id(CLUSTERS_INFO, cluster_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_datacenter_info,
|
||||
lambda datacenter_id, **kwargs: get_id( # pyright: ignore
|
||||
returns=lambda datacenter_id, **kwargs: get_id( # pyright: ignore
|
||||
DATACENTERS_INFO,
|
||||
datacenter_id, # pyright: ignore
|
||||
),
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_storage_info,
|
||||
lambda storage_id, **kwargs: get_id(STORAGES_INFO, storage_id), # pyright: ignore
|
||||
returns=lambda storage_id, **kwargs: get_id(STORAGES_INFO, storage_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.create_template,
|
||||
return_value=lambda *args, **kwargs: random.choice(TEMPLATES_INFO), # pyright: ignore
|
||||
returns=lambda *args, **kwargs: random.choice(TEMPLATES_INFO), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_template_info,
|
||||
lambda template_id, **kwargs: get_id(TEMPLATES_INFO, template_id), # pyright: ignore
|
||||
returns=lambda template_id, **kwargs: get_id(TEMPLATES_INFO, template_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.deploy_from_template,
|
||||
return_value=lambda *args, **kwargs: random.choice(VMS_INFO), # pyright: ignore
|
||||
returns=lambda *args, **kwargs: random.choice(VMS_INFO), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(client.Client.list_snapshots, return_value=SNAPSHOTS_INFO),
|
||||
AutoSpecMethodInfo(client.Client.list_snapshots, returns=SNAPSHOTS_INFO),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_snapshot_info,
|
||||
return_value=lambda snapshot_id, **kwargs: get_id(SNAPSHOTS_INFO, snapshot_id), # pyright: ignore
|
||||
returns=lambda snapshot_id, **kwargs: get_id(SNAPSHOTS_INFO, snapshot_id), # pyright: ignore
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.get_console_connection_info,
|
||||
return_value=CONSOLE_CONNECTION_INFO,
|
||||
returns=CONSOLE_CONNECTION_INFO,
|
||||
),
|
||||
AutoSpecMethodInfo(
|
||||
client.Client.create_snapshot,
|
||||
return_value=lambda *args, **kwargs: random.choice(SNAPSHOTS_INFO), # pyright: ignore
|
||||
returns=lambda *args, **kwargs: random.choice(SNAPSHOTS_INFO), # pyright: ignore
|
||||
),
|
||||
# connect returns None
|
||||
# Test method
|
||||
@ -218,7 +218,7 @@ CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
PROVIDER_VALUES_DICT: typing.Final[gui.ValuesDictType] = {
|
||||
'ovirt_version': '4',
|
||||
'host': 'host.example.com',
|
||||
'port': '443', # '443' is the default value
|
||||
'port': 443, # '443' is the default value
|
||||
'username': 'admin@ovirt@internalsso',
|
||||
'password': 'the_testing_pass',
|
||||
'concurrent_creation_limit': 33,
|
||||
@ -254,11 +254,10 @@ def patch_provider_api(
|
||||
**kwargs: typing.Any,
|
||||
) -> typing.Generator[mock.Mock, None, None]:
|
||||
client = create_client_mock()
|
||||
try:
|
||||
mock.patch('uds.services.OVirt.provider.OVirtProvider.api', returns=client).start()
|
||||
# api is a property, patch it correctly
|
||||
with mock.patch('uds.services.OVirt.provider.OVirtProvider.api', new_callable=mock.PropertyMock, **kwargs) as api:
|
||||
api.return_value = client
|
||||
yield client
|
||||
finally:
|
||||
mock.patch.stopall()
|
||||
|
||||
|
||||
def create_provider(**kwargs: typing.Any) -> 'provider.OVirtProvider':
|
||||
|
||||
@ -46,6 +46,7 @@ class TestOVirtProvider(UDSTransactionTestCase):
|
||||
"""
|
||||
provider = fixtures.create_provider() # Will not use client api, so no need to patch it
|
||||
|
||||
self.assertEqual(provider.ovirt_version.as_str(), fixtures.PROVIDER_VALUES_DICT['ovirt_version'])
|
||||
self.assertEqual(provider.host.as_str(), fixtures.PROVIDER_VALUES_DICT['host'])
|
||||
self.assertEqual(provider.port.as_int(), fixtures.PROVIDER_VALUES_DICT['port'])
|
||||
self.assertEqual(provider.username.as_str(), fixtures.PROVIDER_VALUES_DICT['username'])
|
||||
@ -106,7 +107,6 @@ class TestOVirtProvider(UDSTransactionTestCase):
|
||||
def test_provider_is_available(self) -> None:
|
||||
"""
|
||||
Test the provider is_available
|
||||
Thi is "specieal" because it uses cache
|
||||
"""
|
||||
with fixtures.patch_provider_api() as api:
|
||||
provider = fixtures.create_provider()
|
||||
|
||||
@ -271,88 +271,88 @@ CONSOLE_CONNECTION_INFO: typing.Final[types.services.ConsoleConnectionInfo] = (
|
||||
CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
# connect returns None
|
||||
# Test method
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.test, return_value=True),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.test, returns=True),
|
||||
# get_cluster_info
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_cluster_info, return_value=CLUSTER_INFO),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_cluster_info, returns=CLUSTER_INFO),
|
||||
# get_next_vmid
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_next_vmid, return_value=1),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_next_vmid, returns=1),
|
||||
# is_vmid_available
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.is_vmid_available, return_value=True),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.is_vmid_available, returns=True),
|
||||
# get_node_networks, not called never (ensure it's not called by mistake)
|
||||
# list_node_gpu_devices
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_node_gpu_devices, return_value=['gpu_dev_1', 'gpu_dev_2']),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_node_gpu_devices, returns=['gpu_dev_1', 'gpu_dev_2']),
|
||||
# list_node_vgpus
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_node_vgpus, return_value=VGPUS),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_node_vgpus, returns=VGPUS),
|
||||
# node_has_vgpus_available
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.node_has_vgpus_available, return_value=True),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.node_has_vgpus_available, returns=True),
|
||||
# get_best_node_for_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_best_node_for_machine, return_value=NODE_STATS[0]),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_best_node_for_machine, returns=NODE_STATS[0]),
|
||||
# clone_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.clone_machine, return_value=VM_CREATION_RESULT),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.clone_machine, returns=VM_CREATION_RESULT),
|
||||
# list_ha_groups
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_ha_groups, return_value=HA_GROUPS),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_ha_groups, returns=HA_GROUPS),
|
||||
# enable_machine_ha return None
|
||||
# disable_machine_ha return None
|
||||
# set_protection return None
|
||||
# get_guest_ip_address
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_guest_ip_address, return_value=GUEST_IP_ADDRESS),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_guest_ip_address, returns=GUEST_IP_ADDRESS),
|
||||
# remove_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.remove_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.remove_machine, returns=UPID),
|
||||
# list_snapshots
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_snapshots, return_value=SNAPSHOTS_INFO),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_snapshots, returns=SNAPSHOTS_INFO),
|
||||
# supports_snapshot
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.supports_snapshot, return_value=True),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.supports_snapshot, returns=True),
|
||||
# create_snapshot
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.create_snapshot, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.create_snapshot, returns=UPID),
|
||||
# remove_snapshot
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.remove_snapshot, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.remove_snapshot, returns=UPID),
|
||||
# restore_snapshot
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.restore_snapshot, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.restore_snapshot, returns=UPID),
|
||||
# get_task
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_task, return_value=TASK_STATUS),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_task, returns=TASK_STATUS),
|
||||
# list_machines
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_machines, return_value=VMS_INFO),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_machines, returns=VMS_INFO),
|
||||
# get_machine_pool_info
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_machine_pool_info,
|
||||
return_value=lambda vmid, poolid, **kwargs: VMS_INFO[vmid - 1], # pyright: ignore
|
||||
returns=lambda vmid, poolid, **kwargs: VMS_INFO[vmid - 1], # pyright: ignore
|
||||
),
|
||||
# get_machine_info
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_machine_info,
|
||||
return_value=lambda vmid, *args, **kwargs: VMS_INFO[vmid - 1], # pyright: ignore
|
||||
returns=lambda vmid, *args, **kwargs: VMS_INFO[vmid - 1], # pyright: ignore
|
||||
),
|
||||
# get_machine_configuration
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_machine_configuration,
|
||||
return_value=lambda vmid, **kwargs: VMS_CONFIGURATION[vmid - 1], # pyright: ignore
|
||||
returns=lambda vmid, **kwargs: VMS_CONFIGURATION[vmid - 1], # pyright: ignore
|
||||
),
|
||||
# enable_machine_ha return None
|
||||
# start_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.start_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.start_machine, returns=UPID),
|
||||
# stop_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.stop_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.stop_machine, returns=UPID),
|
||||
# reset_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.reset_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.reset_machine, returns=UPID),
|
||||
# suspend_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.suspend_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.suspend_machine, returns=UPID),
|
||||
# resume_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.resume_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.resume_machine, returns=UPID),
|
||||
# shutdown_machine
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.shutdown_machine, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.shutdown_machine, returns=UPID),
|
||||
# convert_to_template
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.convert_to_template, return_value=UPID),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.convert_to_template, returns=UPID),
|
||||
# get_storage
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_storage,
|
||||
return_value=lambda storage, node, **kwargs: next( # pyright: ignore
|
||||
returns=lambda storage, node, **kwargs: next( # pyright: ignore
|
||||
filter(lambda s: s.storage == storage, STORAGES) # pyright: ignore
|
||||
), # pyright: ignore
|
||||
),
|
||||
# list_storages
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.list_storages,
|
||||
return_value=lambda node, **kwargs: ( # pyright: ignore
|
||||
returns=lambda node, **kwargs: ( # pyright: ignore
|
||||
(list(filter(lambda s: s.node == node, STORAGES))) # pyright: ignore
|
||||
if node is not None
|
||||
else STORAGES # pyright: ignore
|
||||
@ -361,23 +361,23 @@ CLIENT_METHODS_INFO: typing.Final[list[AutoSpecMethodInfo]] = [
|
||||
# get_node_stats
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_node_stats,
|
||||
return_value=lambda node, **kwargs: next( # pyright: ignore
|
||||
returns=lambda node, **kwargs: next( # pyright: ignore
|
||||
filter(lambda n: n.name == node, NODE_STATS) # pyright: ignore
|
||||
),
|
||||
),
|
||||
# list_pools
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_pools, return_value=POOLS),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.list_pools, returns=POOLS),
|
||||
# get_pool_info
|
||||
AutoSpecMethodInfo(
|
||||
client.ProxmoxClient.get_pool_info,
|
||||
return_value=lambda poolid, **kwargs: next( # pyright: ignore
|
||||
returns=lambda poolid, **kwargs: next( # pyright: ignore
|
||||
filter(lambda p: p.poolid == poolid, POOLS) # pyright: ignore
|
||||
),
|
||||
),
|
||||
# get_console_connection
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_console_connection, return_value=CONSOLE_CONNECTION_INFO),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.get_console_connection, returns=CONSOLE_CONNECTION_INFO),
|
||||
# journal
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.journal, return_value=['journal line 1', 'journal line 2']),
|
||||
AutoSpecMethodInfo(client.ProxmoxClient.journal, returns=['journal line 1', 'journal line 2']),
|
||||
]
|
||||
|
||||
PROVIDER_VALUES_DICT: typing.Final[gui.ValuesDictType] = {
|
||||
|
||||
@ -36,7 +36,7 @@ from unittest import mock
|
||||
@dataclasses.dataclass
|
||||
class AutoSpecMethodInfo:
|
||||
name: str|typing.Callable[..., typing.Any]
|
||||
return_value: typing.Any = None # Can be a callable or a value
|
||||
returns: typing.Any = None # Can be a callable or a value
|
||||
|
||||
|
||||
def autospec(cls: type, metods_info: collections.abc.Iterable[AutoSpecMethodInfo], **kwargs: typing.Any) -> mock.Mock:
|
||||
@ -54,9 +54,9 @@ def autospec(cls: type, metods_info: collections.abc.Iterable[AutoSpecMethodInfo
|
||||
# Set the return value for the method or the side_effect
|
||||
name = method_info.name if isinstance(method_info.name, str) else method_info.name.__name__
|
||||
mck = getattr(obj, name)
|
||||
if callable(method_info.return_value):
|
||||
mck.side_effect = method_info.return_value
|
||||
if callable(method_info.returns):
|
||||
mck.side_effect = method_info.returns
|
||||
else:
|
||||
mck.return_value = method_info.return_value
|
||||
mck.return_value = method_info.returns
|
||||
|
||||
return obj
|
||||
Loading…
x
Reference in New Issue
Block a user