API to subscribe to multicast groups

bellows/multicast.py

@@ -80,7 +80,7 @@ async def unsubscribe(self, group_id) -> t.sl_Status:
         try:
             entry, idx = self._multicast[group_id]
         except KeyError:
-            LOGGER.error(
+            LOGGER.debug(
                 "Couldn't find MulticastTableEntry for %s multicast_id", group_id
             )
             return t.sl_Status.INVALID_INDEX

bellows/zigbee/application.py

@@ -1131,6 +1131,20 @@ async def permit_with_link_key(
 
         return await super().permit(time_s)
 
+    async def _subscribe_to_multicast_group(self, group_id: t.Group) -> None:
+        """Ask the coordinator firmware to subscribe to a group, if needed."""
+        if self._multicast is None:
+            return None
+
+        await self._multicast.subscribe(group_id)
+
+    async def _unsubscribe_from_multicast_group(self, group_id: t.Group) -> None:
+        """Ask the coordinator firmware to unsubscribe from a group, if needed."""
+        if self._multicast is None:
+            return None
+
+        await self._multicast.unsubscribe(group_id)
+
     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()

pyproject.toml

@@ -17,7 +17,7 @@ dependencies = [
     "click",
     "click-log>=0.2.1",
     "voluptuous",
-    "zigpy>=0.87.0",
+    "zigpy>=2.1.0",
 ]
 
 [tool.setuptools.packages.find]

tests/test_application.py

@@ -24,6 +24,7 @@
     GetRouteTableEntryRsp,
     GetTxPowerInfoRsp,
 )
+from bellows.multicast import Multicast
 import bellows.types
 import bellows.types as t
 import bellows.types.struct
@@ -2683,3 +2684,42 @@ async def test_set_tx_power(app: ControllerApplication) -> None:
     assert result == 12.0
     assert app._ezsp.setRadioPower.mock_calls == [call(power=12)]
     assert mock_update.mock_calls == [call(app._ezsp, tx_power=12)]
+
+
+async def test_multicast_group_subscription(app: ControllerApplication) -> None:
+    """Test multicast group subscription APIs when there are no XNCP extensions."""
+    app._ezsp._xncp_features = FirmwareFeatures.NONE
+
+    app._multicast = Multicast(app._ezsp)
+    await app._multicast._initialize()
+
+    # Subscribe to a group
+    await app.subscribe_to_multicast_group(0x1234)
+    assert app._ezsp._protocol.setMulticastTableEntry.mock_calls == [
+        call(
+            0,
+            t.EmberMulticastTableEntry(multicastId=0x1234, endpoint=1, networkIndex=0),
+        )
+    ]
+
+    app._ezsp._protocol.setMulticastTableEntry.reset_mock()
+
+    # Unsubscribe from a group
+    await app.unsubscribe_from_multicast_group(0x1234)
+    assert app._ezsp._protocol.setMulticastTableEntry.mock_calls == [
+        call(
+            0,
+            t.EmberMulticastTableEntry(multicastId=0x1234, endpoint=0, networkIndex=0),
+        )
+    ]
+
+
+async def test_multicast_group_subscription_xncp(app: ControllerApplication) -> None:
+    """Test multicast group subscription APIs when XNCP extensions are available."""
+    app._ezsp._xncp_features |= FirmwareFeatures.MEMBER_OF_ALL_GROUPS
+
+    # Subscribe to a group (no-op)
+    await app.subscribe_to_multicast_group(0x1234)
+
+    # Unsubscribe from a group (no-op)
+    await app.unsubscribe_from_multicast_group(0x1234)