Relax unique_id logic (#493)

Author: Snuffy2

custom_components/opnsense/__init__.py

@@ -211,7 +211,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     )
 
     # Trigger repair task and shutdown if device id has changed
-    router_device_id: str = await client.get_device_unique_id()
+    router_device_id: str | None = await client.get_device_unique_id(expected_id=config_device_id)
     _LOGGER.debug(
         "[init async_setup_entry]: config device id: %s, router device id: %s",
         config_device_id,

custom_components/opnsense/config_flow.py

@@ -88,13 +88,16 @@ async def validate_input(
     user_input: MutableMapping[str, Any],
     config_step: str,
     errors: dict[str, Any],
+    expected_id: str | None = None,
 ) -> dict[str, Any]:
     """Check user input for errors."""
     # filtered_user_input: MutableMapping[str, Any] = {key: value for key, value in user_input.items() if key != CONF_PASSWORD}
     # _LOGGER.debug("[validate_input] user_input: %s", filtered_user_input)
 
     try:
-        await _handle_user_input(hass=hass, user_input=user_input, config_step=config_step)
+        await _handle_user_input(
+            hass=hass, user_input=user_input, config_step=config_step, expected_id=expected_id
+        )
     except BelowMinFirmware:
         _log_and_set_error(
             errors=errors,
@@ -258,7 +261,10 @@ def _validate_firmware_version(firmware_version: str) -> None:
 
 
 async def _handle_user_input(
-    hass: HomeAssistant, user_input: MutableMapping[str, Any], config_step: str
+    hass: HomeAssistant,
+    user_input: MutableMapping[str, Any],
+    config_step: str,
+    expected_id: str | None = None,
 ) -> None:
     """Handle and validate the user input."""
     await _clean_and_parse_url(user_input)
@@ -305,7 +311,7 @@ async def _handle_user_input(
     if not user_input.get(CONF_NAME):
         user_input[CONF_NAME] = system_info.get("name") or "OPNsense"
 
-    user_input[CONF_DEVICE_UNIQUE_ID] = await client.get_device_unique_id()
+    user_input[CONF_DEVICE_UNIQUE_ID] = await client.get_device_unique_id(expected_id=expected_id)
     _LOGGER.debug("[handle_user_input] Device Unique ID: %s", user_input[CONF_DEVICE_UNIQUE_ID])
 
     if not user_input.get(CONF_DEVICE_UNIQUE_ID):
@@ -562,7 +568,11 @@ async def async_step_granular_sync(
             self._config.update(user_input)
             # _LOGGER.debug("[config_flow granular_sync] merged config: %s", self._config)
             errors = await validate_input(
-                hass=self.hass, user_input=self._config, config_step="granular_sync", errors=errors
+                hass=self.hass,
+                user_input=self._config,
+                config_step="granular_sync",
+                errors=errors,
+                expected_id=self._config.get(CONF_DEVICE_UNIQUE_ID),
             )
             if not errors:
                 return self.async_create_entry(
@@ -592,7 +602,11 @@ async def async_step_reconfigure(
             self._config.update(user_input)
             # _LOGGER.debug("[config_flow reconfigure] merged config: %s", self._config)
             errors = await validate_input(
-                hass=self.hass, user_input=self._config, config_step="reconfigure", errors=errors
+                hass=self.hass,
+                user_input=self._config,
+                config_step="reconfigure",
+                errors=errors,
+                expected_id=self._config.get(CONF_DEVICE_UNIQUE_ID),
             )
 
             if not errors:
@@ -701,7 +715,11 @@ async def async_step_granular_sync(
             self._config.update(user_input)
             # _LOGGER.debug("[options_flow granular_sync] merged user_input. config: %s. options: %s", self._config, self._options)
             errors = await validate_input(
-                hass=self.hass, user_input=self._config, config_step="granular_sync", errors=errors
+                hass=self.hass,
+                user_input=self._config,
+                config_step="granular_sync",
+                errors=errors,
+                expected_id=self._config.get(CONF_DEVICE_UNIQUE_ID),
             )
             if not errors:
                 if self._options.get(CONF_DEVICE_TRACKER_ENABLED):

custom_components/opnsense/coordinator.py

@@ -1,6 +1,6 @@
 """OPNsense Coordinator."""
 
-from collections.abc import Mapping, MutableMapping
+from collections.abc import Callable, Mapping, MutableMapping
 import copy
 from datetime import timedelta
 import logging
@@ -83,10 +83,14 @@ async def _get_states(self, categories: list) -> MutableMapping[str, Any]:
         state: MutableMapping[str, Any] = {}
         total_time: float = 0
         for cat in categories:
-            method = getattr(self._client, cat.get("function", ""), None)
-            if method:
+            method_name: str = cat.get("function", "")
+            method: Callable | None = getattr(self._client, method_name, None)
+            if method is not None:
                 start_time: float = time.perf_counter()
-                state[cat.get("state_key")] = await method()
+                if method_name == "get_device_unique_id":
+                    state[cat.get("state_key")] = await method(expected_id=self._device_unique_id)
+                else:
+                    state[cat.get("state_key")] = await method()
                 end_time: float = time.perf_counter()
                 elapsed_time: float = end_time - start_time
                 total_time += elapsed_time

custom_components/opnsense/pyopnsense/__init__.py

@@ -684,21 +684,28 @@ async def _filter_configure(self) -> None:
         await self._exec_php(script)
 
     @_log_errors
-    async def get_device_unique_id(self) -> str | None:
+    async def get_device_unique_id(self, expected_id: str | None = None) -> str | None:
         """Get the OPNsense Unique ID."""
         instances = await self._safe_list_get("/api/interfaces/overview/export")
-        mac_addresses = [
-            d.get("macaddr_hw") for d in instances if d.get("is_physical") and "macaddr_hw" in d
-        ]
+        mac_addresses = {
+            d.get("macaddr_hw").replace(":", "_").strip()
+            for d in instances
+            if d.get("is_physical") and d.get("macaddr_hw")
+        }
 
-        unique_mac_addresses: list = sorted(set(mac_addresses))
-        device_unique_id: str | None = unique_mac_addresses[0] if unique_mac_addresses else None
-        if device_unique_id:
-            device_unique_id_fmt = device_unique_id.replace(":", "_").strip()
-            _LOGGER.debug("[get_device_unique_id] device_unique_id: %s", device_unique_id_fmt)
-            return device_unique_id_fmt
-        _LOGGER.debug("[get_device_unique_id] device_unique_id: None")
-        return None
+        if not mac_addresses:
+            _LOGGER.debug("[get_device_unique_id] device_unique_id: None")
+            return None
+
+        if expected_id and expected_id in mac_addresses:
+            _LOGGER.debug(
+                "[get_device_unique_id] device_unique_id (matched expected): %s", expected_id
+            )
+            return expected_id
+
+        device_unique_id = sorted(mac_addresses)[0]
+        _LOGGER.debug("[get_device_unique_id] device_unique_id (first): %s", device_unique_id)
+        return device_unique_id
 
     @_log_errors
     async def get_system_info(self) -> MutableMapping[str, Any]:

tests/conftest.py

@@ -561,7 +561,7 @@ def __init__(self, **kwargs):
                 self._query_counts_reset = False
                 self._query_counts = (1, 1)
 
-            async def get_device_unique_id(self):
+            async def get_device_unique_id(self, expected_id: str | None = None):
                 return self._device_id
 
             async def get_host_firmware_version(self):
@@ -682,7 +682,7 @@ async def set_use_snake_case(self, initial: bool = False) -> None:
             async def get_system_info(self) -> dict:
                 return {"name": "OPNsense"}
 
-            async def get_device_unique_id(self) -> str:
+            async def get_device_unique_id(self, expected_id: str | None = None) -> str:
                 return self._device_id
 
             async def is_plugin_installed(self) -> bool:

tests/test_integration.py

@@ -61,7 +61,7 @@ async def is_plugin_installed(self) -> bool:  # for SYNC_ITEMS_REQUIRING_PLUGIN
     async def get_system_info(self) -> MutableMapping[str, Any]:
         return {"name": "OPNsenseTest"}
 
-    async def get_device_unique_id(self) -> str:
+    async def get_device_unique_id(self, expected_id: str | None = None) -> str:
         return self._device_id
 
     async def get_arp_table(self, resolve_hostnames: bool = False) -> list[dict[str, Any]]:
@@ -80,7 +80,9 @@ def __init__(self, device_id: str = "dev-runtime", firmware: str = "25.1") -> No
         self._firmware = firmware
         self._closed = False
 
-    async def get_device_unique_id(self) -> str:  # used by setup & coordinator
+    async def get_device_unique_id(
+        self, expected_id: str | None = None
+    ) -> str:  # used by setup & coordinator
         return self._device_id
 
     async def get_host_firmware_version(self) -> str:  # used by setup & coordinator

tests/test_pyopnsense.py

@@ -3296,6 +3296,29 @@ async def test_get_device_unique_id_no_mac(make_client) -> None:
     await client.async_close()
 
 
+@pytest.mark.asyncio
+async def test_get_device_unique_id_expected(make_client) -> None:
+    """get_device_unique_id returns expected_id if present even if not the first."""
+    session = MagicMock(spec=aiohttp.ClientSession)
+    client = make_client(session=session)
+    # aa_bb_cc is smaller than bb_cc_dd
+    client._safe_list_get = AsyncMock(
+        return_value=[
+            {"is_physical": True, "macaddr_hw": "aa:bb:cc"},
+            {"is_physical": True, "macaddr_hw": "bb:cc:dd"},
+        ]
+    )
+    # Without expected_id, it returns the first one (aa_bb_cc)
+    assert await client.get_device_unique_id() == "aa_bb_cc"
+
+    # With expected_id bb_cc_dd, it returns bb_cc_dd even if aa_bb_cc is smaller
+    assert await client.get_device_unique_id(expected_id="bb_cc_dd") == "bb_cc_dd"
+
+    # With expected_id not present, it returns the first one
+    assert await client.get_device_unique_id(expected_id="cc_dd_ee") == "aa_bb_cc"
+    await client.async_close()
+
+
 @pytest.mark.asyncio
 async def test_get_firewall_legacy_fallback(make_client) -> None:
     """get_firewall falls back to legacy config for OPNsense < 26.1.1."""