Fall back to the MAC address when a Bravia television reports no CID (#182949)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: Maciej Bieniek <bieniu@users.noreply.github.com>

homeassistant/components/braviatv/__init__.py

@@ -7,11 +7,12 @@ from pybravia import BraviaClient
 
 from homeassistant.components import ssdp
 from homeassistant.const import CONF_HOST, CONF_MAC, Platform
-from homeassistant.core import HomeAssistant
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
 from homeassistant.helpers.service_info.ssdp import SsdpServiceInfo
 
-from .const import CONF_USE_SSL
+from .const import CONF_USE_SSL, DOMAIN
 from .coordinator import BraviaTVConfigEntry, BraviaTVCoordinator
 
 PLATFORMS: Final[list[Platform]] = [
@@ -69,6 +70,44 @@ async def async_unload_entry(
     return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS)
 
 
+async def async_migrate_entry(
+    hass: HomeAssistant, config_entry: BraviaTVConfigEntry
+) -> bool:
+    """Migrate an old config entry."""
+    if config_entry.version == 1 and config_entry.minor_version == 1:
+        # Old entries from a TV without CID have an empty unique ID
+        new_unique_id = config_entry.unique_id or dr.format_mac(
+            config_entry.data[CONF_MAC]
+        )
+
+        if new_unique_id != config_entry.unique_id:
+            device_registry = dr.async_get(hass)
+            if device_entry := device_registry.async_get_device_by_identifier(
+                (DOMAIN, ""), config_entry.entry_id
+            ):
+                new_identifiers = device_entry.identifiers.copy()
+                new_identifiers.discard((DOMAIN, ""))
+                new_identifiers.add((DOMAIN, new_unique_id))
+                device_registry.async_update_device(
+                    device_entry.id, new_identifiers=new_identifiers
+                )
+
+            @callback
+            def update_unique_id(entity_entry: er.RegistryEntry) -> dict[str, str]:
+                """Prefix the entity unique ID, which was the suffix alone."""
+                return {"new_unique_id": f"{new_unique_id}{entity_entry.unique_id}"}
+
+            await er.async_migrate_entries(
+                hass, config_entry.entry_id, update_unique_id
+            )
+
+        hass.config_entries.async_update_entry(
+            config_entry, unique_id=new_unique_id, minor_version=2
+        )
+
+    return True
+
+
 async def update_listener(
     hass: HomeAssistant, config_entry: BraviaTVConfigEntry
 ) -> None:

homeassistant/components/braviatv/config_flow.py

@@ -19,6 +19,7 @@ from homeassistant.const import (
 )
 from homeassistant.helpers import instance_id
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
+from homeassistant.helpers.device_registry import format_mac
 from homeassistant.helpers.service_info.ssdp import (
     ATTR_UPNP_FRIENDLY_NAME,
     ATTR_UPNP_MODEL_NAME,
@@ -42,6 +43,7 @@ class BraviaTVConfigFlow(ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Bravia TV integration."""
 
     VERSION = 1
+    MINOR_VERSION = 2
 
     def __init__(self) -> None:
         """Initialize config flow."""
@@ -84,13 +86,24 @@ class BraviaTVConfigFlow(ConfigFlow, domain=DOMAIN):
         await self.async_connect_device()
 
         system_info = await self.client.get_system_info()
-        cid = system_info[ATTR_CID].lower()
+        mac = system_info[ATTR_MAC]
+        formatted_mac = format_mac(mac)
 
-        self.device_config[CONF_MAC] = system_info[ATTR_MAC]
+        # Some TVs return an empty CID
+        unique_id = system_info[ATTR_CID].lower() or formatted_mac
 
-        await self.async_set_unique_id(cid)
+        self.device_config[CONF_MAC] = mac
+
+        await self.async_set_unique_id(unique_id)
         self._abort_if_unique_id_configured()
 
+        # Disabled entries are not migrated, so they still have an empty unique ID
+        if any(
+            not entry.unique_id and format_mac(entry.data[CONF_MAC]) == formatted_mac
+            for entry in self._async_current_entries(include_ignore=False)
+        ):
+            return self.async_abort(reason="already_configured")
+
         return self.async_create_entry(
             title=f"{system_info['name']} {system_info[ATTR_MODEL]}",
             data=self.device_config,

homeassistant/components/braviatv/diagnostics.py

@@ -8,7 +8,7 @@ from homeassistant.core import HomeAssistant
 
 from .coordinator import BraviaTVConfigEntry
 
-TO_REDACT = {CONF_MAC, CONF_PIN, "macAddr"}
+TO_REDACT = {CONF_MAC, CONF_PIN, "macAddr", "unique_id"}
 
 
 async def async_get_config_entry_diagnostics(

tests/components/braviatv/snapshots/test_diagnostics.ambr

@@ -13,7 +13,7 @@
       }),
       'domain': 'braviatv',
       'entry_id': '3bd2acb0e4f0476d40865546d0d91921',
-      'minor_version': 1,
+      'minor_version': 2,
       'options': dict({
       }),
       'pref_disable_new_entities': False,
@@ -22,7 +22,7 @@
       'subentries': list([
       ]),
       'title': 'BRAVIA TV-Model',
-      'unique_id': 'very_unique_string',
+      'unique_id': '**REDACTED**',
       'version': 1,
     }),
     'device_info': dict({

tests/components/braviatv/test_config_flow.py

@@ -17,7 +17,7 @@ from homeassistant.components.braviatv.const import (
     DOMAIN,
     NICKNAME_PREFIX,
 )
-from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
+from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER, ConfigEntryDisabler
 from homeassistant.const import CONF_CLIENT_ID, CONF_HOST, CONF_MAC, CONF_PIN
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
@@ -455,3 +455,84 @@ async def test_reauth_successful(hass: HomeAssistant, use_psk, new_pin) -> None:
         assert result["type"] is FlowResultType.ABORT
         assert result["reason"] == "reauth_successful"
         assert config_entry.data[CONF_PIN] == new_pin
+
+
+@pytest.mark.parametrize(
+    (
+        "existing_unique_id",
+        "existing_mac",
+        "disabled_by",
+        "expected_type",
+        "expected_unique_ids",
+    ),
+    [
+        pytest.param(
+            "",
+            "11:22:33:44:55:66",
+            None,
+            FlowResultType.CREATE_ENTRY,
+            ["11:22:33:44:55:66", "aa:bb:cc:dd:ee:ff"],
+            id="other_tv_without_cid",
+        ),
+        pytest.param(
+            "aa:bb:cc:dd:ee:ff",
+            "AA:BB:CC:DD:EE:FF",
+            None,
+            FlowResultType.ABORT,
+            ["aa:bb:cc:dd:ee:ff"],
+            id="same_tv_migrated",
+        ),
+        pytest.param(
+            "",
+            "AA:BB:CC:DD:EE:FF",
+            ConfigEntryDisabler.USER,
+            FlowResultType.ABORT,
+            [""],
+            id="same_tv_disabled",
+        ),
+    ],
+)
+async def test_user_flow_with_empty_cid(
+    hass: HomeAssistant,
+    existing_unique_id: str,
+    existing_mac: str,
+    disabled_by: ConfigEntryDisabler | None,
+    expected_type: FlowResultType,
+    expected_unique_ids: list[str],
+) -> None:
+    """Test a television that reports an empty CID is identified by its MAC."""
+    MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=existing_unique_id,
+        data={CONF_HOST: "other-host", CONF_MAC: existing_mac},
+        disabled_by=disabled_by,
+    ).add_to_hass(hass)
+
+    with (
+        patch("pybravia.BraviaClient.connect"),
+        patch("pybravia.BraviaClient.pair"),
+        patch("pybravia.BraviaClient.set_wol_mode"),
+        patch(
+            "pybravia.BraviaClient.get_system_info",
+            return_value=BRAVIA_SYSTEM_INFO
+            | {"cid": "", "macAddr": "AA:BB:CC:DD:EE:FF"},
+        ),
+    ):
+        result = await hass.config_entries.flow.async_init(
+            DOMAIN, context={"source": SOURCE_USER}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={CONF_HOST: "bravia-host"}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={CONF_USE_PSK: True, CONF_USE_SSL: False}
+        )
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"], user_input={CONF_PIN: "secret"}
+        )
+
+    assert result["type"] is expected_type
+    assert (
+        sorted(entry.unique_id for entry in hass.config_entries.async_entries(DOMAIN))
+        == expected_unique_ids
+    )

tests/components/braviatv/test_init.py

@@ -0,0 +1,117 @@
+"""Test the Bravia TV integration setup."""
+
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.braviatv.const import CONF_USE_PSK, DOMAIN
+from homeassistant.const import CONF_HOST, CONF_MAC, CONF_PIN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from tests.common import MockConfigEntry
+
+ENTRY_DATA = {
+    CONF_HOST: "bravia-host",
+    CONF_MAC: "AA:BB:CC:DD:EE:FF",
+    CONF_PIN: "secret",
+    CONF_USE_PSK: True,
+}
+
+
+@pytest.mark.parametrize(
+    ("unique_id", "mac", "expected_unique_id"),
+    [
+        pytest.param("", "AA:BB:CC:DD:EE:FF", "aa:bb:cc:dd:ee:ff", id="empty_cid"),
+        pytest.param(
+            "very_unique_string",
+            "AA:BB:CC:DD:EE:FF",
+            "very_unique_string",
+            id="existing_cid",
+        ),
+        pytest.param("", "", "", id="empty_cid_and_mac"),
+    ],
+)
+async def test_migrate_unique_id(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    unique_id: str,
+    mac: str,
+    expected_unique_id: str,
+) -> None:
+    """Test only an entry stored without a CID gets a new unique ID."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        unique_id=unique_id,
+        data=ENTRY_DATA | {CONF_MAC: mac},
+        version=1,
+        minor_version=1,
+    )
+    config_entry.add_to_hass(hass)
+    entity = entity_registry.async_get_or_create(
+        "media_player", DOMAIN, unique_id, config_entry=config_entry
+    )
+
+    with (
+        patch("homeassistant.components.braviatv.BraviaClient", autospec=True),
+        patch(
+            "homeassistant.components.braviatv.coordinator.BraviaTVCoordinator._async_update_data",
+            return_value=None,
+        ),
+    ):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert config_entry.unique_id == expected_unique_id
+    assert config_entry.minor_version == 2
+
+    migrated_entity = entity_registry.async_get(entity.entity_id)
+    assert migrated_entity is not None
+    assert migrated_entity.unique_id == expected_unique_id
+
+
+async def test_migration_keeps_entities_and_device(
+    hass: HomeAssistant,
+    entity_registry: er.EntityRegistry,
+    device_registry: dr.DeviceRegistry,
+) -> None:
+    """Test entities and the device move along and keep their entity IDs."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN, unique_id="", data=ENTRY_DATA, version=1, minor_version=1
+    )
+    config_entry.add_to_hass(hass)
+
+    device = device_registry.async_get_or_create(
+        config_entry_id=config_entry.entry_id, identifiers={(DOMAIN, "")}
+    )
+    media_player = entity_registry.async_get_or_create(
+        "media_player", DOMAIN, "", config_entry=config_entry, device_id=device.id
+    )
+    button = entity_registry.async_get_or_create(
+        "button", DOMAIN, "_reboot", config_entry=config_entry, device_id=device.id
+    )
+
+    with (
+        patch("homeassistant.components.braviatv.BraviaClient", autospec=True),
+        patch(
+            "homeassistant.components.braviatv.coordinator.BraviaTVCoordinator._async_update_data",
+            return_value=None,
+        ),
+    ):
+        assert await hass.config_entries.async_setup(config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert config_entry.unique_id == "aa:bb:cc:dd:ee:ff"
+    assert config_entry.minor_version == 2
+
+    migrated_media_player = entity_registry.async_get(media_player.entity_id)
+    assert migrated_media_player is not None
+    assert migrated_media_player.unique_id == "aa:bb:cc:dd:ee:ff"
+
+    migrated_button = entity_registry.async_get(button.entity_id)
+    assert migrated_button is not None
+    assert migrated_button.unique_id == "aa:bb:cc:dd:ee:ff_reboot"
+
+    assert device_registry.async_get(device.id).identifiers == {
+        (DOMAIN, "aa:bb:cc:dd:ee:ff")
+    }