diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 9fe94ac5..4a90d4cc 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -3,7 +3,7 @@ import asyncio from asyncio import timeout as asyncio_timeout from collections.abc import AsyncGenerator -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta import importlib.metadata import logging import os @@ -123,6 +123,10 @@ def __init__(self, config: dict) -> None: self._req_lock = asyncio.Lock() self._packet_capture_channel: int | None = None + # The NCP queues a single GPDF for all GPDs + self._gpdf_lock = asyncio.Lock() + self._gpdf_sent: asyncio.Future[None] | None = None + @property def controller_event(self): """Return asyncio.Event for controller app.""" @@ -729,6 +733,11 @@ def ezsp_callback_handler(self, frame_name, args): proxyTableIndex, gpdCommandPayload, ) + elif frame_name == "dGpSentHandler": + # The NCP reports neither the transmit status nor the GPEP handle: both are + # always zero. With one GPDF queued at a time, this is the queued one. + if self._gpdf_sent is not None and not self._gpdf_sent.done(): + self._gpdf_sent.set_result(None) elif frame_name == "messageSentHandler": if self._ezsp.ezsp_version >= 14: ( @@ -1383,6 +1392,89 @@ async def _unsubscribe_from_multicast_group( await self._multicast.unsubscribe(group_id=group_id, endpoint_id=endpoint_id) + async def _send_gpdf( + self, + *, + target: zgp_t.SrcID | tuple[zigpy.types.EUI64, int] | None, + command_id: zgp_t.GPDCommandID, + payload: bytes, + channel: int, + lifetime: timedelta, + ) -> None: + """Send a GPDF from the NCP in its GPD's next receive window.""" + # The NCP only transmits on the operational channel, outside of a GP channel + # delivery scan + if channel != self.state.network_info.channel: + raise zigpy.exceptions.DeliveryError( + f"Cannot send a GPDF on channel {channel}" + ) + + # A Maintenance frame is queued for SrcID 0 (A.1.5.2.2) + match target: + case None: + addr = t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.SrcID, + source_id=zgp_t.SrcID(0x00000000), + source_id_padding=0, + endpoint=0, + ) + case (ieee, endpoint): + addr = t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.IEEE, + gpd_eui64=ieee, + endpoint=endpoint, + ) + case _: + addr = t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.SrcID, + source_id=target, + source_id_padding=0, + endpoint=0, + ) + + # The NCP never expires a queued GPDF, which would block every other GPD, so the + # GPDF is removed once `lifetime` is up + async with self._gpdf_lock: + self._gpdf_sent = asyncio.get_running_loop().create_future() + + try: + (status,) = await self._ezsp.dGpSend( + action=True, + useCca=False, + addr=addr, + gpdCommandId=command_id, + gpdAsdu=payload, + gpepHandle=0, + gpTxQueueEntryLifetimeMs=lifetime // timedelta(milliseconds=1), + ) + + if t.sl_Status.from_ember_status(status) != t.sl_Status.OK: + raise zigpy.exceptions.DeliveryError( + f"Failed to queue GPDF: {status!r}" + ) + + try: + async with asyncio_timeout(lifetime.total_seconds()): + await self._gpdf_sent + except TimeoutError: + (status,) = await self._ezsp.dGpSend( + action=False, + useCca=False, + addr=addr, + gpdCommandId=command_id, + gpdAsdu=b"", + gpepHandle=0, + gpTxQueueEntryLifetimeMs=0, + ) + + # Not finding it means the GPD took it meanwhile + if t.sl_Status.from_ember_status(status) == t.sl_Status.OK: + raise zigpy.exceptions.DeliveryError( + "GPD did not listen before the GPDF expired" + ) from None + finally: + self._gpdf_sent = None + def _handle_id_conflict(self, nwk: t.EmberNodeId) -> None: LOGGER.warning("NWK conflict is reported for 0x%04x", nwk) self.state.counters[COUNTERS_CTRL][COUNTER_NWK_CONFLICTS].increment() diff --git a/tests/test_application.py b/tests/test_application.py index 326ef973..5133d973 100644 --- a/tests/test_application.py +++ b/tests/test_application.py @@ -1,5 +1,6 @@ import asyncio import contextlib +from datetime import timedelta import importlib.metadata import logging from unittest.mock import AsyncMock, MagicMock, PropertyMock, call, patch, sentinel @@ -3261,3 +3262,151 @@ def test_frame_handler_gp_security_failure(app: ControllerApplication) -> None: app.ezsp_callback_handler("gpepIncomingMessageHandler", list(args.values())) assert len(gp_packet_received.mock_calls) == 0 + + +GPD_IEEE = zigpy_t.EUI64.convert("00:11:22:33:44:55:66:77") + + +async def send_gpdf( + app: ControllerApplication, + target: zgp_t.SrcID | tuple[zigpy_t.EUI64, int] | None = zgp_t.SrcID(0x12345678), + *, + channel: int = 25, + lifetime: timedelta = timedelta(seconds=20), +) -> None: + await app.send_gpdf( + target=target, + command_id=zgp_t.GPDCommandID(0xFE), + payload=b"\x00", + channel=channel, + lifetime=lifetime, + ) + + +@pytest.mark.parametrize( + ("target", "addr"), + [ + ( + zgp_t.SrcID(0x12345678), + t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.SrcID, + source_id=zgp_t.SrcID(0x12345678), + source_id_padding=0, + endpoint=0, + ), + ), + ( + (GPD_IEEE, 0x01), + t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.IEEE, + gpd_eui64=GPD_IEEE, + endpoint=0x01, + ), + ), + # A Maintenance frame is queued for SrcID 0 + ( + None, + t.EmberGpAddress( + applicationId=zgp_t.ApplicationID.SrcID, + source_id=zgp_t.SrcID(0x00000000), + source_id_padding=0, + endpoint=0, + ), + ), + ], +) +async def test_send_gpdf( + app: ControllerApplication, + target: zgp_t.SrcID | tuple[zigpy_t.EUI64, int] | None, + addr: t.EmberGpAddress, +) -> None: + """A GPDF is queued in the NCP and sent once the NCP reports it.""" + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS]) + + task = asyncio.create_task(send_gpdf(app, target)) + await asyncio.sleep(0.01) + assert not task.done() + + # The NCP reports neither a status nor the handle + app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0]) + await task + + assert app._ezsp.dGpSend.mock_calls == [ + call( + action=True, + useCca=False, + addr=addr, + gpdCommandId=zgp_t.GPDCommandID(0xFE), + gpdAsdu=b"\x00", + gpepHandle=0, + gpTxQueueEntryLifetimeMs=20_000, + ) + ] + + +async def test_send_gpdf_one_at_a_time(app: ControllerApplication) -> None: + """The NCP queues one GPDF for all GPDs, so a second one waits for the first.""" + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS]) + + first = asyncio.create_task(send_gpdf(app, zgp_t.SrcID(0x00000001))) + second = asyncio.create_task(send_gpdf(app, zgp_t.SrcID(0x00000002))) + await asyncio.sleep(0.01) + assert len(app._ezsp.dGpSend.mock_calls) == 1 + + app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0]) + await first + await asyncio.sleep(0.01) + assert len(app._ezsp.dGpSend.mock_calls) == 2 + assert not second.done() + + app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0]) + await second + + +async def test_send_gpdf_expired(app: ControllerApplication) -> None: + """The NCP never expires a GPDF, so it is removed once its lifetime is up.""" + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.SUCCESS]) + + with pytest.raises(zigpy.exceptions.DeliveryError, match="expired"): + await send_gpdf(app, lifetime=timedelta(milliseconds=10)) + + [_, removal] = app._ezsp.dGpSend.mock_calls + assert removal.kwargs["action"] is False + assert removal.kwargs["addr"].source_id == 0x12345678 + + +async def test_send_gpdf_sent_while_expiring(app: ControllerApplication) -> None: + """A GPDF the NCP no longer holds when it expires was taken by its GPD.""" + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock( + side_effect=[[t.EmberStatus.SUCCESS], [t.EmberStatus.ERR_FATAL]] + ) + + await send_gpdf(app, lifetime=timedelta(milliseconds=10)) + + +async def test_send_gpdf_not_queued(app: ControllerApplication) -> None: + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock(return_value=[t.EmberStatus.ERR_FATAL]) + + with pytest.raises(zigpy.exceptions.DeliveryError, match="Failed to queue"): + await send_gpdf(app) + + +async def test_send_gpdf_other_channel(app: ControllerApplication) -> None: + """The NCP cannot send a GPDF off the operational channel.""" + app.state.network_info.channel = 25 + app._ezsp.dGpSend = AsyncMock() + + with pytest.raises(zigpy.exceptions.DeliveryError, match="channel 15"): + await send_gpdf(app, channel=15) + + assert len(app._ezsp.dGpSend.mock_calls) == 0 + + +def test_gpdf_sent_without_pending(app: ControllerApplication) -> None: + """A GPDF sent after it was given up on is ignored.""" + app.ezsp_callback_handler("dGpSentHandler", [t.EmberStatus.SUCCESS, 0])