OTA fixes (and more)

.pre-commit-config.yaml

@@ -12,7 +12,8 @@ repos:
       - id: mypy
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.3.4
+    rev: v0.4.4
     hooks:
       - id: ruff
-        args: ["--fix", "--exit-non-zero-on-fix", "--config", "pyproject.toml"]
+        args: [--fix]
+      - id: ruff-format

pyproject.toml

@@ -14,19 +14,17 @@ readme = "README.md"
 license = {text = "GPL-3.0"}
 requires-python = ">=3.12"
 dependencies = [
-    "bellows==0.38.1",
+    "bellows==0.39.0",
     "pyserial==3.5",
     "pyserial-asyncio==0.6",
-    "zha-quirks==0.0.112",
+    "zha-quirks==0.0.116",
     "zigpy-deconz==0.23.1",
-    "zigpy>=0.63.5",
+    "zigpy==0.64.0",
     "zigpy-xbee==0.20.1",
     "zigpy-zigate==0.12.0",
     "zigpy-znp==0.12.1",
-    "universal-silabs-flasher==0.0.18",
+    "universal-silabs-flasher==0.0.20",
     "pyserial-asyncio-fast==0.11",
-    "python-slugify==8.0.4",
-    "awesomeversion==24.2.0",
 ]
 
 [tool.setuptools.packages.find]

tests/common.py

@@ -1,19 +1,18 @@
 """Common test objects."""
 
 import asyncio
-from collections.abc import Awaitable
+from collections.abc import Awaitable, Callable
 import logging
 from typing import Any, Optional
 from unittest.mock import AsyncMock, Mock
 
-from slugify import slugify
 import zigpy.types as t
 import zigpy.zcl
 import zigpy.zcl.foundation as zcl_f
 
 from zha.application import Platform
 from zha.application.gateway import Gateway
-from zha.application.platforms import PlatformEntity
+from zha.application.platforms import BaseEntity, GroupEntity, PlatformEntity
 from zha.zigbee.device import Device
 from zha.zigbee.group import Group
 
@@ -167,12 +166,15 @@ def reset_clusters(clusters: list[zigpy.zcl.Cluster]) -> None:
         cluster.write_attributes.reset_mock()
 
 
-def find_entity(device: Device, platform: Platform) -> Optional[PlatformEntity]:
+def find_entity(device: Device, platform: Platform) -> PlatformEntity:
     """Find an entity for the specified platform on the given device."""
     for entity in device.platform_entities.values():
         if platform == entity.PLATFORM:
             return entity
-    return None
+
+    raise KeyError(
+        f"No entity found for platform {platform!r} on device {device}: {device.platform_entities}"
+    )
 
 
 def mock_coro(
@@ -187,71 +189,57 @@ def mock_coro(
     return fut
 
 
-def find_entity_id(
-    domain: str, zha_device: Device, qualifier: Optional[str] = None
-) -> Optional[str]:
-    """Find the entity id under the testing.
+def get_group_entity(
+    group: Group,
+    platform: Platform,
+    entity_type: type[BaseEntity] = BaseEntity,
+    qualifier: str | None = None,
+) -> GroupEntity:
+    """Get the first entity of the specified platform on the given group."""
+    for entity in group.group_entities.values():
+        if platform != entity.PLATFORM:
+            continue
 
-    This is used to get the entity id in order to get the state from the state
-    machine so that we can test state changes.
-    """
-    entities = find_entity_ids(domain, zha_device)
-    if not entities:
-        return None
-    if qualifier:
-        for entity_id in entities:
-            if qualifier in entity_id:
-                return entity_id
-        return None
-    else:
-        return entities[0]
+        if not isinstance(entity, entity_type):
+            continue
 
+        if qualifier is not None and qualifier not in entity.info_object.unique_id:
+            continue
 
-def find_entity_ids(
-    domain: str, zha_device: Device, omit: Optional[list[str]] = None
-) -> list[str]:
-    """Find the entity ids under the testing.
+        return entity
 
-    This is used to get the entity id in order to get the state from the state
-    machine so that we can test state changes.
-    """
-    ieeetail = "".join([f"{o:02x}" for o in zha_device.ieee[:4]])
-    head = f"{domain}.{slugify(f'{zha_device.name} {ieeetail}', separator='_')}"
-
-    entity_ids = [
-        f"{entity.PLATFORM}.{slugify(entity.name, separator='_')}"
-        for entity in zha_device.platform_entities.values()
-    ]
-
-    matches = []
-    res = []
-    for entity_id in entity_ids:
-        if entity_id.startswith(head):
-            matches.append(entity_id)
-
-    if omit:
-        for entity_id in matches:
-            skip = False
-            for o in omit:
-                if o in entity_id:
-                    skip = True
-                    break
-            if not skip:
-                res.append(entity_id)
-    else:
-        res = matches
-    return res
+    raise KeyError(
+        f"No {entity_type} entity found for platform {platform!r} on group {group}: {group.group_entities}"
+    )
 
 
-def async_find_group_entity_id(domain: str, group: Group) -> Optional[str]:
-    """Find the group entity id under test."""
-    entity_id = f"{domain}.{group.name.lower().replace(' ','_')}"
+def get_entity(
+    device: Device,
+    platform: Platform,
+    entity_type: type[BaseEntity] = BaseEntity,
+    exact_entity_type: type[BaseEntity] | None = None,
+    qualifier: str | None = None,
+    qualifier_func: Callable[[BaseEntity], bool] = lambda e: True,
+) -> PlatformEntity:
+    """Get the first entity of the specified platform on the given device."""
+    for entity in device.platform_entities.values():
+        if platform != entity.PLATFORM:
+            continue
+
+        if not isinstance(entity, entity_type):
+            continue
+
+        if exact_entity_type is not None and type(entity) is not exact_entity_type:
+            continue
 
-    entity_ids = [
-        f"{entity.PLATFORM}.{slugify(entity.name, separator='_')}"
-        for entity in group.group_entities.values()
-    ]
+        if qualifier is not None and qualifier not in entity.info_object.unique_id:
+            continue
 
-    if entity_id in entity_ids:
-        return entity_id
-    return None
+        if not qualifier_func(entity):
+            continue
+
+        return entity
+
+    raise KeyError(
+        f"No {entity_type} entity found for platform {platform!r} on device {device}: {device.platform_entities}"
+    )

tests/test_alarm_control_panel.py

@@ -10,13 +10,12 @@
 from zigpy.zcl.clusters import security
 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.gateway import Gateway
 from zha.application.platforms.alarm_control_panel import AlarmControlPanel
 from zha.application.platforms.alarm_control_panel.const import AlarmState
 from zha.zigbee.device import Device
 
-from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
-
 _LOGGER = logging.getLogger(__name__)
 
 

tests/test_binary_sensor.py

@@ -8,15 +8,14 @@
 import zigpy.profiles.zha
 from zigpy.zcl.clusters import general, measurement, security
 
+from tests.common import find_entity, send_attributes_report, update_attribute_cache
+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 import PlatformEntity
 from zha.application.platforms.binary_sensor import IASZone, Occupancy
 from zha.zigbee.device import Device
 
-from .common import find_entity, send_attributes_report, update_attribute_cache
-from .conftest import SIG_EP_INPUT, SIG_EP_OUTPUT, SIG_EP_PROFILE, SIG_EP_TYPE
-
 DEVICE_IAS = {
     1: {
         SIG_EP_PROFILE: zigpy.profiles.zha.PROFILE_ID,

tests/test_button.py

@@ -5,7 +5,6 @@
 from unittest.mock import call, patch
 
 import pytest
-from slugify import slugify
 from zhaquirks.const import (
     DEVICE_TYPE,
     ENDPOINTS,
@@ -24,7 +23,7 @@
 from zigpy.zcl.clusters.manufacturer_specific import ManufacturerSpecificCluster
 import zigpy.zcl.foundation as zcl_f
 
-from tests.common import find_entity, find_entity_id, mock_coro, update_attribute_cache
+from tests.common import get_entity, mock_coro, update_attribute_cache
 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
@@ -120,8 +119,7 @@ async def test_button(
 
     zha_device, cluster = contact_sensor
     assert cluster is not None
-    entity: PlatformEntity = find_entity(zha_device, Platform.BUTTON)  # type: ignore
-    assert entity is not None
+    entity: PlatformEntity = get_entity(zha_device, Platform.BUTTON)
     assert isinstance(entity, Button)
     assert entity.PLATFORM == Platform.BUTTON
 
@@ -137,15 +135,6 @@ async def test_button(
         assert cluster.request.call_args[0][3] == 5  # duration in seconds
 
 
-def get_entity(zha_dev: Device, entity_id: str) -> PlatformEntity:
-    """Get entity."""
-    entities = {
-        entity.PLATFORM + "." + slugify(entity.name, separator="_"): entity
-        for entity in zha_dev.platform_entities.values()
-    }
-    return entities[entity_id]
-
-
 async def test_frost_unlock(
     zha_gateway: Gateway,
     tuya_water_valve: tuple[Device, general.Identify],  # pylint: disable=redefined-outer-name
@@ -154,11 +143,9 @@ async def test_frost_unlock(
 
     zha_device, cluster = tuya_water_valve
     assert cluster is not None
-    entity_id = find_entity_id(
-        Platform.BUTTON, zha_device, qualifier="reset_frost_lock"
+    entity: PlatformEntity = get_entity(
+        zha_device, platform=Platform.BUTTON, entity_type=WriteAttributeButton
     )
-    entity: PlatformEntity = get_entity(zha_device, entity_id)
-    assert entity is not None
     assert isinstance(entity, WriteAttributeButton)
 
     assert entity._attr_device_class == ButtonDeviceClass.RESTART
@@ -258,10 +245,7 @@ async def test_quirks_command_button(
 
     zha_device, cluster = custom_button_device
     assert cluster is not None
-    entity_id = find_entity_id(Platform.BUTTON, zha_device)
-    entity: PlatformEntity = get_entity(zha_device, entity_id)
-    assert isinstance(entity, Button)
-    assert entity is not None
+    entity: PlatformEntity = get_entity(zha_device, platform=Platform.BUTTON)
 
     with patch(
         "zigpy.zcl.Cluster.request",
@@ -283,10 +267,9 @@ async def test_quirks_write_attr_button(
 
     zha_device, cluster = custom_button_device
     assert cluster is not None
-    entity_id = find_entity_id(Platform.BUTTON, zha_device, qualifier="feed")
-    entity: PlatformEntity = get_entity(zha_device, entity_id)
-    assert isinstance(entity, WriteAttributeButton)
-    assert entity is not None
+    entity: PlatformEntity = get_entity(
+        zha_device, platform=Platform.BUTTON, entity_type=WriteAttributeButton
+    )
 
     assert cluster.get(cluster.AttributeDefs.feed.name) == 0