diff --git a/pyproject.toml b/pyproject.toml index 9283af2..15e9633 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/common.py b/tests/common.py index e1dead4..6e298a6 100644 --- a/tests/common.py +++ b/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() diff --git a/tests/test_transport.py b/tests/test_transport.py index 8f0c883..572d0b4 100644 --- a/tests/test_transport.py +++ b/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() diff --git a/uv.lock b/uv.lock index d99fc63..2d44e5a 100644 --- a/uv.lock +++ b/uv.lock @@ -130,14 +130,14 @@ wheels = [ [[package]] name = "aiospinel" -version = "1.2.0" +version = "1.3.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/c7/96/3ee10d7cc34f6b764b617714f9777a16a0b13ce4b63d832f06c93534432c/aiospinel-1.2.0.tar.gz", hash = "sha256:b71a771583a767cb57ff4f6cfa560f1ec22eef1496c790436daf80e808442075", size = 59890, upload-time = "2026-07-09T20:45:25.501Z" } +sdist = { url = "https://files.pythonhosted.org/packages/42/23/d2c838f4c46188b91208d1f7d2c71900d5ed58eb01b5a12f66ff3ae519a3/aiospinel-1.3.0.tar.gz", hash = "sha256:152d737c4d47ffed1195a832045b2c99391dfc04a3ce1e8120b48ad616828194", size = 63129, upload-time = "2026-11-01T20:41:03.185Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b8/9e/239584868623787832574d4530ee3a8e7af4b61ef42370934e6e034aeae2/aiospinel-1.2.0-py3-none-any.whl", hash = "sha256:84824074101bff1ea7186fc541fdd6d39c1a8cda354252f0b80289944f5ef5ca", size = 14424, upload-time = "2026-07-09T20:45:24.155Z" }, + { url = "https://files.pythonhosted.org/packages/c6/2d/1e82f80f59fe34ec6e75beb6036950485e1240c17d5833f4c50aeb47f1ce/aiospinel-1.3.0-py3-none-any.whl", hash = "sha256:2ba4ed566df2d1b9dd5edb75e37d7ffd40844ec56f18a48f447b57376d3195c8", size = 14300, upload-time = "2026-11-01T20:41:01.84Z" }, ] [[package]] @@ -1528,7 +1528,7 @@ testing = [ [package.metadata] requires-dist = [ { name = "aiohttp" }, - { name = "aiospinel", specifier = ">=1.2.0" }, + { name = "aiospinel", specifier = ">=1.3.0" }, { name = "zigpy", specifier = ">=2.4.0" }, ] diff --git a/zigpy_ziggurat/zigbee/transport.py b/zigpy_ziggurat/zigbee/transport.py index 4c7c59d..5145f8d 100644 --- a/zigpy_ziggurat/zigbee/transport.py +++ b/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: