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

Improved openstack tests to avoid problems

Browse Source
This commit is contained in:
Adolfo Gómez García 2024-03-26 03:36:48 +01:00
parent 93ebf82c79
commit 83fc9f2e7f
No known key found for this signature in database
GPG Key ID: DD1ABF20724CDA23
6 changed files with 96 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
server/tests/services/openstack/fixtures.py
View File

@ -372,22 +372,24 @@ def create_client_mock() -> mock.Mock:
@contextlib.contextmanager
def patch_provider_api(
legacy: bool = False,
def patched_provider(
**kwargs: typing.Any,
) -> typing.Generator[mock.Mock, None, None]:
) -> typing.Generator[provider.OpenStackProvider, None, None]:
client = create_client_mock()
path = (
'uds.services.OpenStack.provider_legacy.OpenStackProviderLegacy'
if legacy
else 'uds.services.OpenStack.provider.OpenStackProvider'
)
try:
mock.patch(path + '.api', return_value=client).start()
yield client
finally:
mock.patch.stopall()
provider = create_provider(**kwargs)
with mock.patch.object(provider, 'api') as api:
api.return_value = client
yield provider
@contextlib.contextmanager
def patched_provider_legacy(
**kwargs: typing.Any,
) -> typing.Generator[provider_legacy.OpenStackProviderLegacy, None, None]:
client = create_client_mock()
provider = create_provider_legacy(**kwargs)
with mock.patch.object(provider, 'api') as api:
api.return_value = client
yield provider
def create_provider(**kwargs: typing.Any) -> provider.OpenStackProvider:
"""

74
server/tests/services/openstack/test_provider.py
View File

@ -30,6 +30,7 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
import random
from unittest import mock
@ -49,18 +50,18 @@ class TestOpenstackProvider(UDSTransactionTestCase):
"""
Test the provider
"""
with fixtures.patch_provider_api() as client:
provider = fixtures.create_provider() # Will not use client api, so no need to patch it
with fixtures.patched_provider() as provider:
api = typing.cast(mock.MagicMock, provider.api())
self.assertEqual(provider.test_connection(), types.core.TestResult(True, mock.ANY))
# Ensure test_connection is called
client.test_connection.assert_called_once()
api.test_connection.assert_called_once()
self.assertEqual(provider.is_available(), True)
client.is_available.assert_called_once()
api.is_available.assert_called_once()
# Clear mock calls
client.reset_mock()
api.reset_mock()
OpenStackProvider.test(
env=environment.Environment.testing_environment(), data=fixtures.PROVIDER_VALUES_DICT
)
@ -69,18 +70,18 @@ class TestOpenstackProvider(UDSTransactionTestCase):
"""
Test the provider
"""
with fixtures.patch_provider_api(legacy=True) as client:
provider = fixtures.create_provider_legacy() # Will not use client api, so no need to patch it
with fixtures.patched_provider_legacy() as provider:
api = typing.cast(mock.MagicMock, provider.api())
self.assertEqual(provider.test_connection(), types.core.TestResult(True, mock.ANY))
# Ensure test_connection is called
client.test_connection.assert_called_once()
api.test_connection.assert_called_once()
self.assertEqual(provider.is_available(), True)
client.is_available.assert_called_once()
api.is_available.assert_called_once()
# Clear mock calls
client.reset_mock()
api.reset_mock()
OpenStackProviderLegacy.test(
env=environment.Environment.testing_environment(), data=fixtures.PROVIDER_VALUES_DICT
)
@ -91,8 +92,8 @@ class TestOpenstackProvider(UDSTransactionTestCase):
"""
from uds.services.OpenStack.helpers import get_machines, get_resources, get_volumes
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as _api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
# Ensure exists on db
db_provider = models.Provider.objects.create(
name='test proxmox provider',
@ -107,25 +108,30 @@ class TestOpenstackProvider(UDSTransactionTestCase):
'region': random.choice(fixtures.REGIONS_LIST).id,
}
# Test get_storage
h_machines = get_machines(parameters)
self.assertEqual(len(h_machines), 1)
self.assertEqual(h_machines[0]['name'], 'machines')
self.assertEqual(sorted(i['id'] for i in h_machines[0]['choices']), sorted(i.id for i in fixtures.SERVERS_LIST))
h_resources = get_resources(parameters)
# [{'name': 'availability_zone', 'choices': [...]}, {'name': 'network', 'choices': [...]}, {'name': 'flavor', 'choices': [...]}, {'name': 'security_groups', 'choices': [...]}]
self.assertEqual(len(h_resources), 4)
self.assertEqual(sorted(i['name'] for i in h_resources), ['availability_zone', 'flavor', 'network', 'security_groups'])
def _get_choices_for(name: str) -> list[str]:
return [i['id'] for i in next(i for i in h_resources if i['name'] == name)['choices']]
self.assertEqual(sorted(_get_choices_for('availability_zone')), sorted(i.id for i in fixtures.AVAILABILITY_ZONES_LIST))
self.assertEqual(sorted(_get_choices_for('network')), sorted(i.id for i in fixtures.NETWORKS_LIST))
self.assertEqual(sorted(_get_choices_for('flavor')), sorted(i.id for i in fixtures.FLAVORS_LIST if not i.disabled))
self.assertEqual(sorted(_get_choices_for('security_groups')), sorted(i.id for i in fixtures.SECURITY_GROUPS_LIST))
# [{'name': 'volume', 'choices': [...]}]
h_volumes = get_volumes(parameters)
self.assertEqual(len(h_volumes), 1)
self.assertEqual(h_volumes[0]['name'], 'volume')
self.assertEqual(sorted(i['id'] for i in h_volumes[0]['choices']), sorted(i.id for i in fixtures.VOLUMES_LIST))
# Helpers need a bit more patching to work
# We must patch get_api(parameters: dict[str, str]) -> tuple[openstack_client.OpenstackClient, bool]:
with mock.patch('uds.services.OpenStack.helpers.get_api') as get_api:
get_api.return_value = (prov.api(), False)
h_machines = get_machines(parameters)
self.assertEqual(len(h_machines), 1)
self.assertEqual(h_machines[0]['name'], 'machines')
self.assertEqual(sorted(i['id'] for i in h_machines[0]['choices']), sorted(i.id for i in fixtures.SERVERS_LIST))
h_resources = get_resources(parameters)
# [{'name': 'availability_zone', 'choices': [...]}, {'name': 'network', 'choices': [...]}, {'name': 'flavor', 'choices': [...]}, {'name': 'security_groups', 'choices': [...]}]
self.assertEqual(len(h_resources), 4)
self.assertEqual(sorted(i['name'] for i in h_resources), ['availability_zone', 'flavor', 'network', 'security_groups'])
def _get_choices_for(name: str) -> list[str]:
return [i['id'] for i in next(i for i in h_resources if i['name'] == name)['choices']]
self.assertEqual(sorted(_get_choices_for('availability_zone')), sorted(i.id for i in fixtures.AVAILABILITY_ZONES_LIST))
self.assertEqual(sorted(_get_choices_for('network')), sorted(i.id for i in fixtures.NETWORKS_LIST))
self.assertEqual(sorted(_get_choices_for('flavor')), sorted(i.id for i in fixtures.FLAVORS_LIST if not i.disabled))
self.assertEqual(sorted(_get_choices_for('security_groups')), sorted(i.id for i in fixtures.SECURITY_GROUPS_LIST))
# [{'name': 'volume', 'choices': [...]}]
h_volumes = get_volumes(parameters)
self.assertEqual(len(h_volumes), 1)
self.assertEqual(h_volumes[0]['name'], 'volume')
self.assertEqual(sorted(i['id'] for i in h_volumes[0]['choices']), sorted(i.id for i in fixtures.VOLUMES_LIST))

35
server/tests/services/openstack/test_service.py
View File

@ -31,8 +31,8 @@
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
# from uds.core import types, environment
import typing
from unittest import mock
from . import fixtures
@ -49,23 +49,24 @@ class TestOpenstackService(UDSTransactionTestCase):
"""
Test the service for all kind of providers
"""
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as client:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
api = typing.cast(mock.MagicMock, prov.api())
service = fixtures.create_live_service(prov) # Will use provider patched api
self.assertEqual(service.api, client)
self.assertEqual(service.api, api)
self.assertEqual(service.sanitized_name('a b c'), 'a_b_c')
template = service.make_template('template', 'desc')
self.assertIsInstance(template, openstack_types.VolumeSnapshotInfo)
client.create_volume_snapshot.assert_called_once_with(service.volume.value, 'template', 'desc')
api.create_volume_snapshot.assert_called_once_with(service.volume.value, 'template', 'desc')
template = service.get_template(fixtures.VOLUME_SNAPSHOTS_LIST[0].id)
self.assertIsInstance(template, openstack_types.VolumeSnapshotInfo)
client.get_volume_snapshot.assert_called_once_with(fixtures.VOLUME_SNAPSHOTS_LIST[0].id)
api.get_volume_snapshot.assert_called_once_with(fixtures.VOLUME_SNAPSHOTS_LIST[0].id)
data: typing.Any = service.deploy_from_template('name', fixtures.VOLUME_SNAPSHOTS_LIST[0].id)
self.assertIsInstance(data, openstack_types.ServerInfo)
client.create_server_from_snapshot.assert_called_once_with(
api.create_server_from_snapshot.assert_called_once_with(
snapshot_id=fixtures.VOLUME_SNAPSHOTS_LIST[0].id,
name='name',
availability_zone=service.availability_zone.value,
@ -75,40 +76,40 @@ class TestOpenstackService(UDSTransactionTestCase):
)
data = service.get_machine_status(fixtures.SERVERS_LIST[0].id)
self.assertIsInstance(data, openstack_types.ServerStatus)
client.get_server.assert_called_once_with(fixtures.SERVERS_LIST[0].id)
api.get_server.assert_called_once_with(fixtures.SERVERS_LIST[0].id)
# Reset mocks, get server should be called again
client.reset_mock()
api.reset_mock()
data = service.get_machine_power_state(fixtures.SERVERS_LIST[0].id)
self.assertIsInstance(data, openstack_types.PowerState)
client.get_server.assert_called_once_with(fixtures.SERVERS_LIST[0].id)
api.get_server.assert_called_once_with(fixtures.SERVERS_LIST[0].id)
server = fixtures.SERVERS_LIST[0]
service.start_machine(server.id)
server.power_state = openstack_types.PowerState.SHUTDOWN
client.start_server.assert_called_once_with(server.id)
api.start_server.assert_called_once_with(server.id)
server.power_state = openstack_types.PowerState.RUNNING
service.stop_machine(server.id)
client.stop_server.assert_called_once_with(server.id)
api.stop_server.assert_called_once_with(server.id)
server.power_state = openstack_types.PowerState.RUNNING
service.suspend_machine(server.id)
client.suspend_server.assert_called_once_with(server.id)
api.suspend_server.assert_called_once_with(server.id)
server.power_state = openstack_types.PowerState.SUSPENDED
service.resume_machine(server.id)
client.resume_server.assert_called_once_with(server.id)
api.resume_server.assert_called_once_with(server.id)
service.reset_machine(server.id)
client.reset_server.assert_called_once_with(server.id)
api.reset_server.assert_called_once_with(server.id)
service.delete_machine(server.id)
client.delete_server.assert_called_once_with(server.id)
api.delete_server.assert_called_once_with(server.id)
self.assertTrue(service.is_avaliable())
client.is_available.assert_called_once_with()
api.is_available.assert_called_once_with()
self.assertEqual(service.get_basename(), service.basename.value)
self.assertEqual(service.get_lenname(), service.lenname.value)

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

@ -30,10 +30,13 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
from uds import models
from uds.core import types
from unittest import mock
from . import fixtures
from ...utils.test import UDSTransactionTestCase
@ -43,8 +46,9 @@ class TestProxmovFixedService(UDSTransactionTestCase):
"""
Test the service
"""
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
api = typing.cast(mock.MagicMock, prov.api())
service = fixtures.create_fixed_service(prov) # Will use provider patched api
userservice = fixtures.create_fixed_userservice(service)

20
server/tests/services/openstack/test_userservice.py
View File

@ -30,8 +30,10 @@
"""
Author: Adolfo Gómez, dkmaster at dkmon dot com
"""
import typing
from uds import models
from uds.core import types
from unittest import mock
from uds.services.OpenStack.openstack import types as openstack_types
from uds.services.OpenStack.deployment import Operation
@ -54,8 +56,9 @@ class TestOpenstackLiveDeployment(UDSTransactionTestCase):
"""
# Deploy for cache and deploy for user are the same, so we will test both at the same time
for to_test in ['cache', 'user']:
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
api = typing.cast(mock.MagicMock, prov.api())
service = fixtures.create_live_service(prov)
userservice = fixtures.create_live_userservice(service=service)
publication = userservice.publication()
@ -117,8 +120,8 @@ class TestOpenstackLiveDeployment(UDSTransactionTestCase):
"""
Test the user service
"""
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as _api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
service = fixtures.create_live_service(prov)
userservice = fixtures.create_live_userservice(service=service)
publication = userservice.publication()
@ -161,8 +164,8 @@ class TestOpenstackLiveDeployment(UDSTransactionTestCase):
Test the user service
"""
for keep_on_error in (True, False):
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as _api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
service = fixtures.create_live_service(prov, maintain_on_error=keep_on_error)
userservice = fixtures.create_live_userservice(service=service)
publication = userservice.publication()
@ -195,8 +198,9 @@ class TestOpenstackLiveDeployment(UDSTransactionTestCase):
This test will have keep on error active, and will create incorrectly
so vm will be deleted and put on error state
"""
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
api = typing.cast(mock.MagicMock, prov.api())
service = fixtures.create_live_service(prov, maintain_on_eror=True)
userservice = fixtures.create_live_userservice(service=service)
publication = userservice.publication()

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

@ -50,8 +50,8 @@ class TestOpenstackFixedService(UDSTransactionTestCase):
"""
Test the user service
"""
for prov in (fixtures.create_provider_legacy(), fixtures.create_provider()):
with fixtures.patch_provider_api(legacy=prov.legacy) as _api:
for patcher in (fixtures.patched_provider, fixtures.patched_provider_legacy):
with patcher() as prov:
service = fixtures.create_fixed_service(prov) # Will use provider patched api
userservice = fixtures.create_fixed_userservice(service)
@ -97,7 +97,9 @@ class TestOpenstackFixedService(UDSTransactionTestCase):
# ensure cache is empty, may affect from other tests
userservice.cache.clear()
# Also that machine is stopped
fixtures.get_id(fixtures.SERVERS_LIST, userservice._vmid).power_state = openstack_types.PowerState.SHUTDOWN
fixtures.get_id(fixtures.SERVERS_LIST, userservice._vmid).power_state = (
openstack_types.PowerState.SHUTDOWN
)
state = userservice.set_ready()
self.assertEqual(state, types.states.TaskState.RUNNING)