fix: manage BrokenPipeError thrown by MessageBus.connect() (#9)

Bluetooth-Devices · Aug 27, 2022 · 5d0fbaa · 5d0fbaa
1 parent 03db71e
commit 5d0fbaa
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 1 deletion.
diff --git a/src/bluetooth_adapters/__init__.py b/src/bluetooth_adapters/__init__.py
@@ -46,7 +46,7 @@ async def _get_dbus_managed_objects() -> dict[str, Any]:
         bus = await MessageBus(
             bus_type=BusType.SYSTEM, negotiate_unix_fd=True
         ).connect()
-    except FileNotFoundError as ex:
+    except (FileNotFoundError, BrokenPipeError) as ex:
         _LOGGER.debug("Dbus not available: %s", ex)
         return {}
     msg = Message(

diff --git a/tests/test_init.py b/tests/test_init.py
@@ -37,6 +37,22 @@ async def call(self):
         assert await get_bluetooth_adapters() == []
 
 
+@pytest.mark.asyncio
+async def test_get_bluetooth_adapters_connect_broken_pipe():
+    class MockMessageBus:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def connect(self):
+            raise BrokenPipeError
+
+        async def call(self):
+            return None
+
+    with patch("bluetooth_adapters.MessageBus", MockMessageBus):
+        assert await get_bluetooth_adapters() == []
+
+
 @pytest.mark.asyncio
 async def test_get_bluetooth_adapters_no_call_return():
     class MockMessageBus: