feat: allow custom GATT read timeouts (#264)

Author: jonbng

src/bleak_esphome/backend/client.py

@@ -524,16 +524,18 @@ async def read_gatt_char(
         Args:
         ----
             characteristic: The BleakGATTCharacteristic to read from.
-            **kwargs: Unused
+            **kwargs:
+                timeout (float): Read timeout in seconds. Defaults to 30.0.
 
         Returns:
         -------
             (bytearray) The read data.
 
         """
         self._raise_if_not_connected()
+        timeout = kwargs.get("timeout", GATT_READ_TIMEOUT)
         return await self._client.bluetooth_gatt_read(
-            self._address_as_int, characteristic.handle, GATT_READ_TIMEOUT
+            self._address_as_int, characteristic.handle, timeout
         )
 
     @api_error_as_bleak_error
@@ -546,16 +548,18 @@ async def read_gatt_descriptor(
         Args:
         ----
             descriptor: The BleakGATTDescriptor to read from.
-            **kwargs: Unused
+            **kwargs:
+                timeout (float): Read timeout in seconds. Defaults to 30.0.
 
         Returns:
         -------
             (bytearray) The read data.
 
         """
         self._raise_if_not_connected()
+        timeout = kwargs.get("timeout", GATT_READ_TIMEOUT)
         return await self._client.bluetooth_gatt_read_descriptor(
-            self._address_as_int, descriptor.handle, GATT_READ_TIMEOUT
+            self._address_as_int, descriptor.handle, timeout
         )
 
     @api_error_as_bleak_error

tests/backend/test_client.py

@@ -333,6 +333,103 @@ async def test_client_get_services_and_read_write(
     mock_read.assert_called_once_with(225106397622015, 20, 30)
 
 
+@pytest.mark.asyncio
+async def test_client_read_gatt_char_with_custom_timeout(
+    client_data: ESPHomeClientData,
+    esphome_bluetooth_gatt_services: ESPHomeBluetoothGATTServices,
+) -> None:
+    """Test reading a GATT char with custom timeout."""
+    ble_device = generate_ble_device(
+        "CC:BB:AA:DD:EE:FF", details={"source": ESP_MAC_ADDRESS, "address_type": 1}
+    )
+
+    client = ESPHomeClient(ble_device, client_data=client_data)
+    client._is_connected = True
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_get_services",
+        return_value=esphome_bluetooth_gatt_services,
+    ):
+        services = await client._get_services()
+
+    char = services.get_characteristic("090b7847-e12b-09a8-b04b-8e0922a9abab")
+    assert char is not None
+
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_read",
+    ) as mock_read:
+        await client.read_gatt_char(char, timeout=90.0)
+
+    mock_read.assert_called_once_with(225106397622015, 20, 90.0)
+
+
+@pytest.mark.asyncio
+async def test_client_read_gatt_descriptor_default_timeout(
+    client_data: ESPHomeClientData,
+    esphome_bluetooth_gatt_services: ESPHomeBluetoothGATTServices,
+) -> None:
+    """Test reading a GATT descriptor uses the default timeout."""
+    ble_device = generate_ble_device(
+        "CC:BB:AA:DD:EE:FF", details={"source": ESP_MAC_ADDRESS, "address_type": 1}
+    )
+
+    client = ESPHomeClient(ble_device, client_data=client_data)
+    client._is_connected = True
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_get_services",
+        return_value=esphome_bluetooth_gatt_services,
+    ):
+        services = await client._get_services()
+
+    char = services.get_characteristic("00002a05-0000-1000-8000-00805f9b34fb")
+    assert char is not None
+    descriptor = char.get_descriptor("00002902-0000-1000-8000-00805f9b34fb")
+    assert descriptor is not None
+
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_read_descriptor",
+    ) as mock_read_descriptor:
+        await client.read_gatt_descriptor(descriptor)
+
+    mock_read_descriptor.assert_called_once_with(225106397622015, 9, 30.0)
+
+
+@pytest.mark.asyncio
+async def test_client_read_gatt_descriptor_with_custom_timeout(
+    client_data: ESPHomeClientData,
+    esphome_bluetooth_gatt_services: ESPHomeBluetoothGATTServices,
+) -> None:
+    """Test reading a GATT descriptor with custom timeout."""
+    ble_device = generate_ble_device(
+        "CC:BB:AA:DD:EE:FF", details={"source": ESP_MAC_ADDRESS, "address_type": 1}
+    )
+
+    client = ESPHomeClient(ble_device, client_data=client_data)
+    client._is_connected = True
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_get_services",
+        return_value=esphome_bluetooth_gatt_services,
+    ):
+        services = await client._get_services()
+
+    char = services.get_characteristic("00002a05-0000-1000-8000-00805f9b34fb")
+    assert char is not None
+    descriptor = char.get_descriptor("00002902-0000-1000-8000-00805f9b34fb")
+    assert descriptor is not None
+
+    with patch.object(
+        client._client,
+        "bluetooth_gatt_read_descriptor",
+    ) as mock_read_descriptor:
+        await client.read_gatt_descriptor(descriptor, timeout=90.0)
+
+    mock_read_descriptor.assert_called_once_with(225106397622015, 9, 90.0)
+
+
 @pytest.mark.asyncio
 async def test_bleak_client_get_services_and_read_write(
     client_data: ESPHomeClientData,