self-hosted/home-assistant-core
self-hosted/home-assistant-core
1
0
Fork 0
mirror of https://github.com/home-assistant/core synced 2024-10-15 00:17:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Switch to a different local Bluetooth adapter when one runs out of connection slots (#84331)

Browse source
This commit is contained in:
J. Nick Koston 2022-12-23 08:58:33 -10:00 committed by GitHub
parent f39f3b612a
commit 070aa714a0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 395 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
homeassistant/components/bluetooth/__init__.py
View file

@ -7,12 +7,15 @@ import platform
from typing import TYPE_CHECKING
from awesomeversion import AwesomeVersion
from bleak_retry_connector import BleakSlotManager
from bluetooth_adapters import (
ADAPTER_ADDRESS,
ADAPTER_CONNECTION_SLOTS,
ADAPTER_HW_VERSION,
ADAPTER_MANUFACTURER,
ADAPTER_SW_VERSION,
DEFAULT_ADDRESS,
DEFAULT_CONNECTION_SLOTS,
AdapterDetails,
adapter_human_name,
adapter_model,
@ -165,8 +168,10 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
bluetooth_adapters = get_adapters()
bluetooth_storage = BluetoothStorage(hass)
await bluetooth_storage.async_setup()
slot_manager = BleakSlotManager()
await slot_manager.async_setup()
manager = BluetoothManager(
hass, integration_matcher, bluetooth_adapters, bluetooth_storage
hass, integration_matcher, bluetooth_adapters, bluetooth_storage, slot_manager
)
await manager.async_setup()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, manager.async_stop)
@ -270,7 +275,7 @@ async def async_discover_adapters(
async def async_update_device(
hass: HomeAssistant, entry: ConfigEntry, adapter: str
hass: HomeAssistant, entry: ConfigEntry, adapter: str, details: AdapterDetails
) -> None:
"""Update device registry entry.
@ -279,11 +284,7 @@ async def async_update_device(
update the device with the new location so they can
figure out where the adapter is.
"""
manager: BluetoothManager = hass.data[DATA_MANAGER]
adapters = await manager.async_get_bluetooth_adapters()
details = adapters[adapter]
registry = dr.async_get(manager.hass)
registry.async_get_or_create(
dr.async_get(hass).async_get_or_create(
config_entry_id=entry.entry_id,
name=adapter_human_name(adapter, details[ADAPTER_ADDRESS]),
connections={(dr.CONNECTION_BLUETOOTH, details[ADAPTER_ADDRESS])},
@ -307,6 +308,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
passive = entry.options.get(CONF_PASSIVE)
mode = BluetoothScanningMode.PASSIVE if passive else BluetoothScanningMode.ACTIVE
new_info_callback = async_get_advertisement_callback(hass)
manager: BluetoothManager = hass.data[DATA_MANAGER]
scanner = HaScanner(hass, mode, adapter, address, new_info_callback)
try:
scanner.async_setup()
@ -318,8 +320,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await scanner.async_start()
except ScannerStartError as err:
raise ConfigEntryNotReady from err
entry.async_on_unload(async_register_scanner(hass, scanner, True))
await async_update_device(hass, entry, adapter)
adapters = await manager.async_get_bluetooth_adapters()
details = adapters[adapter]
slots: int = details.get(ADAPTER_CONNECTION_SLOTS) or DEFAULT_CONNECTION_SLOTS
entry.async_on_unload(async_register_scanner(hass, scanner, True, slots))
await async_update_device(hass, entry, adapter, details)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = scanner
entry.async_on_unload(entry.add_update_listener(async_update_listener))
return True

9
homeassistant/components/bluetooth/api.py
View file

@ -172,10 +172,15 @@ def async_rediscover_address(hass: HomeAssistant, address: str) -> None:
@hass_callback
def async_register_scanner(
hass: HomeAssistant, scanner: BaseHaScanner, connectable: bool
hass: HomeAssistant,
scanner: BaseHaScanner,
connectable: bool,
connection_slots: int | None = None,
) -> CALLBACK_TYPE:
"""Register a BleakScanner."""
return _get_manager(hass).async_register_scanner(scanner, connectable)
return _get_manager(hass).async_register_scanner(
scanner, connectable, connection_slots
)
@hass_callback

2
homeassistant/components/bluetooth/base_scanner.py
View file

@ -44,6 +44,7 @@ class BaseHaScanner(ABC):
__slots__ = (
"hass",
"adapter",
"connectable",
"source",
"connector",
@ -68,6 +69,7 @@ class BaseHaScanner(ABC):
self.source = source
self.connector = connector
self._connecting = 0
self.adapter = adapter
self.name = adapter_human_name(adapter, source) if adapter != source else source
self.scanning = True
self._last_detection = 0.0

24
homeassistant/components/bluetooth/manager.py
View file

@ -9,7 +9,7 @@ import logging
from typing import TYPE_CHECKING, Any, Final
from bleak.backends.scanner import AdvertisementDataCallback
from bleak_retry_connector import NO_RSSI_VALUE, RSSI_SWITCH_THRESHOLD
from bleak_retry_connector import NO_RSSI_VALUE, RSSI_SWITCH_THRESHOLD, BleakSlotManager
from bluetooth_adapters import (
ADAPTER_ADDRESS,
ADAPTER_PASSIVE_SCAN,
@ -104,6 +104,7 @@ class BluetoothManager:
integration_matcher: IntegrationMatcher,
bluetooth_adapters: BluetoothAdapters,
storage: BluetoothStorage,
slot_manager: BleakSlotManager,
) -> None:
"""Init bluetooth manager."""
self.hass = hass
@ -131,6 +132,7 @@ class BluetoothManager:
self._sources: dict[str, BaseHaScanner] = {}
self._bluetooth_adapters = bluetooth_adapters
self.storage = storage
self.slot_manager = slot_manager
@property
def supports_passive_scan(self) -> bool:
@ -155,6 +157,7 @@ class BluetoothManager:
)
return {
"adapters": self._adapters,
"slot_manager": self.slot_manager.diagnostics(),
"scanners": scanner_diagnostics,
"connectable_history": [
service_info.as_dict()
@ -642,7 +645,10 @@ class BluetoothManager:
return self._connectable_history if connectable else self._all_history
def async_register_scanner(
self, scanner: BaseHaScanner, connectable: bool
self,
scanner: BaseHaScanner,
connectable: bool,
connection_slots: int | None = None,
) -> CALLBACK_TYPE:
"""Register a new scanner."""
_LOGGER.debug("Registering scanner %s", scanner.name)
@ -653,9 +659,13 @@ class BluetoothManager:
self._advertisement_tracker.async_remove_source(scanner.source)
scanners.remove(scanner)
del self._sources[scanner.source]
if connection_slots:
self.slot_manager.remove_adapter(scanner.adapter)
scanners.append(scanner)
self._sources[scanner.source] = scanner
if connection_slots:
self.slot_manager.register_adapter(scanner.adapter, connection_slots)
return _unregister_scanner
@hass_callback
@ -679,3 +689,13 @@ class BluetoothManager:
)
return _remove_callback
@hass_callback
def async_release_connection_slot(self, device: BLEDevice) -> None:
"""Release a connection slot."""
self.slot_manager.release_slot(device)
@hass_callback
def async_allocate_connection_slot(self, device: BLEDevice) -> bool:
"""Allocate a connection slot."""
return self.slot_manager.allocate_slot(device)

4
homeassistant/components/bluetooth/manifest.json
View file

@ -7,8 +7,8 @@
"quality_scale": "internal",
"requirements": [
"bleak==0.19.2",
"bleak-retry-connector==2.10.2",
"bluetooth-adapters==0.14.1",
"bleak-retry-connector==2.12.1",
"bluetooth-adapters==0.15.2",
"bluetooth-auto-recovery==1.0.3",
"bluetooth-data-tools==0.3.1",
"dbus-fast==1.82.0"

1
homeassistant/components/bluetooth/scanner.py
View file

@ -132,7 +132,6 @@ class HaScanner(BaseHaScanner):
super().__init__(hass, source, adapter)
self.connectable = True
self.mode = mode
self.adapter = adapter
self._start_stop_lock = asyncio.Lock()
self._new_info_callback = new_info_callback
self.scanning = False

28
homeassistant/components/bluetooth/wrappers.py
View file

@ -12,7 +12,12 @@ from bleak import BleakClient, BleakError
from bleak.backends.client import BaseBleakClient, get_platform_client_backend_type
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementDataCallback, BaseBleakScanner
from bleak_retry_connector import NO_RSSI_VALUE, ble_device_description, clear_cache
from bleak_retry_connector import (
NO_RSSI_VALUE,
ble_device_description,
clear_cache,
device_source,
)
from homeassistant.core import CALLBACK_TYPE, callback as hass_callback
from homeassistant.helpers.frame import report
@ -33,6 +38,7 @@ class _HaWrappedBleakBackend:
device: BLEDevice
client: type[BaseBleakClient]
source: str | None
class HaBleakScannerWrapper(BaseBleakScanner):
@ -203,7 +209,15 @@ class HaBleakClientWrapper(BleakClient):
description = ble_device_description(wrapped_backend.device)
rssi = wrapped_backend.device.rssi
_LOGGER.debug("%s: Connecting (last rssi: %s)", description, rssi)
connected = await super().connect(**kwargs)
connected = None
try:
connected = await super().connect(**kwargs)
finally:
# If we failed to connect and its a local adapter (no source)
# we release the connection slot
if not connected and not wrapped_backend.source:
models.MANAGER.async_release_connection_slot(wrapped_backend.device)
if debug_logging:
_LOGGER.debug("%s: Connected (last rssi: %s)", description, rssi)
return connected
@ -213,14 +227,14 @@ class HaBleakClientWrapper(BleakClient):
self, manager: BluetoothManager, ble_device: BLEDevice
) -> _HaWrappedBleakBackend | None:
"""Get the backend for a BLEDevice."""
details = ble_device.details
if not isinstance(details, dict) or "source" not in details:
if not (source := device_source(ble_device)):
# If client is not defined in details
# its the client for this platform
if not manager.async_allocate_connection_slot(ble_device):
return None
cls = get_platform_client_backend_type()
return _HaWrappedBleakBackend(ble_device, cls)
return _HaWrappedBleakBackend(ble_device, cls, source)
source: str = details["source"]
# Make sure the backend can connect to the device
# as some backends have connection limits
if (
@ -230,7 +244,7 @@ class HaBleakClientWrapper(BleakClient):
):
return None
return _HaWrappedBleakBackend(ble_device, scanner.connector.client)
return _HaWrappedBleakBackend(ble_device, scanner.connector.client, source)
@hass_callback
def _async_get_best_available_backend_and_device(

4
homeassistant/package_constraints.txt
View file

@ -10,9 +10,9 @@ atomicwrites-homeassistant==1.4.1
attrs==22.1.0
awesomeversion==22.9.0
bcrypt==3.1.7
bleak-retry-connector==2.10.2
bleak-retry-connector==2.12.1
bleak==0.19.2
bluetooth-adapters==0.14.1
bluetooth-adapters==0.15.2
bluetooth-auto-recovery==1.0.3
bluetooth-data-tools==0.3.1
certifi>=2021.5.30

4
requirements_all.txt
View file

@ -428,7 +428,7 @@ bimmer_connected==0.10.4
bizkaibus==0.1.1
# homeassistant.components.bluetooth
bleak-retry-connector==2.10.2
bleak-retry-connector==2.12.1
# homeassistant.components.bluetooth
bleak==0.19.2
@ -453,7 +453,7 @@ bluemaestro-ble==0.2.0
# bluepy==1.3.0
# homeassistant.components.bluetooth
bluetooth-adapters==0.14.1
bluetooth-adapters==0.15.2
# homeassistant.components.bluetooth
bluetooth-auto-recovery==1.0.3

4
requirements_test_all.txt
View file

@ -352,7 +352,7 @@ bellows==0.34.5
bimmer_connected==0.10.4
# homeassistant.components.bluetooth
bleak-retry-connector==2.10.2
bleak-retry-connector==2.12.1
# homeassistant.components.bluetooth
bleak==0.19.2
@ -367,7 +367,7 @@ blinkpy==0.19.2
bluemaestro-ble==0.2.0
# homeassistant.components.bluetooth
bluetooth-adapters==0.14.1
bluetooth-adapters==0.15.2
# homeassistant.components.bluetooth
bluetooth-auto-recovery==1.0.3

2
tests/components/bluetooth/conftest.py
View file

@ -140,6 +140,7 @@ def two_adapters_fixture():
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 1,
},
"hci1": {
"address": "00:00:00:00:00:02",
@ -150,6 +151,7 @@ def two_adapters_fixture():
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 2,
},
},
):

21
tests/components/bluetooth/test_diagnostics.py
View file

@ -86,6 +86,7 @@ async def test_diagnostics(
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 1,
},
"hci1": {
"address": "00:00:00:00:00:02",
@ -96,6 +97,7 @@ async def test_diagnostics(
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 2,
},
},
"dbus": {
@ -115,6 +117,11 @@ async def test_diagnostics(
}
},
"manager": {
"slot_manager": {
"adapter_slots": {"hci0": 5, "hci1": 2},
"allocations_by_adapter": {"hci0": [], "hci1": []},
"manager": False,
},
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",
@ -125,6 +132,7 @@ async def test_diagnostics(
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 1,
},
"hci1": {
"address": "00:00:00:00:00:02",
@ -135,6 +143,7 @@ async def test_diagnostics(
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"vendor_id": "cc01",
"connection_slots": 2,
},
},
"advertisement_tracker": {
@ -274,6 +283,7 @@ async def test_diagnostics_macos(
inject_advertisement(hass, switchbot_device, switchbot_adv)
diag = await get_diagnostics_for_config_entry(hass, hass_client, entry1)
assert diag == {
"adapters": {
"Core Bluetooth": {
@ -287,6 +297,11 @@ async def test_diagnostics_macos(
}
},
"manager": {
"slot_manager": {
"adapter_slots": {"Core Bluetooth": 5},
"allocations_by_adapter": {"Core Bluetooth": []},
"manager": False,
},
"adapters": {
"Core Bluetooth": {
"address": "00:00:00:00:00:00",
@ -457,6 +472,7 @@ async def test_diagnostics_remote_adapter(
inject_advertisement(hass, switchbot_device, switchbot_adv)
diag = await get_diagnostics_for_config_entry(hass, hass_client, entry1)
assert diag == {
"adapters": {
"hci0": {
@ -472,6 +488,11 @@ async def test_diagnostics_remote_adapter(
},
"dbus": {},
"manager": {
"slot_manager": {
"adapter_slots": {"hci0": 5},
"allocations_by_adapter": {"hci0": []},
"manager": False,
},
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",

298
tests/components/bluetooth/test_wrappers.py Normal file
View file

@ -0,0 +1,298 @@
"""Tests for the Bluetooth integration."""
from collections.abc import Callable
from typing import Union
from unittest.mock import patch
import bleak
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
import pytest
from homeassistant.components.bluetooth import (
BaseHaRemoteScanner,
BluetoothServiceInfoBleak,
HaBluetoothConnector,
async_get_advertisement_callback,
)
from homeassistant.components.bluetooth.usage import (
install_multiple_bleak_catcher,
uninstall_multiple_bleak_catcher,
)
from homeassistant.core import HomeAssistant
from . import _get_manager, generate_advertisement_data
class FakeScanner(BaseHaRemoteScanner):
"""Fake scanner."""
def __init__(
self,
hass: HomeAssistant,
scanner_id: str,
name: str,
new_info_callback: Callable[[BluetoothServiceInfoBleak], None],
connector: None,
connectable: bool,
) -> None:
"""Initialize the scanner."""
super().__init__(
hass, scanner_id, name, new_info_callback, connector, connectable
)
self._details: dict[str, str | HaBluetoothConnector] = {}
def inject_advertisement(
self, device: BLEDevice, advertisement_data: AdvertisementData
) -> None:
"""Inject an advertisement."""
self._async_on_advertisement(
device.address,
advertisement_data.rssi,
device.name,
advertisement_data.service_uuids,
advertisement_data.service_data,
advertisement_data.manufacturer_data,
advertisement_data.tx_power,
device.details | {"scanner_specific_data": "test"},
)
class BaseFakeBleakClient:
"""Base class for fake bleak clients."""
def __init__(self, address_or_ble_device: Union[BLEDevice, str], **kwargs):
"""Initialize the fake bleak client."""
self._device_path = "/dev/test"
self._address = address_or_ble_device.address
async def disconnect(self, *args, **kwargs):
"""Disconnect.""" ""
async def get_services(self, *args, **kwargs):
"""Get services."""
return []
class FakeBleakClient(BaseFakeBleakClient):
"""Fake bleak client."""
async def connect(self, *args, **kwargs):
"""Connect."""
return True
class FakeBleakClientFailsToConnect(BaseFakeBleakClient):
"""Fake bleak client that fails to connect."""
async def connect(self, *args, **kwargs):
"""Connect."""
return False
class FakeBleakClientRaisesOnConnect(BaseFakeBleakClient):
"""Fake bleak client that raises on connect."""
async def connect(self, *args, **kwargs):
"""Connect."""
raise Exception("Test exception")
def _generate_ble_device_and_adv_data(
interface: str, mac: str
) -> tuple[BLEDevice, AdvertisementData]:
"""Generate a BLE device with adv data."""
return (
BLEDevice(
mac,
"any",
delegate="",
details={"path": f"/org/bluez/{interface}/dev_{mac}"},
),
generate_advertisement_data(),
)
@pytest.fixture(name="install_bleak_catcher")
def install_bleak_catcher_fixture():
"""Fixture that installs the bleak catcher."""
install_multiple_bleak_catcher()
yield
uninstall_multiple_bleak_catcher()
@pytest.fixture(name="mock_platform_client")
def mock_platform_client_fixture():
"""Fixture that mocks the platform client."""
with patch(
"homeassistant.components.bluetooth.wrappers.get_platform_client_backend_type",
return_value=FakeBleakClient,
):
yield
@pytest.fixture(name="mock_platform_client_that_fails_to_connect")
def mock_platform_client_that_fails_to_connect_fixture():
"""Fixture that mocks the platform client that fails to connect."""
with patch(
"homeassistant.components.bluetooth.wrappers.get_platform_client_backend_type",
return_value=FakeBleakClientFailsToConnect,
):
yield
@pytest.fixture(name="mock_platform_client_that_raises_on_connect")
def mock_platform_client_that_raises_on_connect_fixture():
"""Fixture that mocks the platform client that fails to connect."""
with patch(
"homeassistant.components.bluetooth.wrappers.get_platform_client_backend_type",
return_value=FakeBleakClientRaisesOnConnect,
):
yield
def _generate_scanners_with_fake_devices(hass):
"""Generate scanners with fake devices."""
manager = _get_manager()
hci0_device_advs = {}
for i in range(10):
device, adv_data = _generate_ble_device_and_adv_data(
"hci0", f"00:00:00:00:00:{i:02x}"
)
hci0_device_advs[device.address] = (device, adv_data)
hci1_device_advs = {}
for i in range(10):
device, adv_data = _generate_ble_device_and_adv_data(
"hci1", f"00:00:00:00:00:{i:02x}"
)
hci1_device_advs[device.address] = (device, adv_data)
new_info_callback = async_get_advertisement_callback(hass)
scanner_hci0 = FakeScanner(
hass, "00:00:00:00:00:01", "hci0", new_info_callback, None, True
)
scanner_hci1 = FakeScanner(
hass, "00:00:00:00:00:02", "hci1", new_info_callback, None, True
)
for (device, adv_data) in hci0_device_advs.values():
scanner_hci0.inject_advertisement(device, adv_data)
for (device, adv_data) in hci1_device_advs.values():
scanner_hci1.inject_advertisement(device, adv_data)
cancel_hci0 = manager.async_register_scanner(scanner_hci0, True, 2)
cancel_hci1 = manager.async_register_scanner(scanner_hci1, True, 1)
return hci0_device_advs, cancel_hci0, cancel_hci1
async def test_test_switch_adapters_when_out_of_slots(
hass, two_adapters, enable_bluetooth, install_bleak_catcher, mock_platform_client
):
"""Ensure we try another scanner when one runs out of slots."""
manager = _get_manager()
hci0_device_advs, cancel_hci0, cancel_hci1 = _generate_scanners_with_fake_devices(
hass
)
# hci0 has 2 slots, hci1 has 1 slot
with patch.object(
manager.slot_manager, "release_slot"
) as release_slot_mock, patch.object(
manager.slot_manager, "allocate_slot", return_value=True
) as allocate_slot_mock:
ble_device = hci0_device_advs["00:00:00:00:00:01"][0]
client = bleak.BleakClient(ble_device)
assert await client.connect() is True
assert allocate_slot_mock.call_count == 1
assert release_slot_mock.call_count == 0
# All adapters are out of slots
with patch.object(
manager.slot_manager, "release_slot"
) as release_slot_mock, patch.object(
manager.slot_manager, "allocate_slot", return_value=False
) as allocate_slot_mock:
ble_device = hci0_device_advs["00:00:00:00:00:02"][0]
client = bleak.BleakClient(ble_device)
with pytest.raises(bleak.exc.BleakError):
await client.connect()
assert allocate_slot_mock.call_count == 2
assert release_slot_mock.call_count == 0
# When hci0 runs out of slots, we should try hci1
def _allocate_slot_mock(ble_device: BLEDevice):
if "hci1" in ble_device.details["path"]:
return True
return False
with patch.object(
manager.slot_manager, "release_slot"
) as release_slot_mock, patch.object(
manager.slot_manager, "allocate_slot", _allocate_slot_mock
) as allocate_slot_mock:
ble_device = hci0_device_advs["00:00:00:00:00:03"][0]
client = bleak.BleakClient(ble_device)
await client.connect() is True
assert release_slot_mock.call_count == 0
cancel_hci0()
cancel_hci1()
async def test_release_slot_on_connect_failure(
hass,
two_adapters,
enable_bluetooth,
install_bleak_catcher,
mock_platform_client_that_fails_to_connect,
):
"""Ensure the slot gets released on connection failure."""
manager = _get_manager()
hci0_device_advs, cancel_hci0, cancel_hci1 = _generate_scanners_with_fake_devices(
hass
)
# hci0 has 2 slots, hci1 has 1 slot
with patch.object(
manager.slot_manager, "release_slot"
) as release_slot_mock, patch.object(
manager.slot_manager, "allocate_slot", return_value=True
) as allocate_slot_mock:
ble_device = hci0_device_advs["00:00:00:00:00:01"][0]
client = bleak.BleakClient(ble_device)
assert await client.connect() is False
assert allocate_slot_mock.call_count == 1
assert release_slot_mock.call_count == 1
cancel_hci0()
cancel_hci1()
async def test_release_slot_on_connect_exception(
hass,
two_adapters,
enable_bluetooth,
install_bleak_catcher,
mock_platform_client_that_raises_on_connect,
):
"""Ensure the slot gets released on connection exception."""
manager = _get_manager()
hci0_device_advs, cancel_hci0, cancel_hci1 = _generate_scanners_with_fake_devices(
hass
)
# hci0 has 2 slots, hci1 has 1 slot
with patch.object(
manager.slot_manager, "release_slot"
) as release_slot_mock, patch.object(
manager.slot_manager, "allocate_slot", return_value=True
) as allocate_slot_mock:
ble_device = hci0_device_advs["00:00:00:00:00:01"][0]
client = bleak.BleakClient(ble_device)
with pytest.raises(Exception):
assert await client.connect() is False
assert allocate_slot_mock.call_count == 1
assert release_slot_mock.call_count == 1
cancel_hci0()
cancel_hci1()