Fix invalid uniqueness constraint on unique_id (#159)

* Fix invalid uniqueness constraint on unique_id

* remove magic #

tests/test_alarm_control_panel.py

@@ -11,6 +11,7 @@
 import zigpy.zcl.foundation as zcl_f
 
 from tests.conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
+from zha.application import Platform
 from zha.application.gateway import Gateway
 from zha.application.platforms.alarm_control_panel import AlarmControlPanel
 from zha.application.platforms.alarm_control_panel.const import AlarmState
@@ -49,7 +50,7 @@ async def test_alarm_control_panel(
     zha_device: Device = await device_joined(zigpy_device)
     cluster: security.IasAce = zigpy_device.endpoints.get(1).ias_ace
     alarm_entity: AlarmControlPanel = zha_device.platform_entities.get(
-        "00:0d:6f:00:0a:90:69:e7-1"
+        (Platform.ALARM_CONTROL_PANEL, "00:0d:6f:00:0a:90:69:e7-1")
     )
     assert alarm_entity is not None
     assert isinstance(alarm_entity, AlarmControlPanel)

tests/test_device.py

@@ -17,6 +17,7 @@
 import zigpy.zdo.types as zdo_t
 
 from tests.conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_TYPE
+from zha.application import Platform
 from zha.application.const import (
     CLUSTER_COMMAND_SERVER,
     CLUSTER_COMMANDS_CLIENT,
@@ -766,27 +767,51 @@ async def test_device_properties(
     assert zha_device.sw_version is None
 
     assert len(zha_device.platform_entities) == 3
-    assert "00:0d:6f:00:0a:90:69:e7-3-0-lqi" in zha_device.platform_entities
-    assert "00:0d:6f:00:0a:90:69:e7-3-0-rssi" in zha_device.platform_entities
-    assert "00:0d:6f:00:0a:90:69:e7-3-6" in zha_device.platform_entities
+    assert (
+        Platform.SENSOR,
+        "00:0d:6f:00:0a:90:69:e7-3-0-lqi",
+    ) in zha_device.platform_entities
+    assert (
+        Platform.SENSOR,
+        "00:0d:6f:00:0a:90:69:e7-3-0-rssi",
+    ) in zha_device.platform_entities
+    assert (
+        Platform.SWITCH,
+        "00:0d:6f:00:0a:90:69:e7-3-6",
+    ) in zha_device.platform_entities
 
     assert isinstance(
-        zha_device.platform_entities["00:0d:6f:00:0a:90:69:e7-3-0-lqi"], LQISensor
+        zha_device.platform_entities[
+            (Platform.SENSOR, "00:0d:6f:00:0a:90:69:e7-3-0-lqi")
+        ],
+        LQISensor,
     )
     assert isinstance(
-        zha_device.platform_entities["00:0d:6f:00:0a:90:69:e7-3-0-rssi"], RSSISensor
+        zha_device.platform_entities[
+            (Platform.SENSOR, "00:0d:6f:00:0a:90:69:e7-3-0-rssi")
+        ],
+        RSSISensor,
     )
     assert isinstance(
-        zha_device.platform_entities["00:0d:6f:00:0a:90:69:e7-3-6"], Switch
+        zha_device.platform_entities[(Platform.SWITCH, "00:0d:6f:00:0a:90:69:e7-3-6")],
+        Switch,
     )
 
-    assert zha_device.get_platform_entity("00:0d:6f:00:0a:90:69:e7-3-0-lqi") is not None
+    assert (
+        zha_device.get_platform_entity(
+            Platform.SENSOR, "00:0d:6f:00:0a:90:69:e7-3-0-lqi"
+        )
+        is not None
+    )
     assert isinstance(
-        zha_device.get_platform_entity("00:0d:6f:00:0a:90:69:e7-3-0-lqi"), LQISensor
+        zha_device.get_platform_entity(
+            Platform.SENSOR, "00:0d:6f:00:0a:90:69:e7-3-0-lqi"
+        ),
+        LQISensor,
     )
 
     with pytest.raises(KeyError, match="Entity foo not found"):
-        zha_device.get_platform_entity("foo")
+        zha_device.get_platform_entity("bar", "foo")
 
     # test things are none when they aren't returned by Zigpy
     zigpy_dev.node_desc = None

zha/application/platforms/__init__.py

@@ -289,14 +289,14 @@ def __init__(
         self._device: Device = device
         self._endpoint = endpoint
         # we double create these in discovery tests because we reissue the create calls to count and prove them out
-        if self.unique_id not in self._device.platform_entities:
-            self._device.platform_entities[self.unique_id] = self
+        if (self.PLATFORM, self.unique_id) not in self._device.platform_entities:
+            self._device.platform_entities[(self.PLATFORM, self.unique_id)] = self
         else:
             _LOGGER.debug(
                 "Not registering entity %r, unique id %r already exists: %r",
                 self,
-                self.unique_id,
-                self._device.platform_entities[self.unique_id],
+                (self.PLATFORM, self.unique_id),
+                self._device.platform_entities[(self.PLATFORM, self.unique_id)],
             )
 
     @classmethod

zha/application/platforms/sensor/__init__.py

@@ -381,8 +381,8 @@ def __init__(
         self._device.gateway.global_updater.register_update_listener(self.update)
 
         # we double create these in discovery tests because we reissue the create calls to count and prove them out
-        if self.unique_id not in self._device.platform_entities:
-            self._device.platform_entities[self.unique_id] = self
+        if (self.PLATFORM, self.unique_id) not in self._device.platform_entities:
+            self._device.platform_entities[(self.PLATFORM, self.unique_id)] = self
 
     @functools.cached_property
     def identifiers(self) -> DeviceCounterSensorIdentifiers:

zha/zigbee/device.py

@@ -29,7 +29,7 @@
 import zigpy.zdo.types as zdo_types
 from zigpy.zdo.types import RouteStatus, _NeighborEnums
 
-from zha.application import discovery
+from zha.application import Platform, discovery
 from zha.application.const import (
     ATTR_ARGS,
     ATTR_ATTRIBUTE,
@@ -233,7 +233,7 @@ def __init__(
         self._checkins_missed_count: int = 0
         self._on_network: bool = True
 
-        self._platform_entities: dict[str, PlatformEntity] = {}
+        self._platform_entities: dict[tuple[Platform, str], PlatformEntity] = {}
         self.semaphore: asyncio.Semaphore = asyncio.Semaphore(3)
         self._zdo_handler: ZDOClusterHandler = ZDOClusterHandler(self)
         self.status: DeviceStatus = DeviceStatus.CREATED
@@ -494,13 +494,13 @@ def sw_version(self, sw_build_id: int) -> None:
         self._sw_build_id = sw_build_id
 
     @property
-    def platform_entities(self) -> dict[str, PlatformEntity]:
+    def platform_entities(self) -> dict[tuple[Platform, str], PlatformEntity]:
         """Return the platform entities for this device."""
         return self._platform_entities
 
-    def get_platform_entity(self, unique_id: str) -> PlatformEntity:
+    def get_platform_entity(self, platform: Platform, unique_id: str) -> PlatformEntity:
         """Get a platform entity by unique id."""
-        entity = self._platform_entities.get(unique_id)
+        entity = self._platform_entities.get((platform, unique_id))
         if entity is None:
             raise KeyError(f"Entity {unique_id} not found")
         return entity
@@ -672,8 +672,8 @@ def extended_device_info(self) -> ExtendedDeviceInfo:
             **self.device_info.__dict__,
             active_coordinator=self.is_active_coordinator,
             entities={
-                unique_id: platform_entity.info_object
-                for unique_id, platform_entity in self.platform_entities.items()
+                platform_entity.unique_id: platform_entity.info_object
+                for platform_entity in self.platform_entities.values()
             },
             neighbors=[
                 NeighborInfo(