Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion bellows/ezsp/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -770,7 +770,7 @@ async def send_xncp_frame(

LOGGER.debug("Received XNCP frame: %s", rsp_frame)

if t.sl_Status.from_ember_status(rsp_frame.status) != t.sl_Status.OK:
if rsp_frame.status != xncp.XncpStatus.OK:
raise InvalidCommandError(f"XNCP response error: {rsp_frame.status}")

return rsp_frame.payload
Expand Down
35 changes: 25 additions & 10 deletions bellows/ezsp/xncp.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,7 @@

import zigpy.types as t

from bellows.types import (
EmberApsFrame,
EmberStatus,
EzspMfgTokenId,
RouteRecordStatus,
sl_Status,
)
from bellows.types import EmberApsFrame, EzspMfgTokenId, RouteRecordStatus, sl_Status

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -68,24 +62,45 @@ class XncpCommandId(t.enum16):
UNKNOWN = 0xFFFF


class XncpStatus(t.enum8):
"""Status byte of an XNCP frame.

The firmware types this byte as the SDK's native status type: an `EmberStatus` in
Gecko SDK 4.x builds, the low octet of an `sl_status_t` in Simplicity SDK builds.
The values current firmware emits happen not to overlap, so both encodings fit
into one enum, but only `OK` (zero under either) is relied upon: every other value
is treated as an opaque failure.
"""

OK = 0x00

# Gecko SDK 4.x: `EmberStatus`
EMBER_BAD_ARGUMENT = 0x02
EMBER_NOT_FOUND = 0x03

# Simplicity SDK: low octet of an `sl_status_t`
SL_STATUS_INVALID_PARAMETER = 0x21
SL_STATUS_NOT_FOUND = 0x2D


@dataclasses.dataclass
class XncpCommand:
command_id: XncpCommandId
status: EmberStatus
status: XncpStatus
payload: XncpCommandPayload

@classmethod
def from_payload(cls, payload: XncpCommandPayload) -> XncpCommand:
return cls(
command_id=REV_COMMANDS[type(payload)],
status=EmberStatus.SUCCESS,
status=XncpStatus.OK,
payload=payload,
)

@classmethod
def from_bytes(cls, data: bytes) -> XncpCommand:
command_id, data = XncpCommandId.deserialize(data)
status, data = EmberStatus.deserialize(data)
status, data = XncpStatus.deserialize(data)

if command_id not in COMMANDS:
raise ValueError(
Expand Down
36 changes: 35 additions & 1 deletion tests/test_xncp.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import logging
from unittest.mock import AsyncMock, call, patch

import pytest
Expand All @@ -17,7 +18,7 @@ async def test_xncp_failure(ezsp_f: EZSP) -> None:
command = xncp.XncpCommand.from_payload(
xncp.GetSupportedFeaturesRsp(features=xncp.FirmwareFeatures.MANUAL_SOURCE_ROUTE)
)
command.status = t.EmberStatus.ERR_FATAL
command.status = xncp.XncpStatus.EMBER_BAD_ARGUMENT

ezsp_f._mock_commands["customFrame"] = customFrame = AsyncMock(
return_value=[
Expand All @@ -34,6 +35,39 @@ async def test_xncp_failure(ezsp_f: EZSP) -> None:
]


@pytest.mark.parametrize(
"rsp",
[
# Gecko SDK 4.x firmware replies with an `EmberStatus`: `EMBER_NOT_FOUND`
b"\x02\x80\x03",
# Simplicity SDK firmware replies with the low octet of an
# `sl_status_t`: `SL_STATUS_NOT_FOUND`
b"\x02\x80\x2D",
],
)
async def test_xncp_missing_mfg_token_override(
ezsp_f: EZSP, rsp: bytes, caplog
) -> None:
"""Test that a token without an override fails quietly, regardless of the SDK."""
ezsp_f._mock_commands["customFrame"] = AsyncMock(
return_value=[t.EmberStatus.SUCCESS, rsp]
)
ezsp_f._mock_commands["getMfgToken"] = AsyncMock(return_value=[b"\xFF" * 8])
ezsp_f._xncp_features |= xncp.FirmwareFeatures.MFG_TOKEN_OVERRIDES

with caplog.at_level(logging.WARNING, logger="bellows"):
assert (
await ezsp_f.get_mfg_token(t.EzspMfgTokenId.MFG_CUSTOM_EUI_64)
) == b"\xFF" * 8

# An unsupported override is an expected condition, not a warning
assert [
r.getMessage()
for r in caplog.records
if r.name.startswith("bellows.") and r.levelno >= logging.WARNING
] == []


async def test_xncp_failure_multiprotocol(ezsp_f: EZSP) -> None:
"""Test XNCP failure with multiprotocol firmware."""
ezsp_f._mock_commands["customFrame"] = customFrame = AsyncMock(
Expand Down
Loading