separate permit_with_key and permit_with_link_key methods

tests/test_api.py

@@ -416,6 +416,26 @@ def test_handle_tx_status_duplicate(api):
     assert send_fut.set_exception.call_count == 0
 
 
+def test_handle_registration_status(api):
+    frame_id = 0x12
+    status = xbee_api.RegistrationStatus.SUCCESS
+    fut = asyncio.Future()
+    api._awaiting[frame_id] = (fut,)
+    api._handle_registration_status(frame_id, status)
+    assert fut.done() is True
+    assert fut.result() == xbee_api.RegistrationStatus.SUCCESS
+    assert fut.exception() is None
+
+    frame_id = 0x13
+    status = xbee_api.RegistrationStatus.KEY_TABLE_IS_FULL
+    fut = asyncio.Future()
+    api._awaiting[frame_id] = (fut,)
+    api._handle_registration_status(frame_id, status)
+    assert fut.done() is True
+    with pytest.raises(RuntimeError, match="Registration Status: KEY_TABLE_IS_FULL"):
+        fut.result()
+
+
 async def test_command_mode_at_cmd(api):
     command = "+++"
 

tests/test_application.py

@@ -417,7 +417,7 @@ async def test_permit_with_key(app):
     app._api._command = mock.AsyncMock(return_value=xbee_t.TXStatus.SUCCESS)
     app._api._at_command = mock.AsyncMock(return_value="OK")
     node = t.EUI64(b"\x01\x02\x03\x04\x05\x06\x07\x08")
-    code = "C9A7D2441A711695CD62170D3328EA2B423D"
+    code = b"\xC9\xA7\xD2\x44\x1A\x71\x16\x95\xCD\x62\x17\x0D\x33\x28\xEA\x2B\x42\x3D"
     time_s = 500
     await app.permit_with_key(node=node, code=code, time_s=time_s)
     app._api._at_command.assert_called_once_with("KT", time_s)
@@ -426,6 +426,19 @@ async def test_permit_with_key(app):
     )
 
 
+async def test_permit_with_link_key(app):
+    app._api._command = mock.AsyncMock(return_value=xbee_t.TXStatus.SUCCESS)
+    app._api._at_command = mock.AsyncMock(return_value="OK")
+    node = t.EUI64(b"\x01\x02\x03\x04\x05\x06\x07\x08")
+    link_key = b"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F"
+    time_s = 500
+    await app.permit_with_link_key(node=node, link_key=link_key, time_s=time_s)
+    app._api._at_command.assert_called_once_with("KT", time_s)
+    app._api._command.assert_called_once_with(
+        "register_joining_device", node, 0xFFFE, 0, link_key
+    )
+
+
 async def _test_request(
     app, expect_reply=True, send_success=True, send_timeout=False, **kwargs
 ):

zigpy_xbee/zigbee/application.py

@@ -319,14 +319,21 @@ async def permit_ncp(self, time_s=60):
         await self._api._at_command("NJ", time_s)
         await self._api._at_command("AC")
 
-    async def permit_with_key(self, node: EUI64, code: bytes, time_s=500, key_type=1):
+    async def permit_with_link_key(
+        self, node: EUI64, link_key: zigpy.types.KeyData, time_s: int = 500, key_type=0
+    ):
+        """Permits a new device to join with the given IEEE and link key."""
         assert 0x1E <= time_s <= 0xFFFF
         await self._api._at_command("KT", time_s)
         reserved = 0xFFFE
         # Key type:
         # 0 = Pre-configured Link Key (KY command of the joining device)
         # 1 = Install Code With CRC (I? command of the joining device)
-        await self._api.register_joining_device(node, reserved, key_type, code)
+        await self._api.register_joining_device(node, reserved, key_type, link_key)
+
+    async def permit_with_key(self, node: EUI64, code: bytes, time_s=500):
+        """Permits a new device to join with the given IEEE and Install Code."""
+        await self.permit_with_link_key(node, code, time_s, key_type=1)
 
     def handle_modem_status(self, status):
         LOGGER.info("Modem status update: %s (%s)", status.name, status.value)