From 67bd7efb29a65972bc2c2b06fe373da8a79e00a9 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sun, 26 Nov 2023 13:59:07 -0600 Subject: [PATCH] Add test for GATT error while doing a read (#736) --- aioesphomeapi/client.py | 11 +++++++---- tests/test_client.py | 18 ++++++++++++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/aioesphomeapi/client.py b/aioesphomeapi/client.py index e3778bf..c652c01 100644 --- a/aioesphomeapi/client.py +++ b/aioesphomeapi/client.py @@ -467,7 +467,7 @@ class APIClient: timeout: float = 10.0, ) -> message.Message: message_filter = partial(self._filter_bluetooth_message, address, handle) - resp = await self._get_connection().send_messages_await_response_complex( + [resp] = await self._get_connection().send_messages_await_response_complex( (request,), message_filter, message_filter, @@ -475,10 +475,13 @@ class APIClient: timeout, ) - if isinstance(resp[0], BluetoothGATTErrorResponse): - raise BluetoothGATTAPIError(BluetoothGATTError.from_pb(resp[0])) + if ( + type(resp) # pylint: disable=unidiomatic-typecheck + is BluetoothGATTErrorResponse + ): + raise BluetoothGATTAPIError(BluetoothGATTError.from_pb(resp)) - return resp[0] + return resp async def subscribe_bluetooth_le_advertisements( self, on_bluetooth_le_advertisement: Callable[[BluetoothLEAdvertisement], None] diff --git a/tests/test_client.py b/tests/test_client.py index 1b87da4..6c0d80f 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -980,6 +980,24 @@ async def test_bluetooth_gatt_read( assert await read_task == b"1234" +@pytest.mark.asyncio +async def test_bluetooth_gatt_read_error( + api_client: tuple[ + APIClient, APIConnection, asyncio.Transport, APIPlaintextFrameHelper + ], +) -> None: + """Test bluetooth_gatt_read that errors.""" + client, connection, transport, protocol = api_client + read_task = asyncio.create_task(client.bluetooth_gatt_read(1234, 1234)) + await asyncio.sleep(0) + error_response: message.Message = BluetoothGATTErrorResponse( + address=1234, handle=1234 + ) + mock_data_received(protocol, generate_plaintext_packet(error_response)) + with pytest.raises(BluetoothGATTAPIError): + await read_task + + @pytest.mark.asyncio async def test_bluetooth_gatt_read_descriptor( api_client: tuple[