feat: add function to clear the cache (#75)

src/bleak_retry_connector/__init__.py

@@ -172,6 +172,13 @@ def set_cached_services(self, services: BleakGATTServiceCollection | None) -> No
         This was only kept for backwards compatibility.
         """
 
+    async def clear_cache(self) -> bool:
+        """Clear the cached services."""
+        if hasattr(super(), "clear_cache"):
+            return await super().clear_cache()
+        _LOGGER.warning("clear_cache not implemented in bleak version")
+        return False
+
 
 def ble_device_has_changed(original: BLEDevice, new: BLEDevice) -> bool:
     """Check if the device has changed."""
@@ -187,6 +194,19 @@ def ble_device_has_changed(original: BLEDevice, new: BLEDevice) -> bool:
     )
 
 
+async def clear_cache(address: str) -> bool:
+    """Clear the cache for a device."""
+    if not IS_LINUX or not await get_device(address):
+        return False
+    with contextlib.suppress(Exception):
+        manager = await get_global_bluez_manager()
+        manager._services_cache.pop(
+            address.upper(), None
+        )  # pylint: disable=protected-access
+        return True
+    return False
+
+
 def ble_device_description(device: BLEDevice) -> str:
     """Get the device description."""
     details = device.details

tests/test_init.py

@@ -27,6 +27,7 @@
     ble_device_description,
     ble_device_has_changed,
     calculate_backoff_time,
+    clear_cache,
     establish_connection,
     get_connected_devices,
     get_device,
@@ -996,6 +997,75 @@ def __init__(self):
     assert device.details["path"] == "/org/bluez/hci0/dev_FA_23_9D_AA_45_46"
 
 
+@pytest.mark.asyncio
+async def test_clear_cache():
+    class FakeBluezManager:
+        def __init__(self):
+            self._services_cache = {"FA:23:9D:AA:45:46": "test"}
+            self._properties = {
+                "/org/bluez/hci0/dev_FA_23_9D_AA_45_46": {
+                    "UUID": "service",
+                    "Primary": True,
+                    "Characteristics": [],
+                    defs.DEVICE_INTERFACE: {
+                        "Address": "FA:23:9D:AA:45:46",
+                        "Alias": "FA:23:9D:AA:45:46",
+                        "RSSI": -30,
+                    },
+                    defs.GATT_SERVICE_INTERFACE: True,
+                },
+                "/org/bluez/hci1/dev_FA_23_9D_AA_45_46": {
+                    "UUID": "service",
+                    "Primary": True,
+                    "Characteristics": [],
+                    defs.DEVICE_INTERFACE: {
+                        "Address": "FA:23:9D:AA:45:46",
+                        "Alias": "FA:23:9D:AA:45:46",
+                        "RSSI": -79,
+                    },
+                    defs.GATT_SERVICE_INTERFACE: True,
+                },
+                "/org/bluez/hci2/dev_FA_23_9D_AA_45_46": {
+                    "UUID": "service",
+                    "Primary": True,
+                    "Characteristics": [],
+                    defs.DEVICE_INTERFACE: {
+                        "Address": "FA:23:9D:AA:45:46",
+                        "Alias": "FA:23:9D:AA:45:46",
+                        "RSSI": -80,
+                    },
+                    defs.GATT_SERVICE_INTERFACE: True,
+                },
+                "/org/bluez/hci3/dev_FA_23_9D_AA_45_46": {
+                    "UUID": "service",
+                    "Primary": True,
+                    "Characteristics": [],
+                    defs.DEVICE_INTERFACE: {
+                        "Address": "FA:23:9D:AA:45:46",
+                        "Alias": "FA:23:9D:AA:45:46",
+                        "RSSI": -31,
+                    },
+                    defs.GATT_SERVICE_INTERFACE: True,
+                },
+            }
+
+    bluez_manager = FakeBluezManager()
+
+    bleak_retry_connector.get_global_bluez_manager = AsyncMock(
+        return_value=bluez_manager
+    )
+    bleak_retry_connector.defs = defs
+
+    with patch.object(bleak_retry_connector, "IS_LINUX", True):
+        device = await get_device("FA:23:9D:AA:45:46")
+
+        assert device is not None
+        assert device.details["path"] == "/org/bluez/hci0/dev_FA_23_9D_AA_45_46"
+
+        assert await clear_cache("FA:23:9D:AA:45:46")
+        assert bluez_manager._services_cache == {}
+
+
 @pytest.mark.asyncio
 async def test_get_device_mac_os():
     class FakeBluezManager: