velbus: fix unique_id collision for Property sensor entities (#176343)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

homeassistant/components/velbus/__init__.py

@@ -8,12 +8,17 @@ import shutil
 
 from velbusaio.controller import Velbus
 from velbusaio.exceptions import VelbusConnectionFailed
+from velbusaio.helpers import get_property_key_map
 
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import CONF_PORT, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady, PlatformNotReady
-from homeassistant.helpers import config_validation as cv, device_registry as dr
+from homeassistant.helpers import (
+    config_validation as cv,
+    device_registry as dr,
+    entity_registry as er,
+)
 from homeassistant.helpers.storage import STORAGE_DIR
 from homeassistant.helpers.typing import ConfigType
 
@@ -87,6 +92,47 @@ def _migrate_device_identifiers(hass: HomeAssistant, entry_id: str) -> None:
             dev_reg.async_update_device(device.id, new_identifiers=new_identifier)
 
 
+async def _migrate_property_unique_ids(hass: HomeAssistant, entry_id: str) -> None:
+    """Ensure property entity unique_ids use {serial}-{property_key} format."""
+    ent_reg = er.async_get(hass)
+
+    property_key_map = await hass.async_add_executor_job(get_property_key_map)
+    for entry in er.async_entries_for_config_entry(ent_reg, entry_id):
+        if not entry.original_name:
+            continue
+        property_key = property_key_map.get(entry.original_name)
+        if property_key is None:
+            continue
+        # Derive the serial from the entity's own unique_id, not from the device
+        # registry, which another integration could overwrite. The program select
+        # historically used `{serial}-{channel}-program_select`; every other property
+        # uses channel number 0 (`{serial}-0`). Regular channels are always >=1, so a
+        # `-0` suffix and the `-program_select` suffix only ever belong to properties.
+        if entry.unique_id.endswith("-program_select"):
+            serial = entry.unique_id.removesuffix("-program_select").rsplit("-", 1)[0]
+        elif entry.unique_id.endswith("-0"):
+            serial = entry.unique_id.removesuffix("-0")
+        else:
+            continue
+
+        expected_unique_id = f"{serial}-{property_key}"
+        if ent_reg.async_get_entity_id(entry.domain, DOMAIN, expected_unique_id):
+            # Target unique_id already exists (created by new code) — remove stale entry
+            _LOGGER.debug(
+                "Removing stale entity %s with outdated unique_id %s",
+                entry.entity_id,
+                entry.unique_id,
+            )
+            ent_reg.async_remove(entry.entity_id)
+        else:
+            _LOGGER.debug(
+                "Migrating unique_id %s → %s", entry.unique_id, expected_unique_id
+            )
+            ent_reg.async_update_entity(
+                entry.entity_id, new_unique_id=expected_unique_id
+            )
+
+
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the actions for the Velbus component."""
     async_setup_services(hass)
@@ -108,11 +154,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: VelbusConfigEntry) -> bo
             translation_key="connection_failed",
         ) from error
 
+    _migrate_device_identifiers(hass, entry.entry_id)
+    # Migrate unique ids before the bus scan to preserve entity history
+    await _migrate_property_unique_ids(hass, entry.entry_id)
+
     task = hass.async_create_task(velbus_scan_task(controller, hass, entry.entry_id))
     entry.runtime_data = VelbusData(controller=controller, scan_task=task)
 
-    _migrate_device_identifiers(hass, entry.entry_id)
-
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
     return True

homeassistant/components/velbus/entity.py

@@ -32,8 +32,15 @@ class VelbusEntity(Entity):
         self._channel = channel
         self._module_address = str(channel.get_module_address())
         self._attr_name = channel.get_name()
-        serial = channel.get_module_serial() or self._module_address
-        self._attr_unique_id = f"{serial}-{channel.get_channel_number()}"
+        serial = channel.get_module_serial()
+        # Modules like the VMB4RY report a serial of "0"; fall back to the module
+        # address so entities on multiple such modules keep distinct unique ids.
+        if serial in (None, "", "0"):
+            serial = self._module_address
+        if isinstance(channel, VelbusProperty):
+            self._attr_unique_id = f"{serial}-{channel.get_property_key()}"
+        else:
+            self._attr_unique_id = f"{serial}-{channel.get_channel_number()}"
 
     def _get_identifier(self) -> str:
         """Return the identifier of the entity."""

homeassistant/components/velbus/select.py

@@ -42,7 +42,6 @@ class VelbusSelect(VelbusEntity, SelectEntity):
         """Initialize a select Velbus entity."""
         super().__init__(channel)
         self._attr_options = self._channel.get_options()
-        self._attr_unique_id = f"{self._attr_unique_id}-program_select"  # pylint: disable=home-assistant-entity-unique-id-redundant-platform
 
     @api_call
     @override

tests/components/velbus/conftest.py

@@ -177,6 +177,7 @@ def mock_select() -> AsyncMock:
     channel = AsyncMock(spec=SelectedProgram)
     channel.get_categories.return_value = ["select"]
     channel.get_name.return_value = "select"
+    channel.get_property_key.return_value = "SelectedProgram"
     channel.get_module_address.return_value = 88
     channel.get_channel_number.return_value = 33
     channel.get_module_type_name.return_value = "VMB4RYNO"
@@ -241,8 +242,10 @@ def mock_lightsensor() -> AsyncMock:
     channel = AsyncMock(spec=LightValue)
     channel.get_categories.return_value = ["sensor"]
     channel.get_name.return_value = "LightSensor"
+    channel.get_property_key.return_value = "LightValue"
     channel.get_module_address.return_value = 2
-    channel.get_channel_number.return_value = 4
+    # Properties always report channel number 0 (Property.get_channel_number)
+    channel.get_channel_number.return_value = 0
     channel.get_module_type_name.return_value = "VMB7IN"
     channel.get_module_type.return_value = 8
     channel.get_full_name.return_value = "Input"

tests/components/velbus/snapshots/test_select.ambr

@@ -39,7 +39,7 @@
     'suggested_object_id': None,
     'supported_features': 0,
     'translation_key': 'select_program',
-    'unique_id': 'qwerty1234567-33-program_select',
+    'unique_id': 'qwerty1234567-SelectedProgram',
     'unit_of_measurement': None,
   })
 # ---

tests/components/velbus/snapshots/test_sensor.ambr

@@ -151,7 +151,7 @@
     'suggested_object_id': None,
     'supported_features': 0,
     'translation_key': None,
-    'unique_id': 'a1b2c3d4e5f6-4',
+    'unique_id': 'a1b2c3d4e5f6-LightValue',
     'unit_of_measurement': 'illuminance',
   })
 # ---

tests/components/velbus/test_entity.py

@@ -0,0 +1,34 @@
+"""Tests for the Velbus entity base class."""
+
+from unittest.mock import AsyncMock
+
+import pytest
+from velbusaio.channels import Channel as VelbusChannel
+
+from homeassistant.components.velbus.entity import VelbusEntity
+
+
+@pytest.mark.parametrize(
+    ("module_serial", "expected_unique_id"),
+    [
+        ("a1b2c3d4e5f6", "a1b2c3d4e5f6-2"),
+        (None, "5-2"),
+        ("", "5-2"),
+        # Modules like the VMB4RY report a serial of "0"; without a fallback two
+        # such modules would share the same unique_id.
+        ("0", "5-2"),
+    ],
+)
+def test_unique_id_falls_back_to_module_address(
+    module_serial: str | None, expected_unique_id: str
+) -> None:
+    """Test that a missing or "0" module serial falls back to the module address."""
+    channel = AsyncMock(spec=VelbusChannel)
+    channel.get_module_address.return_value = 5
+    channel.get_channel_number.return_value = 2
+    channel.get_module_serial.return_value = module_serial
+    channel.get_name.return_value = "channel"
+
+    entity = VelbusEntity(channel)
+
+    assert entity.unique_id == expected_unique_id

tests/components/velbus/test_init.py

@@ -308,3 +308,211 @@ async def test_remove_config_entry_device_detaches_subdevices(
         config_entry.entry_id not in sub_device_after.config_entries
         and sub_device_after.via_device_id is None
     )
+
+
+# velbus-aio maps both the spec key and the display name to the class name, because
+# Property.get_name() returned the spec key before velbus-aio 2026.4.1 and the display
+# name from that release onwards; both forms exist as original_name in the wild.
+_PROPERTY_KEY_MAP = {
+    "selected_program": "SelectedProgram",
+    "Selected program": "SelectedProgram",
+    "light_value": "LightValue",
+    "Light value": "LightValue",
+}
+
+
+@pytest.mark.parametrize(
+    ("domain", "device_serial", "old_unique_id", "original_name", "expected_unique_id"),
+    [
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-0-program_select",
+            "selected_program",
+            "test_serial-SelectedProgram",
+            id="rename_select_spec_key",
+        ),
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-0-program_select",
+            "Selected program",
+            "test_serial-SelectedProgram",
+            id="rename_select_display_name",
+        ),
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-5-program_select",
+            "selected_program",
+            "test_serial-SelectedProgram",
+            id="rename_select_legacy_channel",
+        ),
+        pytest.param(
+            "sensor",
+            "test_serial",
+            "test_serial-0",
+            "light_value",
+            "test_serial-LightValue",
+            id="rename_sensor",
+        ),
+        pytest.param(
+            "sensor",
+            "overwritten_serial",
+            "test_serial-0",
+            "light_value",
+            "test_serial-LightValue",
+            id="serial_taken_from_unique_id_not_device",
+        ),
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-SelectedProgram",
+            "selected_program",
+            "test_serial-SelectedProgram",
+            id="already_correct",
+        ),
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-old_format",
+            None,
+            "test_serial-old_format",
+            id="skipped_without_name",
+        ),
+        pytest.param(
+            "select",
+            "test_serial",
+            "test_serial-old_format",
+            "not_a_property",
+            "test_serial-old_format",
+            id="skipped_unknown_name",
+        ),
+        pytest.param(
+            "sensor",
+            "test_serial",
+            "test_serial-3",
+            "light_value",
+            "test_serial-3",
+            id="skipped_colliding_channel",
+        ),
+    ],
+)
+async def test_migrate_property_unique_ids(
+    hass: HomeAssistant,
+    config_entry: VelbusConfigEntry,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    controller: MagicMock,
+    domain: str,
+    device_serial: str,
+    old_unique_id: str,
+    original_name: str | None,
+    expected_unique_id: str,
+) -> None:
+    """Test the property unique_id migration for every legacy and skip case."""
+    device = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id,
+        identifiers={(DOMAIN, "1")},
+        serial_number=device_serial,
+    )
+    entity = entity_registry.async_get_or_create(
+        domain,
+        DOMAIN,
+        old_unique_id,
+        config_entry=config_entry,
+        device_id=device.id,
+        original_name=original_name,
+    )
+
+    with patch(
+        "homeassistant.components.velbus.get_property_key_map",
+        return_value=_PROPERTY_KEY_MAP,
+    ):
+        await init_integration(hass, config_entry)
+
+    migrated = entity_registry.async_get(entity.entity_id)
+    assert migrated
+    assert migrated.unique_id == expected_unique_id
+
+
+async def test_migrate_property_unique_ids_remove_stale(
+    hass: HomeAssistant,
+    config_entry: VelbusConfigEntry,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    controller: MagicMock,
+) -> None:
+    """Test that a stale property entity is removed when the correct one already exists."""
+    device = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id,
+        identifiers={(DOMAIN, "1")},
+        serial_number="test_serial",
+    )
+    entity_registry.async_get_or_create(
+        "select",
+        DOMAIN,
+        "test_serial-SelectedProgram",
+        config_entry=config_entry,
+        device_id=device.id,
+        original_name="selected_program",
+    )
+    entity_registry.async_get_or_create(
+        "select",
+        DOMAIN,
+        "test_serial-0-program_select",
+        config_entry=config_entry,
+        device_id=device.id,
+        original_name="selected_program",
+    )
+
+    with patch(
+        "homeassistant.components.velbus.get_property_key_map",
+        return_value=_PROPERTY_KEY_MAP,
+    ):
+        await init_integration(hass, config_entry)
+
+    assert not entity_registry.async_get_entity_id(
+        "select", DOMAIN, "test_serial-0-program_select"
+    )
+    assert entity_registry.async_get_entity_id(
+        "select", DOMAIN, "test_serial-SelectedProgram"
+    )
+
+
+async def test_migrate_property_unique_ids_preserves_entity_id(
+    hass: HomeAssistant,
+    config_entry: VelbusConfigEntry,
+    device_registry: dr.DeviceRegistry,
+    entity_registry: er.EntityRegistry,
+    controller: MagicMock,
+) -> None:
+    """Test that a migrated property keeps its entity_id once the bus scan registers it."""
+    device = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id,
+        identifiers={(DOMAIN, "2")},
+        serial_number="a1b2c3d4e5f6",
+    )
+    # Same serial as the scanned LightValue property, so migrating before the scan makes
+    # the scan reuse this entry instead of registering a second one.
+    legacy_entity = entity_registry.async_get_or_create(
+        "sensor",
+        DOMAIN,
+        "a1b2c3d4e5f6-0",
+        config_entry=config_entry,
+        device_id=device.id,
+        original_name="light_value",
+        suggested_object_id="legacy_light_value",
+    )
+    assert legacy_entity.entity_id == "sensor.legacy_light_value"
+
+    with patch(
+        "homeassistant.components.velbus.get_property_key_map",
+        return_value=_PROPERTY_KEY_MAP,
+    ):
+        await init_integration(hass, config_entry)
+
+    assert (
+        entity_registry.async_get_entity_id("sensor", DOMAIN, "a1b2c3d4e5f6-LightValue")
+        == "sensor.legacy_light_value"
+    )