Skip to content

Commit

Permalink
fix: handle AuthError while probing for adapters
Browse files Browse the repository at this point in the history
  • Loading branch information
bdraco committed May 4, 2024
1 parent fd9595c commit e3c82a5
Show file tree
Hide file tree
Showing 3 changed files with 32 additions and 2 deletions.
5 changes: 5 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,11 @@ build_command = "pip install poetry && poetry build"
[tool.pytest.ini_options]
addopts = "-v -Wdefault --cov=bluetooth_adapters --cov-report=term-missing:skip-covered"
pythonpath = ["src"]
log_format = "%(asctime)s.%(msecs)03d %(levelname)-8s %(threadName)s %(name)s:%(filename)s:%(lineno)s %(message)s"
log_date_format = "%Y-%m-%d %H:%M:%S"
asyncio_mode = "auto"
log_cli = "true"
log_level = "NOTSET"

[tool.coverage.run]
branch = true
Expand Down
10 changes: 9 additions & 1 deletion src/bluetooth_adapters/dbus.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,10 +7,11 @@
from typing import Any

try:
from dbus_fast import BusType, Message, MessageType, unpack_variants
from dbus_fast import AuthError, BusType, Message, MessageType, unpack_variants
from dbus_fast.aio import MessageBus
except (AttributeError, ImportError):
# dbus_fast is not available on Windows
AuthError = None
BusType = None
Message = None
MessageType = None
Expand Down Expand Up @@ -101,6 +102,13 @@ async def get_dbus_managed_objects() -> dict[str, Any]:
async def _get_dbus_managed_objects() -> dict[str, Any]:
try:
bus = await MessageBus(bus_type=BusType.SYSTEM).connect()
except AuthError as ex:
_LOGGER.debug(
"DBus authentication error; make sure the DBus socket "
"is available and the user has the correct permissions: %s",
ex,
)
return {}
except FileNotFoundError as ex:
if is_docker_env():
_LOGGER.debug(
Expand Down
19 changes: 18 additions & 1 deletion tests/test_init.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,9 +9,10 @@
from bleak.backends.scanner import AdvertisementData

try:
from dbus_fast import MessageType
from dbus_fast import AuthError, MessageType
except (AttributeError, ImportError):
MessageType = None
AuthError = None
# dbus_fast is not available on Windows
from uart_devices import BluetoothDevice as UARTBluetoothDevice
from uart_devices import UARTDevice
Expand Down Expand Up @@ -80,6 +81,22 @@ def __init__(self, *args, **kwargs):
assert await get_bluetooth_adapters() == []


@pytest.mark.asyncio
@pytest.mark.skipif(
MessageType is None or get_dbus_managed_objects is None,
reason="dbus_fast is not available",
)
async def test_get_bluetooth_adapters_auth_eror():
"""Test get_bluetooth_adapters with auth error."""

class MockMessageBus:
def __init__(self, *args, **kwargs):
raise AuthError

with patch("bluetooth_adapters.dbus.MessageBus", MockMessageBus):
assert await get_bluetooth_adapters() == []


@pytest.mark.asyncio
@pytest.mark.skipif(
MessageType is None or get_dbus_managed_objects is None,
Expand Down

0 comments on commit e3c82a5

Please sign in to comment.