Openstack get mac external dhcp

src/uds/services/OpenStack/deployment.py

@@ -168,10 +168,11 @@ def op_create_checker(self) -> types.states.TaskState:
         """
         # Checks if machine has been created
         # Should have a mac address assigned and be running
-        if self.service().get_mac(self, self._vmid) and self.service().is_running(self, self._vmid):
-            server_info = self.service().api.get_server_info(self._vmid).validated()
-            self._mac = server_info.addresses[0].mac
-            self._ip = server_info.addresses[0].ip
+        mac = self.service().get_mac(self, self._vmid)
+        if mac and self.service().is_running(self, self._vmid):
+            self._mac = mac
+            # external DHCP leaves Nova 'addresses' empty; the guarded getters avoid indexing it
+            self._ip = self.service().get_ip(self, self._vmid)
             return types.states.TaskState.FINISHED
 
         return types.states.TaskState.RUNNING

src/uds/services/OpenStack/helpers.py

@@ -48,7 +48,7 @@ def get_api(parameters: dict[str, str]) -> tuple[client.OpenStackClient, bool]:
     from .provider import OpenStackProvider
 
     provider = typing.cast(
-        typing.Union[OpenStackProviderLegacy, OpenStackProvider],
+        OpenStackProviderLegacy | OpenStackProvider,
         models.Provider.objects.get(uuid=parameters['prov_uuid']).get_instance(),
     )
 

src/uds/services/OpenStack/openstack/client.py

@@ -563,6 +563,29 @@
             )
         ]
 
+    def list_ports(
+        self,
+        network_id: str | None = None,
+        device_id: str | None = None,
+    ) -> list[openstack_types.PortInfo]:
+        query = ''
+        if network_id is not None:
+            query += f'&network_id={network_id}'
+        if device_id is not None:
+            query += f'&device_id={device_id}'
+        if query:
+            query = '?' + query[1:]  # Replace leading '&' with '?'
+
+        return [
+            openstack_types.PortInfo.from_dict(p)
+            for p in self._get_recurring_from_endpoint(
+                endpoint_types=NETWORKS_ENDPOINT_TYPES,
+                path=f'/v2.0/ports{query}',
+                error_message='List ports',
+                key='ports',
+            )
+        ]
+
     @decorators.cached(prefix='sgps', timeout=consts.cache.EXTREME_CACHE_TIMEOUT, key_helper=cache_key_helper)
     def list_security_groups(self) -> list[openstack_types.SecurityGroupInfo]:
         return [
@@ -586,23 +609,29 @@
             path=f'/servers/{server_id}',
             error_message='Get Server information',
         )
-        server_info = openstack_types.ServerInfo.from_dict(r.json()['server'])
-        # # If no addresses got, try the servers get IPS
-        # if len(server_info.addresses) == 0 or server_info.addresses[0].mac == '':
-        #     try:
-        #         r = self._request_from_endpoint(
-        #             'get',
-        #             endpoints_types=COMPUTE_ENDPOINT_TYPES,
-        #             path=f'/servers/{server_id}/ips',
-        #             error_message='Get Server IPs information',
-        #         )
-        #         # First element, the network name, don't care about it
-        #         data: dict[str, typing.Any] = r.json()
-        #         addrs = openstack_types.ServerInfo.AddresInfo.from_addresses(data['addresses'])
-        #         server_info.addresses = addrs if len(addrs) and addrs[0].ip and addrs[0].mac else server_info.addresses
-        #     except Exception:
-        #         pass  # Not found, all is fine
-        return server_info
+        return openstack_types.ServerInfo.from_dict(r.json()['server'])
+
+    @decorators.cached(prefix='svr_mac', timeout=consts.cache.SHORT_CACHE_TIMEOUT, key_helper=cache_key_helper)
+    def get_server_mac(self, server_id: str, **kwargs: typing.Any) -> str:
+        # Returning '' makes the state checker retry instead of forcing the userservice
+        # to ERROR, so degrade to it whenever the mac cannot be resolved yet.
+        net_info = self.get_server_info(server_id).validated().addresses
+
+        if net_info and net_info[0].mac:
+            return net_info[0].mac
+
+        # 'addresses' is empty under external DHCP, but the Neutron port still carries the mac.
+        try:
+            ports = self.list_ports(device_id=server_id)
+        except exceptions.services.generics.Error as e:
+            # never let a transient/403 error reach the state checker: '' just retries
+            logger.debug('Listing Neutron ports for %s failed, mac unresolved: %s', server_id, e)
+            return ''
+        for port in ports:
+            if port.mac_address:
+                return port.mac_address
+        return ''
+
 
     @decorators.cached(prefix='vol', timeout=consts.cache.SHORTEST_CACHE_TIMEOUT, key_helper=cache_key_helper)
     def get_volume_info(self, volume_id: str, **kwargs: typing.Any) -> openstack_types.VolumeInfo:
@@ -802,7 +831,7 @@
             r = self._session.get(self._identity_endpoint, headers=self._get_request_headers())
         except Exception:
             logger.exception('Testing')
             raise Exception('Connection error')
 
         try:
             values = r.json()['versions']['values']

src/uds/services/OpenStack/openstack/types.py

@@ -245,6 +245,30 @@ def from_str(s: str) -> 'PortStatus':
             return PortStatus.ERROR
 
 
+@dataclasses.dataclass
+class PortInfo:
+    id: str
+    name: str
+    status: PortStatus
+    device_id: str  # Owner server id (the VM this port is attached to)
+    network_id: str
+    mac_address: str
+    fixed_ips: list[str]  # List of ip addresses assigned to the port (may be empty)
+
+    @staticmethod
+    def from_dict(d: dict[str, typing.Any]) -> 'PortInfo':
+        return PortInfo(
+            id=d['id'],
+            name=d.get('name') or d['id'],
+            status=PortStatus.from_str(d.get('status', PortStatus.ERROR.value)),
+            device_id=d.get('device_id') or '',
+            network_id=d.get('network_id') or '',
+            # Neutron always assigns a mac to the port, regardless of DHCP/IP allocation
+            mac_address=(d.get('mac_address') or '').upper(),
+            fixed_ips=[fi['ip_address'] for fi in d.get('fixed_ips', []) if fi.get('ip_address')],
+        )
+
+
 @dataclasses.dataclass
 class ServerInfo:
 

src/uds/services/OpenStack/provider_legacy.py

@@ -175,7 +175,7 @@ class OpenStackProviderLegacy(ServiceProvider):
     legacy = True
 
     # Own variables
-    _api: typing.Optional[client.OpenStackClient] = None
+    _api: client.OpenStackClient | None = None
 
     def initialize(self, values: 'types.core.ValuesType') -> None:
         """
@@ -187,9 +187,9 @@ def initialize(self, values: 'types.core.ValuesType') -> None:
             self.timeout.value = validators.validate_timeout(self.timeout.value)
 
     def api(
-        self, projectid: typing.Optional[str] = None, region: typing.Optional[str] = None
+        self, projectid: str | None = None, region: str | None = None
     ) -> client.OpenStackClient:
-        proxies: typing.Optional[dict[str, str]] = None
+        proxies: dict[str, str] | None = None
         if self.https_proxy.value.strip():
             proxies = {'https': self.https_proxy.value}
         return client.OpenStackClient(

src/uds/services/OpenStack/service.py

@@ -263,8 +263,7 @@ def get_mac(
         *,
         for_unique_id: bool = False,
     ) -> str:
-        net_info = self.api.get_server_info(vmid).validated().addresses
-        return '' if not net_info else net_info[0].mac
+        return self.api.get_server_mac(vmid)
 
     def is_running(
         self, caller_instance: 'DynamicUserService | DynamicPublication | None', vmid: str

src/uds/services/OpenStack/service_fixed.py

@@ -48,7 +48,7 @@
     from .provider import OpenStackProvider
     from .provider_legacy import OpenStackProviderLegacy
 
-    AnyOpenStackProvider: typing.TypeAlias = typing.Union[OpenStackProvider, OpenStackProviderLegacy]
+    AnyOpenStackProvider: typing.TypeAlias = OpenStackProvider | OpenStackProviderLegacy
 
 logger = logging.getLogger(__name__)
 
@@ -105,7 +105,7 @@ class OpenStackServiceFixed(FixedService):  # pylint: disable=too-many-public-me
 
     prov_uuid = gui.HiddenField()
 
-    _api: typing.Optional['client.OpenStackClient'] = None
+    _api: 'client.OpenStackClient | None' = None
 
     @property
     def api(self) -> 'client.OpenStackClient':
@@ -164,7 +164,7 @@ def enumerate_assignables(self) -> collections.abc.Iterable[types.ui.ChoiceItem]
             ]
 
     def get_and_assign(self) -> str:
-        found_vmid: typing.Optional[str] = None
+        found_vmid: str | None = None
         try:
             with self._assigned_access() as assigned:
                 for checking_vmid in self.sorted_assignables_list():
@@ -196,7 +196,9 @@ def get_and_assign(self) -> str:
         return found_vmid
 
     def get_mac(self, vmid: str) -> str:
-        return self.api.get_server_info(vmid).addresses[0].mac
+        # Returning '' lets the caller retry instead of crashing when the mac cannot be
+        # resolved yet (machine missing, or external DHCP leaving 'addresses' empty).
+        return self.api.get_server_mac(vmid) or ''
 
     def get_ip(self, vmid: str) -> str:
         return self.api.get_server_info(vmid).addresses[0].ip

tests/services/openstack/test_service_fixed_get_mac.py

@@ -0,0 +1,129 @@
+# -*- coding: utf-8 -*-
+
+#
+# Copyright (c) 2024 Virtual Cable S.L.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+#
+#    * Redistributions of source code must retain the above copyright notice,
+#      this list of conditions and the following disclaimer.
+#    * Redistributions in binary form must reproduce the above copyright notice,
+#      this list of conditions and the following disclaimer in the documentation
+#      and/or other materials provided with the distribution.
+#    * Neither the name of Virtual Cable S.L. nor the names of its contributors
+#      may be used to endorse or promote products derived from this software
+#      without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+"""
+Tests for OpenStackServiceFixed.get_mac unique_id resolution.
+
+Mirrors the live-service coverage: server 'addresses' (internal DHCP), Neutron port
+fallback (external DHCP), and resilience to not-found / permission errors -> ''.
+"""
+import copy
+import typing
+from unittest import mock
+
+from uds.core import exceptions
+
+from . import fixtures
+
+from ...utils.test import UDSTransactionTestCase
+
+from uds.services.OpenStack.openstack import types as openstack_types
+
+
+def _server_with_addresses(
+    addresses: list[openstack_types.ServerInfo.AddresInfo],
+) -> openstack_types.ServerInfo:
+    server = copy.deepcopy(fixtures.SERVERS_LIST[0])
+    server.status = openstack_types.ServerStatus.ACTIVE
+    server.addresses = addresses
+    return server
+
+
+def _port(mac: str, *, device_id: str = 'sid1', ip: str = '') -> openstack_types.PortInfo:
+    return openstack_types.PortInfo(
+        id='pid1',
+        name='port1',
+        status=openstack_types.PortStatus.ACTIVE,
+        device_id=device_id,
+        network_id='net1',
+        mac_address=mac,
+        fixed_ips=[ip] if ip else [],
+    )
+
+
+class TestOpenStackFixedGetMac(UDSTransactionTestCase):
+    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()
+        service._api = api  # the api property returns the cached client without hitting the provider
+        return service, api
+
+    def test_get_mac_from_server_addresses(self) -> None:
+        # Internal DHCP: server addresses carry the mac
+        service, api = self._service_with_api()
+        api.get_server_info.return_value = _server_with_addresses(
+            [openstack_types.ServerInfo.AddresInfo(version=4, ip='10.0.0.5', mac='FA:16:3E:00:00:01', type='fixed')]
+        )
+
+        mac = service.get_mac('sid1')
+
+        self.assertEqual(mac, 'FA:16:3E:00:00:01')
+        api.list_ports.assert_not_called()
+
+    def test_get_mac_falls_back_to_neutron_port(self) -> None:
+        # External DHCP: server addresses empty -> use the Neutron port mac
+        service, api = self._service_with_api()
+        api.get_server_info.return_value = _server_with_addresses([])
+        api.list_ports.return_value = [_port('FA:16:3E:AB:CD:EF', device_id='sid1')]
+
+        mac = service.get_mac('sid1')
+
+        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
+        service, api = self._service_with_api()
+        api.get_server_info.side_effect = exceptions.services.generics.NotFoundError('Not found')
+
+        mac = 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:
+        # External DHCP but the port is not there yet -> '' instead of IndexError
+        service, api = self._service_with_api()
+        api.get_server_info.return_value = _server_with_addresses([])
+        api.list_ports.return_value = []
+
+        mac = service.get_mac('sid1')
+
+        self.assertEqual(mac, '')
+
+    def test_get_mac_empty_when_port_lookup_permission_denied(self) -> None:
+        # Neutron policy forbids listing ports (403) -> '' instead of crashing
+        service, api = self._service_with_api()
+        api.get_server_info.return_value = _server_with_addresses([])
+        api.list_ports.side_effect = exceptions.services.generics.Error('Forbidden')
+
+        mac = service.get_mac('sid1')
+
+        self.assertEqual(mac, '')