fix: add a timeout in case dbus fails to respond (#6)

pyproject.toml

@@ -30,6 +30,7 @@ Sphinx = {version = "^5.0", optional = true}
 sphinx-rtd-theme = {version = "^1.0", optional = true}
 myst-parser = {version = "^0.18", optional = true}
 dbus-next = ">=0.2.3"
+async-timeout = ">=4.0.2"
 
 [tool.poetry.extras]
 docs = [

src/bluetooth_adapters/__init__.py

@@ -1,14 +1,18 @@
 __version__ = "0.1.2"
 
+import asyncio
 import logging
 
+import async_timeout
 from dbus_next import BusType, Message, MessageType
 from dbus_next.aio import MessageBus
 
 __all__ = ["get_bluetooth_adapters"]
 
 _LOGGER = logging.getLogger(__name__)
 
+REPLY_TIMEOUT = 8
+
 
 async def get_bluetooth_adapters() -> list[str]:
     """Return a list of bluetooth adapters."""
@@ -26,7 +30,12 @@ async def get_bluetooth_adapters() -> list[str]:
         interface="org.freedesktop.DBus.ObjectManager",
         member="GetManagedObjects",
     )
-    reply = await bus.call(msg)
+    try:
+        async with async_timeout.timeout(REPLY_TIMEOUT):
+            reply = await bus.call(msg)
+    except asyncio.TimeoutError:
+        _LOGGER.debug("D-bus timeout waiting for reply to GetManagedObjects")
+        return adapters
     bus.disconnect()
     if not reply or reply.message_type != MessageType.METHOD_RETURN:
         _LOGGER.debug("Unexpected replay: %s", reply)

tests/test_init.py

@@ -1,8 +1,11 @@
+import asyncio
+from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from dbus_next import MessageType
 
+import bluetooth_adapters
 from bluetooth_adapters import get_bluetooth_adapters
 
 
@@ -50,6 +53,27 @@ async def call(self):
         assert await get_bluetooth_adapters() == []
 
 
+@pytest.mark.asyncio
+async def test_get_bluetooth_adapters_times_out():
+    async def _stall(*args: Any) -> None:
+        await asyncio.sleep(10)
+
+    class MockMessageBus:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def connect(self):
+            return AsyncMock(
+                disconnect=MagicMock(),
+                call=AsyncMock(side_effect=_stall),
+            )
+
+    with patch.object(bluetooth_adapters, "REPLY_TIMEOUT", 0), patch(
+        "bluetooth_adapters.MessageBus", MockMessageBus
+    ):
+        assert await get_bluetooth_adapters() == []
+
+
 @pytest.mark.asyncio
 async def test_get_bluetooth_adapters_no_wrong_return():
     class MockMessageBus: