Codex updated: Fix IndexError in issue_cluster_command for IAS ACE arm response

tests/test_device.py

@@ -18,11 +18,11 @@
 import zigpy.types
 from zigpy.typing import UNDEFINED
 from zigpy.zcl import ClusterType
-from zigpy.zcl.clusters import general
+from zigpy.zcl.clusters import general, security
 from zigpy.zcl.clusters.general import Ota, PowerConfiguration
 from zigpy.zcl.clusters.lighting import Color
 from zigpy.zcl.clusters.measurement import CarbonDioxideConcentration
-from zigpy.zcl.foundation import Status, WriteAttributesResponse
+from zigpy.zcl.foundation import DefaultResponse, Status, WriteAttributesResponse
 from zigpy.zcl.helpers import ReportingConfig
 import zigpy.zdo.types as zdo_t
 
@@ -86,6 +86,21 @@ def zigpy_device(zha_gateway: Gateway, with_basic_cluster: bool = True, **kwargs
     return create_mock_zigpy_device(zha_gateway, endpoints, **kwargs)
 
 
+class _PlainStructResponse(zigpy.types.Struct):
+    payload: zigpy.types.uint16_t
+
+
+async def _issue_on_cluster_command(zha_device: Device) -> None:
+    await zha_device.issue_cluster_command(
+        3,
+        general.OnOff.cluster_id,
+        general.OnOff.ServerCommandDefs.on.id,
+        CLUSTER_COMMAND_SERVER,
+        None,
+        {},
+    )
+
+
 def zigpy_device_mains(zha_gateway: Gateway, with_basic_cluster: bool = True):
     """Return a ZigpyDevice with a switch cluster."""
     in_clusters = [general.OnOff.cluster_id]
@@ -566,7 +581,7 @@ async def test_write_zigbee_attribute(
 async def test_issue_cluster_command(
     zha_gateway: Gateway,
 ) -> None:
-    """Test issue_cluster_command method."""
+    """Test issuing cluster commands and validating their responses."""
     zigpy_dev = zigpy_device(zha_gateway, with_basic_cluster=True)
     zha_device = await join_zigpy_device(zha_gateway, zigpy_dev)
 
@@ -583,19 +598,79 @@ async def test_issue_cluster_command(
             None,
         )
 
-    cluster = zigpy_dev.endpoints[3].on_off
+    successful_response = DefaultResponse(
+        command_id=general.OnOff.ServerCommandDefs.on.id,
+        status=Status.SUCCESS,
+    )
+    with patch(
+        "zigpy.zcl.Cluster.request", return_value=successful_response
+    ) as request_mock:
+        await _issue_on_cluster_command(zha_device)
+    request_mock.assert_awaited_once()
+
+    failure_response = DefaultResponse(
+        command_id=general.OnOff.ServerCommandDefs.on.id,
+        status=Status.FAILURE,
+    )
+    with (
+        patch("zigpy.zcl.Cluster.request", return_value=failure_response),
+        pytest.raises(
+            ZHAException, match="Failed to issue cluster command with status"
+        ),
+    ):
+        await _issue_on_cluster_command(zha_device)
 
-    with patch("zigpy.zcl.Cluster.request", return_value=[0x5, Status.SUCCESS]):
+    plain_response = _PlainStructResponse(payload=0x1234)
+    parsed_response, remaining = _PlainStructResponse.deserialize(
+        plain_response.serialize()
+    )
+    assert remaining == b""
+    with patch("zigpy.zcl.Cluster.request", return_value=parsed_response):
+        await _issue_on_cluster_command(zha_device)
+
+    for malformed_response in (
+        0,
+        (0, Status.SUCCESS),
+        [0, Status.SUCCESS],
+        b"\x00\x00",
+    ):
+        with (
+            patch("zigpy.zcl.Cluster.request", return_value=malformed_response),
+            pytest.raises(ZHAException, match="unexpected response"),
+        ):
+            await _issue_on_cluster_command(zha_device)
+
+
+async def test_issue_cluster_command_accepts_typed_ias_ace_arm_response(
+    zha_gateway: Gateway,
+) -> None:
+    """Test a one-field IAS ACE arm response is a valid command payload."""
+    zigpy_dev = zigpy_device(zha_gateway)
+    zigpy_dev.endpoints[3].add_input_cluster(security.IasAce.cluster_id)
+    zha_device = await join_zigpy_device(zha_gateway, zigpy_dev)
+
+    arm_response = security.IasAce.ClientCommandDefs.arm_response.schema(
+        arm_notification=security.IasAce.ArmNotification.All_Zones_Armed,
+    )
+    assert len(arm_response) == 1
+    assert not hasattr(arm_response, "status")
+
+    with patch("zigpy.zcl.Cluster.request", return_value=arm_response) as request_mock:
         await zha_device.issue_cluster_command(
             3,
-            general.OnOff.cluster_id,
-            general.OnOff.ServerCommandDefs.on.id,
+            security.IasAce.cluster_id,
+            security.IasAce.ServerCommandDefs.arm.id,
             CLUSTER_COMMAND_SERVER,
             None,
-            {},
+            {
+                "arm_mode": security.IasAce.ArmMode.Arm_All_Zones,
+                "arm_disarm_code": "1234",
+                "zone_id": 0,
+            },
         )
 
-        assert cluster.request.await_count == 1
+    request_mock.assert_awaited_once()
+    assert request_mock.await_args.kwargs["expect_reply"] is True
 
 
 async def test_async_add_to_group_remove_from_group(

zha/zigbee/device.py

@@ -20,7 +20,7 @@
 from zigpy.device import Device as ZigpyDevice
 import zigpy.exceptions
 from zigpy.profiles import PROFILES
-from zigpy.types import uint1_t, uint8_t, uint16_t
+from zigpy.types import Struct, uint1_t, uint8_t, uint16_t
 from zigpy.types.named import EUI64, NWK, ExtendedPanId
 from zigpy.typing import UNDEFINED, UndefinedType
 import zigpy.zcl
@@ -1451,10 +1451,17 @@ async def issue_cluster_command(
             return  # client commands don't return a response
         if isinstance(response, Exception):
             raise ZHAException("Failed to issue cluster command") from response
-        if response[1] is not ZclStatus.SUCCESS:
+        if not isinstance(response, Struct):
             raise ZHAException(
-                f"Failed to issue cluster command with status: {response[1]}"
+                f"Failed to issue cluster command with unexpected response: {response!r}"
             )
+        status = getattr(response, "status", None)
+        if (
+            isinstance(status, Enum)
+            and isinstance(status, int)
+            and int(status) != int(ZclStatus.SUCCESS)
+        ):
+            raise ZHAException(f"Failed to issue cluster command with status: {status}")
 
     async def async_add_to_group(self, group_id: int) -> None:
         """Add this device to the provided zigbee group."""