diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py index e0c1dcfe..2991e77a 100644 --- a/bellows/ezsp/__init__.py +++ b/bellows/ezsp/__init__.py @@ -659,7 +659,14 @@ async def send_xncp_frame( if status != t.EmberStatus.SUCCESS: raise InvalidCommandError("XNCP is not supported") - rsp_frame = xncp.XncpCommand.from_bytes(data) + try: + rsp_frame = xncp.XncpCommand.from_bytes(data) + except ValueError: + raise InvalidCommandError(f"Invalid XNCP response: {data!r}") + + if isinstance(rsp_frame.payload, xncp.Unknown): + raise InvalidCommandError(f"XNCP firmware does not support {payload}") + LOGGER.debug("Received XNCP frame: %s", rsp_frame) if rsp_frame.status != t.EmberStatus.SUCCESS: diff --git a/bellows/ezsp/xncp.py b/bellows/ezsp/xncp.py index f09c1064..ca9da4dd 100644 --- a/bellows/ezsp/xncp.py +++ b/bellows/ezsp/xncp.py @@ -159,3 +159,8 @@ class GetFlowControlTypeReq(XncpCommandPayload): @register_command(XncpCommandId.GET_FLOW_CONTROL_TYPE_RSP) class GetFlowControlTypeRsp(XncpCommandPayload): flow_control_type: FlowControlType + + +@register_command(XncpCommandId.UNKNOWN) +class Unknown(XncpCommandPayload): + pass diff --git a/tests/test_xncp.py b/tests/test_xncp.py index 60a7daee..61f5b7df 100644 --- a/tests/test_xncp.py +++ b/tests/test_xncp.py @@ -34,6 +34,37 @@ async def test_xncp_failure(ezsp_f: EZSP) -> None: ] +async def test_xncp_failure_multiprotocol(ezsp_f: EZSP) -> None: + """Test XNCP failure with multiprotocol firmware.""" + ezsp_f._mock_commands["customFrame"] = customFrame = AsyncMock( + return_value=[t.EmberStatus.SUCCESS, b""] + ) + + with pytest.raises(InvalidCommandError): + await ezsp_f.xncp_get_supported_firmware_features() + + assert customFrame.mock_calls == [ + call(xncp.XncpCommand.from_payload(xncp.GetSupportedFeaturesReq()).serialize()) + ] + + +async def test_xncp_failure_unknown(ezsp_f: EZSP) -> None: + """Test XNCP failure, unknown command.""" + ezsp_f._mock_commands["customFrame"] = customFrame = AsyncMock( + return_value=[ + t.EmberStatus.SUCCESS, + xncp.XncpCommand.from_payload(xncp.Unknown()).serialize(), + ] + ) + + with pytest.raises(InvalidCommandError): + await ezsp_f.xncp_get_supported_firmware_features() + + assert customFrame.mock_calls == [ + call(xncp.XncpCommand.from_payload(xncp.GetSupportedFeaturesReq()).serialize()) + ] + + async def test_xncp_get_supported_firmware_features(ezsp_f: EZSP) -> None: """Test XNCP get_supported_firmware_features.""" ezsp_f._mock_commands["customFrame"] = customFrame = AsyncMock(