Add test for GATT error while doing a read (#736)

This commit is contained in:
J. Nick Koston 2023-11-26 13:59:07 -06:00 committed by GitHub
parent 1a7ce50f63
commit 67bd7efb29
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 4 deletions

View File

@ -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]

View File

@ -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[