Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 15 additions & 4 deletions tests/services/openstack/test_service_fixed_get_mac.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,18 @@

from ...utils.test import UDSTransactionTestCase

from uds.services.OpenStack.openstack import client
from uds.services.OpenStack.openstack import types as openstack_types

# Undecorated mac-resolution logic (the @cached wrapper would need a real cache backend).
# Running it against the mocked api keeps these tests exercising the real behaviour while
# the service simply delegates to api.get_server_mac().
_real_get_server_mac = getattr(client.OpenStackClient.get_server_mac, '__wrapped__')


def _bind_real_get_server_mac(api: mock.MagicMock) -> None:
api.get_server_mac.side_effect = lambda vmid, **kw: _real_get_server_mac(api, vmid, **kw)


def _server_with_addresses(
addresses: list[openstack_types.ServerInfo.AddresInfo],
Expand Down Expand Up @@ -72,6 +82,7 @@ def _service_with_api(self) -> tuple[typing.Any, mock.MagicMock]:
with fixtures.patched_provider() as prov:
service = fixtures.create_fixed_service(prov)
api = mock.MagicMock()
_bind_real_get_server_mac(api)
service._api = api # the api property returns the cached client without hitting the provider
return service, api

Expand All @@ -98,14 +109,14 @@ def test_get_mac_falls_back_to_neutron_port(self) -> None:
self.assertEqual(mac, 'FA:16:3E:AB:CD:EF')
api.list_ports.assert_called_once_with(device_id='sid1')

def test_get_mac_empty_when_server_not_found(self) -> None:
# Machine missing: no IndexError, no leak -> '' so the caller can retry
def test_get_mac_raises_when_server_not_found(self) -> None:
# Server missing: get_server_mac does not swallow it; the caller handles NotFoundError.
service, api = self._service_with_api()
api.get_server_info.side_effect = exceptions.services.generics.NotFoundError('Not found')

mac = service.get_mac('sid1')
with self.assertRaises(exceptions.services.generics.NotFoundError):
service.get_mac('sid1')

self.assertEqual(mac, '')
api.list_ports.assert_not_called()

def test_get_mac_empty_when_no_addresses_and_no_ports(self) -> None:
Expand Down
20 changes: 16 additions & 4 deletions tests/services/openstack/test_service_get_mac.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,18 @@

from ...utils.test import UDSTransactionTestCase

from uds.services.OpenStack.openstack import client
from uds.services.OpenStack.openstack import types as openstack_types

# Undecorated mac-resolution logic (the @cached wrapper would need a real cache backend).
# Running it against the mocked api keeps these tests exercising the real behaviour while
# the service simply delegates to api.get_server_mac().
_real_get_server_mac = getattr(client.OpenStackClient.get_server_mac, '__wrapped__')


def _bind_real_get_server_mac(api: mock.MagicMock) -> None:
api.get_server_mac.side_effect = lambda vmid, **kw: _real_get_server_mac(api, vmid, **kw)


def _server_with_addresses(
addresses: list[openstack_types.ServerInfo.AddresInfo],
Expand Down Expand Up @@ -75,6 +85,7 @@ def _service_with_api(self) -> tuple[typing.Any, mock.MagicMock]:
with fixtures.patched_provider() as prov:
service = fixtures.create_live_service(prov)
api = mock.MagicMock()
_bind_real_get_server_mac(api)
service.cached_api = api # property returns this without hitting the provider
return service, api

Expand Down Expand Up @@ -105,14 +116,15 @@ def test_get_mac_falls_back_to_neutron_port(self) -> None:
self.assertEqual(mac, 'FA:16:3E:AB:CD:EF')
api.list_ports.assert_called_once_with(device_id='sid1')

def test_get_mac_empty_when_server_not_found(self) -> None:
# Machine still being prepared: server not queryable yet -> '' (retry, no ERROR)
def test_get_mac_raises_when_server_not_found(self) -> None:
# Server not queryable yet: get_server_mac does not swallow it; the caller/state
# checker handles the NotFoundError.
service, api = self._service_with_api()
api.get_server_info.side_effect = exceptions.services.generics.NotFoundError('Not found')

mac = service.get_mac(None, 'sid1', for_unique_id=True)
with self.assertRaises(exceptions.services.generics.NotFoundError):
service.get_mac(None, 'sid1', for_unique_id=True)

self.assertEqual(mac, '')
api.list_ports.assert_not_called()

def test_get_mac_empty_when_no_addresses_and_no_ports(self) -> None:
Expand Down
Loading