Bump aiospinel to 1.3.0

pyproject.toml

@@ -16,7 +16,7 @@ requires-python = ">=3.12"
 dependencies = [
     "zigpy>=2.4.0",
     "aiohttp",
-    "aiospinel>=1.2.0",
+    "aiospinel>=1.3.0",
 ]
 
 [project.entry-points."zigpy.radio"]

tests/common.py

@@ -11,7 +11,16 @@
 
 from aiohttp import web
 from aiohttp.test_utils import TestServer
-import aiospinel
+from aiospinel import (
+    CommandID,
+    HDLCLiteFrame,
+    HDLCSpecial,
+    PackedUInt21,
+    PropertyID,
+    SpinelFrame,
+    SpinelHeader,
+    Status,
+)
 import pytest
 import zigpy.config
 import zigpy.types as t
@@ -407,13 +416,13 @@ class SyntheticSpinelRcp:
     def __init__(
         self,
         *,
-        get_prop_id: aiospinel.PackedUInt21 = PROP_VENDOR_ZIGGURAT,
-        set_prop_id: aiospinel.PackedUInt21 = PROP_VENDOR_ZIGGURAT,
+        get_status: Status | None = None,
+        set_status: Status | None = None,
     ) -> None:
         self.url = ""
         self.tunnel_writes: list[bytes] = []
-        self._get_prop_id = get_prop_id
-        self._set_prop_id = set_prop_id
+        self._get_status = get_status
+        self._set_status = set_status
         self._server: asyncio.Server | None = None
         self._writers: list[asyncio.StreamWriter] = []
 
@@ -440,55 +449,59 @@ async def _serve(
                 break
             buffer += data
             while True:
-                chunk, flag, rest = buffer.partition(
-                    bytes([aiospinel.HDLCSpecial.FLAG])
-                )
+                chunk, flag, rest = buffer.partition(bytes([HDLCSpecial.FLAG]))
                 if not flag:
                     buffer = bytearray(chunk)
                     break
                 buffer = bytearray(rest)
                 if chunk:
-                    self._handle(aiospinel.HDLCLiteFrame.from_bytes(chunk), writer)
+                    self._handle(HDLCLiteFrame.from_bytes(chunk), writer)
 
-    def _handle(
-        self, hdlc: aiospinel.HDLCLiteFrame, writer: asyncio.StreamWriter
-    ) -> None:
-        frame = aiospinel.SpinelFrame.from_bytes(hdlc.data)
+    def _handle(self, hdlc: HDLCLiteFrame, writer: asyncio.StreamWriter) -> None:
+        frame = SpinelFrame.from_bytes(hdlc.data)
         tid = frame.header.transaction_id
-        if frame.command_id == aiospinel.CommandID.PROP_VALUE_GET:
-            self._respond(writer, tid, self._get_prop_id.serialize())
-        elif frame.command_id == aiospinel.CommandID.PROP_VALUE_SET:
-            _, rest = aiospinel.PackedUInt21.deserialize(frame.data)
+        if frame.command_id == CommandID.PROP_VALUE_GET:
+            self._respond_prop(writer, tid, self._get_status)
+        elif frame.command_id == CommandID.PROP_VALUE_SET:
+            _, rest = PackedUInt21.deserialize(frame.data)
             length = int.from_bytes(rest[:2], "little")
             self.tunnel_writes.append(rest[2 : 2 + length])
-            self._respond(writer, tid, self._set_prop_id.serialize())
+            self._respond_prop(writer, tid, self._set_status)
+
+    def _respond_prop(
+        self,
+        writer: asyncio.StreamWriter,
+        tid: int | None,
+        status: Status | None,
+    ) -> None:
+        if status is None:
+            data = PackedUInt21(PROP_VENDOR_ZIGGURAT).serialize()
+        else:
+            data = PropertyID.LAST_STATUS.serialize() + status.serialize()
+        self._respond(writer, tid, data)
 
     def _respond(
         self, writer: asyncio.StreamWriter, tid: int | None, data: bytes
     ) -> None:
-        frame = aiospinel.SpinelFrame(
-            header=aiospinel.SpinelHeader(
-                flag=0b10, network_link_id=0, transaction_id=tid
-            ),
-            command_id=aiospinel.CommandID.PROP_VALUE_IS,
+        frame = SpinelFrame(
+            header=SpinelHeader(flag=0b10, network_link_id=0, transaction_id=tid),
+            command_id=CommandID.PROP_VALUE_IS,
             data=data,
         )
-        writer.write(aiospinel.HDLCLiteFrame(data=frame.serialize()).serialize())
+        writer.write(HDLCLiteFrame(data=frame.serialize()).serialize())
 
     async def push_stream_frame(self, payload: bytes) -> None:
         data = (
-            PROP_VENDOR_ZIGGURAT.serialize()
+            PackedUInt21(PROP_VENDOR_ZIGGURAT).serialize()
             + len(payload).to_bytes(2, "little")
             + payload
         )
-        frame = aiospinel.SpinelFrame(
-            header=aiospinel.SpinelHeader(
-                flag=0b10, network_link_id=0, transaction_id=0
-            ),
-            command_id=aiospinel.CommandID.PROP_VALUE_IS,
+        frame = SpinelFrame(
+            header=SpinelHeader(flag=0b10, network_link_id=0, transaction_id=0),
+            command_id=CommandID.PROP_VALUE_IS,
             data=data,
         )
-        encoded = aiospinel.HDLCLiteFrame(data=frame.serialize()).serialize()
+        encoded = HDLCLiteFrame(data=frame.serialize()).serialize()
         for writer in self._writers:
             writer.write(encoded)
             await writer.drain()

tests/test_transport.py

@@ -3,7 +3,7 @@
 
 import asyncio
 
-import aiospinel
+from aiospinel import PropertyStatusError, Status
 import pytest
 
 from tests.common import (
@@ -97,7 +97,7 @@ def boom(frame: bytes) -> None:
 
 
 async def test_spinel_connect_rejects_foreign_firmware() -> None:
-    rcp = SyntheticSpinelRcp(get_prop_id=aiospinel.PackedUInt21(0x0001))
+    rcp = SyntheticSpinelRcp(get_status=Status.PROP_NOT_FOUND)
     await rcp.start()
     try:
         with pytest.raises(ConnectionError, match="does not embed"):
@@ -107,11 +107,11 @@ async def test_spinel_connect_rejects_foreign_firmware() -> None:
 
 
 async def test_spinel_tunnel_write_rejected() -> None:
-    rcp = SyntheticSpinelRcp(set_prop_id=aiospinel.PackedUInt21(0x0001))
+    rcp = SyntheticSpinelRcp(set_status=Status.INVALID_ARGUMENT)
     await rcp.start()
     transport = await connect_transport(rcp.url, lambda frame: None, lambda exc: None)
     try:
-        with pytest.raises(ConnectionError, match="Tunnel write rejected"):
+        with pytest.raises(PropertyStatusError):
             await transport.send_frame(b"\x01")
     finally:
         await transport.disconnect()

zigpy_ziggurat/zigbee/transport.py

@@ -8,14 +8,14 @@
 from typing import Any, Protocol, cast
 
 import aiohttp
-import aiospinel
+from aiospinel import PropertyStatusError, SpinelProtocol
 import zigpy.serial
 
 _LOGGER = logging.getLogger(__name__)
 
 WEBSOCKET_HEARTBEAT = 15
 
-PROP_VENDOR_ZIGGURAT = aiospinel.PackedUInt21(0x3D5A)
+PROP_VENDOR_ZIGGURAT = 0x3D5A
 
 # The callback the API installs to receive a device -> host binary frame.
 OnFrame = Callable[[bytes], None]
@@ -56,7 +56,7 @@ async def connect_transport(
 # -- serial (Spinel tunnel) ------------------------------------------------------
 
 
-class _SpinelProtocol(aiospinel.SpinelProtocol):
+class _SpinelProtocol(SpinelProtocol):
     """Tunnels binary frames over the vendor Spinel stream property."""
 
     def __init__(self, on_frame: OnFrame, on_lost: OnLost) -> None:
@@ -80,32 +80,18 @@ def _stream_frame_received(self, data: bytes) -> None:
             _LOGGER.exception("Failed to handle frame: %r", data)
 
     async def start_ziggurat(self) -> None:
-        rsp = await self.send_command(
-            aiospinel.CommandID.PROP_VALUE_GET,
-            PROP_VENDOR_ZIGGURAT.serialize(),
-        )
-        prop_id, _ = aiospinel.PackedUInt21.deserialize(rsp.data)
-        if prop_id != PROP_VENDOR_ZIGGURAT:
-            raise ConnectionError(
-                f"Firmware does not embed the Ziggurat stack: {rsp!r}"
-            )
+        try:
+            await self.get_property(PROP_VENDOR_ZIGGURAT)
+        except PropertyStatusError as exc:
+            raise ConnectionError("Firmware does not embed the Ziggurat stack") from exc
         _LOGGER.debug("Embedded Ziggurat firmware detected")
 
     async def tunnel_send(self, frame: bytes) -> None:
         # No retries: a timed-out tunnel write must not resend the request (the first
         # copy may already have been processed).
-        rsp = await self.send_command(
-            aiospinel.CommandID.PROP_VALUE_SET,
-            (
-                PROP_VENDOR_ZIGGURAT.serialize()
-                + len(frame).to_bytes(2, "little")
-                + frame
-            ),
-            retries=0,
+        await self.set_property(
+            PROP_VENDOR_ZIGGURAT, len(frame).to_bytes(2, "little") + frame, retries=0
         )
-        prop_id, _ = aiospinel.PackedUInt21.deserialize(rsp.data)
-        if prop_id != PROP_VENDOR_ZIGGURAT:
-            raise ConnectionError(f"Tunnel write rejected: {rsp!r}")
 
 
 class SpinelTransport:
@@ -141,8 +127,7 @@ async def connect(self) -> None:
 
     async def disconnect(self) -> None:
         if self._protocol is not None:
-            self._protocol.close()
-            await self._protocol.wait_until_closed()
+            await self._protocol.disconnect()
             self._protocol = None
 
     async def send_frame(self, frame: bytes) -> None: