Gracefully handle a few more XNCP parsing failures (#653)

Author: puddly

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:

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

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(