jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 09:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump bleak to 0.19.0 (#80349)

Browse Source
This commit is contained in:
J. Nick Koston 2022-10-15 07:57:23 -10:00 committed by GitHub
parent 3460e0b074
commit d12cbab6c4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
33 changed files with 638 additions and 390 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/bluetooth/const.py
View File

@ -74,4 +74,4 @@ ADAPTER_HW_VERSION: Final = "hw_version"
ADAPTER_PASSIVE_SCAN: Final = "passive_scan"
NO_RSSI_VALUE: Final = -1000
NO_RSSI_VALUE: Final = -127

153
homeassistant/components/bluetooth/manager.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Iterable
from dataclasses import replace
from datetime import datetime, timedelta
import itertools
import logging
@ -121,7 +122,8 @@ class BluetoothManager:
self._bleak_callbacks: list[
tuple[AdvertisementDataCallback, dict[str, set[str]]]
] = []
self._history: dict[str, BluetoothServiceInfoBleak] = {}
self._all_history: dict[str, BluetoothServiceInfoBleak] = {}
self._non_connectable_history: dict[str, BluetoothServiceInfoBleak] = {}
self._connectable_history: dict[str, BluetoothServiceInfoBleak] = {}
self._non_connectable_scanners: list[BaseHaScanner] = []
self._connectable_scanners: list[BaseHaScanner] = []
@ -155,8 +157,9 @@ class BluetoothManager:
service_info.as_dict()
for service_info in self._connectable_history.values()
],
"history": [
service_info.as_dict() for service_info in self._history.values()
"non_connectable_history": [
service_info.as_dict()
for service_info in self._non_connectable_history.values()
],
"advertisement_tracker": self._advertisement_tracker.async_diagnostics(),
}
@ -189,7 +192,7 @@ class BluetoothManager:
# Everything is connectable so it fall into both
# buckets since the host system can only provide
# connectable devices
self._history = history.copy()
self._all_history = history.copy()
self._connectable_history = history.copy()
self.async_setup_unavailable_tracking()
@ -202,32 +205,32 @@ class BluetoothManager:
self._cancel_unavailable_tracking = None
uninstall_multiple_bleak_catcher()
async def async_get_devices_by_address(
@hass_callback
def async_get_discovered_devices_and_advertisement_data_by_address(
self, address: str, connectable: bool
) -> list[BLEDevice]:
"""Get devices by address."""
) -> list[tuple[BLEDevice, AdvertisementData]]:
"""Get devices and advertisement_data by address."""
types_ = (True,) if connectable else (True, False)
return [
device
for device in await asyncio.gather(
*(
scanner.async_get_device_by_address(address)
for type_ in types_
for scanner in self._get_scanners_by_type(type_)
)
device_advertisement_data
for device_advertisement_data in (
scanner.discovered_devices_and_advertisement_data.get(address)
for type_ in types_
for scanner in self._get_scanners_by_type(type_)
)
if device is not None
if device_advertisement_data is not None
]
@hass_callback
def async_all_discovered_devices(self, connectable: bool) -> Iterable[BLEDevice]:
"""Return all of discovered devices from all the scanners including duplicates."""
def _async_all_discovered_addresses(self, connectable: bool) -> Iterable[str]:
"""Return all of discovered addresses from all the scanners including duplicates."""
yield from itertools.chain.from_iterable(
scanner.discovered_devices for scanner in self._get_scanners_by_type(True)
scanner.discovered_devices_and_advertisement_data
for scanner in self._get_scanners_by_type(True)
)
if not connectable:
yield from itertools.chain.from_iterable(
scanner.discovered_devices
scanner.discovered_devices_and_advertisement_data
for scanner in self._get_scanners_by_type(False)
)
@ -253,33 +256,38 @@ class BluetoothManager:
"""Watch for unavailable devices and cleanup state history."""
monotonic_now = MONOTONIC_TIME()
connectable_history = self._connectable_history
all_history = self._history
removed_addresses: set[str] = set()
non_connectable_history = self._non_connectable_history
all_history = self._all_history
tracker = self._advertisement_tracker
intervals = tracker.intervals
for connectable in (True, False):
unavailable_callbacks = self._get_unavailable_callbacks_by_type(connectable)
intervals = self._advertisement_tracker.intervals
history = connectable_history if connectable else all_history
history_set = set(history)
active_addresses = {
device.address
for device in self.async_all_discovered_devices(connectable)
}
disappeared = history_set.difference(active_addresses)
disappeared = set(history).difference(
self._async_all_discovered_addresses(connectable)
)
for address in disappeared:
#
# For non-connectable devices we also check the device has exceeded
# the advertising interval before we mark it as unavailable
# since it may have gone to sleep and since we do not need an active connection
# to it we can only determine its availability by the lack of advertisements
#
if not connectable and (advertising_interval := intervals.get(address)):
time_since_seen = monotonic_now - history[address].time
if time_since_seen <= advertising_interval:
continue
if not connectable:
#
# For non-connectable devices we also check the device has exceeded
# the advertising interval before we mark it as unavailable
# since it may have gone to sleep and since we do not need an active connection
# to it we can only determine its availability by the lack of advertisements
#
if advertising_interval := intervals.get(address):
time_since_seen = monotonic_now - all_history[address].time
if time_since_seen <= advertising_interval:
continue
non_connectable_history.pop(address, None)
# The second loop (connectable=False) is responsible for removing
# the device from all the interval tracking since it is no longer
# available for both connectable and non-connectable
tracker.async_remove_address(address)
service_info = history.pop(address)
removed_addresses.add(address)
if not (callbacks := unavailable_callbacks.get(address)):
continue
@ -290,14 +298,10 @@ class BluetoothManager:
except Exception: # pylint: disable=broad-except
_LOGGER.exception("Error in unavailable callback")
# If we removed the device from both the connectable history
# and all history then we can remove it from the advertisement tracker
for address in removed_addresses:
if address not in connectable_history and address not in all_history:
self._advertisement_tracker.async_remove_address(address)
def _prefer_previous_adv_from_different_source(
self, old: BluetoothServiceInfoBleak, new: BluetoothServiceInfoBleak
self,
old: BluetoothServiceInfoBleak,
new: BluetoothServiceInfoBleak,
) -> bool:
"""Prefer previous advertisement from a different source if it is better."""
if new.time - old.time > (
@ -308,8 +312,8 @@ class BluetoothManager:
# If the old advertisement is stale, any new advertisement is preferred
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (time elapsed:%s > stale seconds:%s)",
new.advertisement.local_name,
new.device.address,
new.name,
new.address,
old.source,
old.connectable,
new.source,
@ -318,19 +322,21 @@ class BluetoothManager:
stale_seconds,
)
return False
if new.device.rssi - RSSI_SWITCH_THRESHOLD > (old.device.rssi or NO_RSSI_VALUE):
if (new.rssi or NO_RSSI_VALUE) - RSSI_SWITCH_THRESHOLD > (
old.rssi or NO_RSSI_VALUE
):
# If new advertisement is RSSI_SWITCH_THRESHOLD more, the new one is preferred
_LOGGER.debug(
"%s (%s): Switching from %s[%s] to %s[%s] (new rssi:%s - threshold:%s > old rssi:%s)",
new.advertisement.local_name,
new.device.address,
new.name,
new.address,
old.source,
old.connectable,
new.source,
new.connectable,
new.device.rssi,
new.rssi,
RSSI_SWITCH_THRESHOLD,
old.device.rssi,
old.rssi,
)
return False
return True
@ -355,9 +361,9 @@ class BluetoothManager:
return
device = service_info.device
connectable = service_info.connectable
address = device.address
all_history = self._connectable_history if connectable else self._history
all_history = self._all_history
source = service_info.source
if (
(old_service_info := all_history.get(address))
@ -368,11 +374,11 @@ class BluetoothManager:
):
return
self._history[address] = service_info
if connectable:
if connectable := service_info.connectable:
self._connectable_history[address] = service_info
# Bleak callbacks must get a connectable device
else:
self._non_connectable_history[address] = service_info
all_history[address] = service_info
# Track advertisement intervals to determine when we need to
# switch adapters or mark a device as unavailable
@ -393,11 +399,18 @@ class BluetoothManager:
):
return
if connectable:
if is_connectable_by_any_source := address in self._connectable_history:
# Bleak callbacks must get a connectable device
for callback_filters in self._bleak_callbacks:
_dispatch_bleak_callback(*callback_filters, device, advertisement_data)
if not connectable and is_connectable_by_any_source:
# Since we have a connectable path and our BleakClient will
# route any connection attempts to the connectable path, we
# mark the service_info as connectable so that the callbacks
# will be called and the device can be discovered.
service_info = replace(service_info, connectable=True)
matched_domains = self._integration_matcher.match_domains(service_info)
_LOGGER.debug(
"%s: %s %s connectable: %s match: %s rssi: %s",
@ -406,7 +419,7 @@ class BluetoothManager:
advertisement_data,
connectable,
matched_domains,
device.rssi,
advertisement_data.rssi,
)
for match in self._callback_index.match_callbacks(service_info):
@ -518,27 +531,23 @@ class BluetoothManager:
def _get_scanners_by_type(self, connectable: bool) -> list[BaseHaScanner]:
"""Return the scanners by type."""
return (
self._connectable_scanners
if connectable
else self._non_connectable_scanners
)
if connectable:
return self._connectable_scanners
return self._non_connectable_scanners
def _get_unavailable_callbacks_by_type(
self, connectable: bool
) -> dict[str, list[Callable[[BluetoothServiceInfoBleak], None]]]:
"""Return the unavailable callbacks by type."""
return (
self._connectable_unavailable_callbacks
if connectable
else self._unavailable_callbacks
)
if connectable:
return self._connectable_unavailable_callbacks
return self._unavailable_callbacks
def _get_history_by_type(
self, connectable: bool
) -> dict[str, BluetoothServiceInfoBleak]:
"""Return the history by type."""
return self._connectable_history if connectable else self._history
return self._connectable_history if connectable else self._all_history
def async_register_scanner(
self, scanner: BaseHaScanner, connectable: bool

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

@ -6,8 +6,8 @@
"after_dependencies": ["hassio"],
"quality_scale": "internal",
"requirements": [
"bleak==0.18.1",
"bleak-retry-connector==2.1.3",
"bleak==0.19.0",
"bleak-retry-connector==2.2.0",
"bluetooth-adapters==0.6.0",
"bluetooth-auto-recovery==0.3.4",
"dbus-fast==1.45.0"

30
homeassistant/components/bluetooth/models.py
View File

@ -115,9 +115,12 @@ class BaseHaScanner:
def discovered_devices(self) -> list[BLEDevice]:
"""Return a list of discovered devices."""
@property
@abstractmethod
async def async_get_device_by_address(self, address: str) -> BLEDevice | None:
"""Get a device by address."""
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices and their advertisement data."""
async def async_diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the scanner."""
@ -127,7 +130,6 @@ class BaseHaScanner:
{
"name": device.name,
"address": device.address,
"rssi": device.rssi,
}
for device in self.discovered_devices
],
@ -285,7 +287,7 @@ class HaBleakClientWrapper(BleakClient):
"""Connect to the specified GATT server."""
if not self._backend:
wrapped_backend = (
self._async_get_backend() or await self._async_get_fallback_backend()
self._async_get_backend() or self._async_get_fallback_backend()
)
self._backend = wrapped_backend.client(
await freshen_ble_device(wrapped_backend.device)
@ -329,7 +331,8 @@ class HaBleakClientWrapper(BleakClient):
return None
async def _async_get_fallback_backend(self) -> _HaWrappedBleakBackend:
@hass_callback
def _async_get_fallback_backend(self) -> _HaWrappedBleakBackend:
"""Get a fallback backend for the given address."""
#
# The preferred backend cannot currently connect the device
@ -340,13 +343,20 @@ class HaBleakClientWrapper(BleakClient):
#
assert MANAGER is not None
address = self.__address
devices = await MANAGER.async_get_devices_by_address(address, True)
for ble_device in sorted(
devices,
key=lambda ble_device: ble_device.rssi or NO_RSSI_VALUE,
device_advertisement_datas = (
MANAGER.async_get_discovered_devices_and_advertisement_data_by_address(
address, True
)
)
for device_advertisement_data in sorted(
device_advertisement_datas,
key=lambda device_advertisement_data: device_advertisement_data[1].rssi
or NO_RSSI_VALUE,
reverse=True,
):
if backend := self._async_get_backend_for_ble_device(ble_device):
if backend := self._async_get_backend_for_ble_device(
device_advertisement_data[0]
):
return backend
raise BleakError(

18
homeassistant/components/bluetooth/scanner.py
View File

@ -17,7 +17,6 @@ from bleak.backends.bluezdbus.advertisement_monitor import OrPattern
from bleak.backends.bluezdbus.scanner import BlueZScannerArgs
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData, AdvertisementDataCallback
from bleak_retry_connector import get_device_by_adapter
from dbus_fast import InvalidMessageError
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
@ -144,6 +143,13 @@ class HaScanner(BaseHaScanner):
"""Return a list of discovered devices."""
return self.scanner.discovered_devices
@property
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices and advertisement data."""
return self.scanner.discovered_devices_and_advertisement_data
@hass_callback
def async_setup(self) -> None:
"""Set up the scanner."""
@ -151,16 +157,6 @@ class HaScanner(BaseHaScanner):
self._async_detection_callback, self.mode, self.adapter
)
async def async_get_device_by_address(self, address: str) -> BLEDevice | None:
"""Get a device by address."""
if platform.system() == "Linux":
return await get_device_by_adapter(address, self.adapter)
# We don't have a fast version of this for MacOS yet
return next(
(device for device in self.discovered_devices if device.address == address),
None,
)
async def async_diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the scanner."""
base_diag = await super().async_diagnostics()

41
homeassistant/components/esphome/bluetooth/scanner.py
View File

@ -37,7 +37,9 @@ class ESPHomeScanner(BaseHaScanner):
"""Initialize the scanner."""
super().__init__(hass, scanner_id)
self._new_info_callback = new_info_callback
self._discovered_devices: dict[str, BLEDevice] = {}
self._discovered_device_advertisement_datas: dict[
str, tuple[BLEDevice, AdvertisementData]
] = {}
self._discovered_device_timestamps: dict[str, float] = {}
self._connector = connector
self._connectable = connectable
@ -61,17 +63,23 @@ class ESPHomeScanner(BaseHaScanner):
if now - timestamp > FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
]
for address in expired:
del self._discovered_devices[address]
del self._discovered_device_advertisement_datas[address]
del self._discovered_device_timestamps[address]
@property
def discovered_devices(self) -> list[BLEDevice]:
"""Return a list of discovered devices."""
return list(self._discovered_devices.values())
return [
device_advertisement_data[0]
for device_advertisement_data in self._discovered_device_advertisement_datas.values()
]
async def async_get_device_by_address(self, address: str) -> BLEDevice | None:
"""Get a device by address."""
return self._discovered_devices.get(address)
@property
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices and advertisement data."""
return self._discovered_device_advertisement_datas
@callback
def async_on_advertisement(self, adv: BluetoothLEAdvertisement) -> None:
@ -79,32 +87,39 @@ class ESPHomeScanner(BaseHaScanner):
now = time.monotonic()
address = ":".join(TWO_CHAR.findall("%012X" % adv.address)) # must be upper
name = adv.name
if prev_discovery := self._discovered_devices.get(address):
if prev_discovery := self._discovered_device_advertisement_datas.get(address):
# If the last discovery had the full local name
# and this one doesn't, keep the old one as we
# always want the full local name over the short one
if len(prev_discovery.name) > len(adv.name):
name = prev_discovery.name
prev_device = prev_discovery[0]
if len(prev_device.name) > len(adv.name):
name = prev_device.name
advertisement_data = AdvertisementData( # type: ignore[no-untyped-call]
advertisement_data = AdvertisementData(
local_name=None if name == "" else name,
manufacturer_data=adv.manufacturer_data,
service_data=adv.service_data,
service_uuids=adv.service_uuids,
rssi=adv.rssi,
tx_power=-127,
platform_data=(),
)
device = BLEDevice( # type: ignore[no-untyped-call]
address=address,
name=name,
details=self._details,
rssi=adv.rssi,
rssi=adv.rssi, # deprecated, will be removed in newer bleak
)
self._discovered_device_advertisement_datas[address] = (
device,
advertisement_data,
)
self._discovered_devices[address] = device
self._discovered_device_timestamps[address] = now
self._new_info_callback(
BluetoothServiceInfoBleak(
name=advertisement_data.local_name or device.name or device.address,
address=device.address,
rssi=device.rssi,
rssi=adv.rssi,
manufacturer_data=advertisement_data.manufacturer_data,
service_data=advertisement_data.service_data,
service_uuids=advertisement_data.service_uuids,

9
homeassistant/components/homekit_controller/sensor.py
View File

@ -10,7 +10,10 @@ from aiohomekit.model.characteristics import Characteristic, CharacteristicsType
from aiohomekit.model.characteristics.const import ThreadNodeCapabilities, ThreadStatus
from aiohomekit.model.services import Service, ServicesTypes
from homeassistant.components.bluetooth import async_ble_device_from_address
from homeassistant.components.bluetooth import (
async_ble_device_from_address,
async_last_service_info,
)
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -571,8 +574,8 @@ class RSSISensor(HomeKitEntity, SensorEntity):
def native_value(self) -> int | None:
"""Return the current rssi value."""
address = self._accessory.pairing_data["AccessoryAddress"]
ble_device = async_ble_device_from_address(self.hass, address)
return ble_device.rssi if ble_device else None
last_service_info = async_last_service_info(self.hass, address)
return last_service_info.rssi if last_service_info else None
async def async_setup_entry(

2
homeassistant/components/switchbot/coordinator.py
View File

@ -60,6 +60,7 @@ class SwitchbotDataUpdateCoordinator(PassiveBluetoothDataUpdateCoordinator):
self.device_name = device_name
self.base_unique_id = base_unique_id
self.model = model
self.service_info: bluetooth.BluetoothServiceInfoBleak | None = None
self._ready_event = asyncio.Event()
@callback
@ -70,6 +71,7 @@ class SwitchbotDataUpdateCoordinator(PassiveBluetoothDataUpdateCoordinator):
) -> None:
"""Handle a Bluetooth event."""
self.ble_device = service_info.device
self.service_info = service_info
if adv := switchbot.parse_advertisement_data(
service_info.device, service_info.advertisement
):

3
homeassistant/components/switchbot/sensor.py
View File

@ -117,4 +117,5 @@ class SwitchbotRSSISensor(SwitchBotSensor):
@property
def native_value(self) -> str | int:
"""Return the state of the sensor."""
return self.coordinator.ble_device.rssi
assert self.coordinator.service_info is not None
return self.coordinator.service_info.rssi

6
homeassistant/package_constraints.txt
View File

@ -10,8 +10,8 @@ atomicwrites-homeassistant==1.4.1
attrs==21.2.0
awesomeversion==22.9.0
bcrypt==3.1.7
bleak-retry-connector==2.1.3
bleak==0.18.1
bleak-retry-connector==2.2.0
bleak==0.19.0
bluetooth-adapters==0.6.0
bluetooth-auto-recovery==0.3.4
certifi>=2021.5.30
@ -20,7 +20,7 @@ cryptography==38.0.1
dbus-fast==1.45.0
fnvhash==0.1.0
hass-nabucasa==0.56.0
home-assistant-bluetooth==1.3.0
home-assistant-bluetooth==1.6.0
home-assistant-frontend==20221010.0
httpx==0.23.0
ifaddr==0.1.7

2
pyproject.toml
View File

@ -36,7 +36,7 @@ dependencies = [
# When bumping httpx, please check the version pins of
# httpcore, anyio, and h11 in gen_requirements_all
"httpx==0.23.0",
"home-assistant-bluetooth==1.3.0",
"home-assistant-bluetooth==1.6.0",
"ifaddr==0.1.7",
"jinja2==3.1.2",
"lru-dict==1.1.8",

2
requirements.txt
View File

@ -11,7 +11,7 @@ bcrypt==3.1.7
certifi>=2021.5.30
ciso8601==2.2.0
httpx==0.23.0
home-assistant-bluetooth==1.3.0
home-assistant-bluetooth==1.6.0
ifaddr==0.1.7
jinja2==3.1.2
lru-dict==1.1.8

4
requirements_all.txt
View File

@ -413,10 +413,10 @@ bimmer_connected==0.10.4
bizkaibus==0.1.1
# homeassistant.components.bluetooth
bleak-retry-connector==2.1.3
bleak-retry-connector==2.2.0
# homeassistant.components.bluetooth
bleak==0.18.1
bleak==0.19.0
# homeassistant.components.blebox
blebox_uniapi==2.1.0

4
requirements_test_all.txt
View File

@ -337,10 +337,10 @@ bellows==0.34.2
bimmer_connected==0.10.4
# homeassistant.components.bluetooth
bleak-retry-connector==2.1.3
bleak-retry-connector==2.2.0
# homeassistant.components.bluetooth
bleak==0.18.1
bleak==0.19.0
# homeassistant.components.blebox
blebox_uniapi==2.1.0

7
tests/components/airthings_ble/__init__.py
View File

@ -5,10 +5,11 @@ from unittest.mock import patch
from airthings_ble import AirthingsBluetoothDeviceData, AirthingsDevice
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth.models import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
def patch_async_setup_entry(return_value=True):
"""Patch async setup entry to return True."""
@ -48,7 +49,7 @@ WAVE_SERVICE_INFO = BluetoothServiceInfoBleak(
"cc:cc:cc:cc:cc:cc",
"cc-cc-cc-cc-cc-cc",
),
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
manufacturer_data={820: b"\xe4/\xa5\xae\t\x00"},
service_uuids=["b42e1c08-ade7-11e4-89d3-123b93f75cba"],
),
@ -68,7 +69,7 @@ UNKNOWN_SERVICE_INFO = BluetoothServiceInfoBleak(
"cc:cc:cc:cc:cc:cc",
"unknown",
),
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
manufacturer_data={},
service_uuids=[],
),

34
tests/components/bluetooth/__init__.py
View File

@ -2,6 +2,7 @@
import time
from typing import Any
from unittest.mock import patch
from bleak.backends.scanner import AdvertisementData, BLEDevice
@ -27,8 +28,27 @@ __all__ = (
"inject_bluetooth_service_info",
"patch_all_discovered_devices",
"patch_discovered_devices",
"generate_advertisement_data",
)
ADVERTISEMENT_DATA_DEFAULTS = {
"local_name": "",
"manufacturer_data": {},
"service_data": {},
"service_uuids": [],
"rssi": -127,
"platform_data": ((),),
"tx_power": -127,
}
def generate_advertisement_data(**kwargs: Any) -> AdvertisementData:
"""Generate advertisement data with defaults."""
new = kwargs.copy()
for key, value in ADVERTISEMENT_DATA_DEFAULTS.items():
new.setdefault(key, value)
return AdvertisementData(**new)
def _get_manager() -> BluetoothManager:
"""Return the bluetooth manager."""
@ -77,7 +97,7 @@ def inject_advertisement_with_time_and_source_connectable(
models.BluetoothServiceInfoBleak(
name=adv.local_name or device.name or device.address,
address=device.address,
rssi=device.rssi,
rssi=adv.rssi,
manufacturer_data=adv.manufacturer_data,
service_data=adv.service_data,
service_uuids=adv.service_uuids,
@ -94,17 +114,17 @@ def inject_bluetooth_service_info_bleak(
hass: HomeAssistant, info: models.BluetoothServiceInfoBleak
) -> None:
"""Inject an advertisement into the manager with connectable status."""
advertisement_data = AdvertisementData( # type: ignore[no-untyped-call]
advertisement_data = generate_advertisement_data(
local_name=None if info.name == "" else info.name,
manufacturer_data=info.manufacturer_data,
service_data=info.service_data,
service_uuids=info.service_uuids,
rssi=info.rssi,
)
device = BLEDevice( # type: ignore[no-untyped-call]
address=info.address,
name=info.name,
details={},
rssi=info.rssi,
)
inject_advertisement_with_time_and_source_connectable(
hass,
@ -120,17 +140,17 @@ def inject_bluetooth_service_info(
hass: HomeAssistant, info: models.BluetoothServiceInfo
) -> None:
"""Inject a BluetoothServiceInfo into the manager."""
advertisement_data = AdvertisementData( # type: ignore[no-untyped-call]
advertisement_data = generate_advertisement_data( # type: ignore[no-untyped-call]
local_name=None if info.name == "" else info.name,
manufacturer_data=info.manufacturer_data,
service_data=info.service_data,
service_uuids=info.service_uuids,
rssi=info.rssi,
)
device = BLEDevice( # type: ignore[no-untyped-call]
address=info.address,
name=info.name,
details={},
rssi=info.rssi,
)
inject_advertisement(hass, device, advertisement_data)
@ -138,7 +158,9 @@ def inject_bluetooth_service_info(
def patch_all_discovered_devices(mock_discovered: list[BLEDevice]) -> None:
"""Mock all the discovered devices from all the scanners."""
return patch.object(
_get_manager(), "async_all_discovered_devices", return_value=mock_discovered
_get_manager(),
"_async_all_discovered_addresses",
return_value={ble_device.address for ble_device in mock_discovered},
)

65
tests/components/bluetooth/test_advertisement_tracker.py
View File

@ -21,7 +21,11 @@ from homeassistant.components.bluetooth.models import BaseHaScanner
from homeassistant.core import callback
from homeassistant.util import dt as dt_util
from . import inject_advertisement_with_time_and_source
from . import (
generate_advertisement_data,
inject_advertisement_with_time_and_source,
inject_advertisement_with_time_and_source_connectable,
)
from tests.common import async_fire_time_changed
@ -33,8 +37,8 @@ async def test_advertisment_interval_shorter_than_adapter_stack_timeout(
):
"""Test we can determine the advertisement interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_device = BLEDevice("44:44:33:11:23:12", "wohand")
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@ -77,8 +81,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_connectab
):
"""Test device with a long advertisement interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_device = BLEDevice("44:44:33:11:23:18", "wohand")
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@ -124,7 +128,7 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
"""Test device with a long advertisement interval with an adapter change."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@ -179,7 +183,7 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_not_conne
"""Test device with a long advertisement interval that is not connectable not reaching the advertising interval."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False
@ -227,9 +231,11 @@ async def test_advertisment_interval_shorter_than_adapter_stack_timeout_adapter_
):
"""Test device with a short advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
switchbot_device = BLEDevice("44:44:33:11:23:5C", "wohand")
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-100,
)
switchbot_device_went_unavailable = False
@ -248,9 +254,18 @@ async def test_advertisment_interval_shorter_than_adapter_stack_timeout_adapter_
"original",
)
switchbot_adv_better_rssi = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-30,
)
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
hass, switchbot_device, switchbot_adv, start_monotonic_time + (i * 2), "new"
hass,
switchbot_device,
switchbot_adv_better_rssi,
start_monotonic_time + (i * 2),
"new",
)
switchbot_device_unavailable_cancel = async_track_unavailable(
@ -282,8 +297,10 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
"""Test device with a long advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-100,
)
switchbot_device_went_unavailable = False
@ -291,8 +308,11 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
"""Fake scanner."""
@property
def discovered_devices(self) -> list[BLEDevice]:
return []
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices."""
return {}
scanner = FakeScanner(hass, "new")
cancel_scanner = async_register_scanner(hass, scanner, False)
@ -304,21 +324,28 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
switchbot_device_went_unavailable = True
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
inject_advertisement_with_time_and_source_connectable(
hass,
switchbot_device,
switchbot_adv,
start_monotonic_time + (i * 2),
"original",
connectable=False,
)
switchbot_better_rssi_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-30,
)
for i in range(ADVERTISING_TIMES_NEEDED):
inject_advertisement_with_time_and_source(
inject_advertisement_with_time_and_source_connectable(
hass,
switchbot_device,
switchbot_adv,
switchbot_better_rssi_adv,
start_monotonic_time + (i * ONE_HOUR_SECONDS),
"new",
connectable=False,
)
switchbot_device_unavailable_cancel = async_track_unavailable(
@ -364,7 +391,7 @@ async def test_advertisment_interval_longer_increasing_than_adapter_stack_timeou
"""Test device with a increasing advertisement interval with an adapter change that is not connectable."""
start_monotonic_time = time.monotonic()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_device_went_unavailable = False

67
tests/components/bluetooth/test_diagnostics.py
View File

@ -3,12 +3,12 @@
from unittest.mock import ANY, patch
from bleak.backends.scanner import AdvertisementData, BLEDevice
from bleak.backends.scanner import BLEDevice
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth.const import DEFAULT_ADDRESS
from . import inject_advertisement
from . import generate_advertisement_data, inject_advertisement
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
@ -96,11 +96,6 @@ async def test_diagnostics(
}
},
"manager": {
"advertisement_tracker": {
"intervals": {},
"sources": {},
"timings": {},
},
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",
@ -115,13 +110,18 @@ async def test_diagnostics(
"sw_version": "BlueZ 4.63",
},
},
"advertisement_tracker": {
"intervals": {},
"sources": {},
"timings": {},
},
"connectable_history": [],
"history": [],
"non_connectable_history": [],
"scanners": [
{
"adapter": "hci0",
"discovered_devices": [
{"address": "44:44:33:11:23:45", "name": "x", "rssi": -60}
{"address": "44:44:33:11:23:45", "name": "x"}
],
"last_detection": ANY,
"name": "hci0 (00:00:00:00:00:01)",
@ -132,7 +132,7 @@ async def test_diagnostics(
{
"adapter": "hci0",
"discovered_devices": [
{"address": "44:44:33:11:23:45", "name": "x", "rssi": -60}
{"address": "44:44:33:11:23:45", "name": "x"}
],
"last_detection": ANY,
"name": "hci0 (00:00:00:00:00:01)",
@ -143,7 +143,7 @@ async def test_diagnostics(
{
"adapter": "hci1",
"discovered_devices": [
{"address": "44:44:33:11:23:45", "name": "x", "rssi": -60}
{"address": "44:44:33:11:23:45", "name": "x"}
],
"last_detection": ANY,
"name": "hci1 (00:00:00:00:00:02)",
@ -166,7 +166,7 @@ async def test_diagnostics_macos(
# error if the test is not running on linux since we won't have the correct
# deps installed when testing on MacOS.
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
@ -203,11 +203,6 @@ async def test_diagnostics_macos(
}
},
"manager": {
"advertisement_tracker": {
"intervals": {},
"sources": {"44:44:33:11:23:45": "local"},
"timings": {"44:44:33:11:23:45": [ANY]},
},
"adapters": {
"Core Bluetooth": {
"address": "00:00:00:00:00:00",
@ -215,39 +210,41 @@ async def test_diagnostics_macos(
"sw_version": ANY,
}
},
"advertisement_tracker": {
"intervals": {},
"sources": {"44:44:33:11:23:45": "local"},
"timings": {"44:44:33:11:23:45": [ANY]},
},
"connectable_history": [
{
"address": "44:44:33:11:23:45",
"advertisement": ANY,
"advertisement": [
"wohand",
{"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}},
{},
[],
-127,
-127,
[[]],
],
"connectable": True,
"manufacturer_data": ANY,
"manufacturer_data": {
"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}
},
"name": "wohand",
"rssi": 0,
"service_data": {},
"service_uuids": [],
"source": "local",
"time": ANY,
}
],
"history": [
{
"address": "44:44:33:11:23:45",
"advertisement": ANY,
"connectable": True,
"manufacturer_data": ANY,
"name": "wohand",
"rssi": 0,
"rssi": -127,
"service_data": {},
"service_uuids": [],
"source": "local",
"time": ANY,
}
],
"non_connectable_history": [],
"scanners": [
{
"adapter": "Core Bluetooth",
"discovered_devices": [
{"address": "44:44:33:11:23:45", "name": "x", "rssi": -60}
{"address": "44:44:33:11:23:45", "name": "x"}
],
"last_detection": ANY,
"name": "Core Bluetooth",

224
tests/components/bluetooth/test_init.py
View File

@ -44,6 +44,7 @@ from homeassistant.util import dt as dt_util
from . import (
_get_manager,
async_setup_with_default_adapter,
generate_advertisement_data,
inject_advertisement,
inject_advertisement_with_time_and_source_connectable,
patch_discovered_devices,
@ -334,7 +335,9 @@ async def test_discovery_match_by_service_uuid(
assert len(mock_bleak_scanner_start.mock_calls) == 1
wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement(hass, wrong_device, wrong_adv)
await hass.async_block_till_done()
@ -342,7 +345,7 @@ async def test_discovery_match_by_service_uuid(
assert len(mock_config_flow.mock_calls) == 0
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
@ -379,7 +382,9 @@ async def test_discovery_match_by_service_uuid_connectable(
assert len(mock_bleak_scanner_start.mock_calls) == 1
wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement_with_time_and_source_connectable(
hass, wrong_device, wrong_adv, time.monotonic(), "any", True
@ -389,7 +394,7 @@ async def test_discovery_match_by_service_uuid_connectable(
assert len(_domains_from_mock_config_flow(mock_config_flow)) == 0
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
@ -424,7 +429,9 @@ async def test_discovery_match_by_service_uuid_not_connectable(
assert len(mock_bleak_scanner_start.mock_calls) == 1
wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement_with_time_and_source_connectable(
hass, wrong_device, wrong_adv, time.monotonic(), "any", False
@ -434,7 +441,7 @@ async def test_discovery_match_by_service_uuid_not_connectable(
assert len(_domains_from_mock_config_flow(mock_config_flow)) == 0
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
@ -467,7 +474,9 @@ async def test_discovery_match_by_name_connectable_false(
assert len(mock_bleak_scanner_start.mock_calls) == 1
wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement_with_time_and_source_connectable(
hass, wrong_device, wrong_adv, time.monotonic(), "any", False
@ -477,7 +486,7 @@ async def test_discovery_match_by_name_connectable_false(
assert len(_domains_from_mock_config_flow(mock_config_flow)) == 0
qingping_device = BLEDevice("44:44:33:11:23:45", "Qingping Motion & Light")
qingping_adv = AdvertisementData(
qingping_adv = generate_advertisement_data(
local_name="Qingping Motion & Light",
service_data={
"0000fdcd-0000-1000-8000-00805f9b34fb": b"H\x12\xcd\xd5`4-X\x08\x04\x01\xe8\x00\x00\x0f\x01{"
@ -493,8 +502,20 @@ async def test_discovery_match_by_name_connectable_false(
mock_config_flow.reset_mock()
# Make sure it will also take a connectable device
qingping_adv_with_better_rssi = generate_advertisement_data(
local_name="Qingping Motion & Light",
service_data={
"0000fdcd-0000-1000-8000-00805f9b34fb": b"H\x12\xcd\xd5`4-X\x08\x04\x01\xe8\x00\x00\x0f\x02{"
},
rssi=-30,
)
inject_advertisement_with_time_and_source_connectable(
hass, qingping_device, qingping_adv, time.monotonic(), "any", True
hass,
qingping_device,
qingping_adv_with_better_rssi,
time.monotonic(),
"any",
True,
)
await hass.async_block_till_done()
assert _domains_from_mock_config_flow(mock_config_flow) == ["qingping"]
@ -517,7 +538,9 @@ async def test_discovery_match_by_local_name(
assert len(mock_bleak_scanner_start.mock_calls) == 1
wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement(hass, wrong_device, wrong_adv)
await hass.async_block_till_done()
@ -525,7 +548,7 @@ async def test_discovery_match_by_local_name(
assert len(mock_config_flow.mock_calls) == 0
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
@ -559,12 +582,12 @@ async def test_discovery_match_by_manufacturer_id_and_manufacturer_data_start(
assert len(mock_bleak_scanner_start.mock_calls) == 1
hkc_device = BLEDevice("44:44:33:11:23:45", "lock")
hkc_adv_no_mfr_data = AdvertisementData(
hkc_adv_no_mfr_data = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={},
)
hkc_adv = AdvertisementData(
hkc_adv = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={76: b"\x06\x02\x03\x99"},
@ -593,7 +616,7 @@ async def test_discovery_match_by_manufacturer_id_and_manufacturer_data_start(
mock_config_flow.reset_mock()
not_hkc_device = BLEDevice("44:44:33:11:23:21", "lock")
not_hkc_adv = AdvertisementData(
not_hkc_adv = generate_advertisement_data(
local_name="lock", service_uuids=[], manufacturer_data={76: b"\x02"}
)
@ -602,7 +625,7 @@ async def test_discovery_match_by_manufacturer_id_and_manufacturer_data_start(
assert len(mock_config_flow.mock_calls) == 0
not_apple_device = BLEDevice("44:44:33:11:23:23", "lock")
not_apple_adv = AdvertisementData(
not_apple_adv = generate_advertisement_data(
local_name="lock", service_uuids=[], manufacturer_data={21: b"\x02"}
)
@ -642,36 +665,38 @@ async def test_discovery_match_by_service_data_uuid_then_others(
assert len(mock_bleak_scanner_start.mock_calls) == 1
device = BLEDevice("44:44:33:11:23:45", "lock")
adv_without_service_data_uuid = AdvertisementData(
adv_without_service_data_uuid = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={},
)
adv_with_mfr_data = AdvertisementData(
adv_with_mfr_data = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={323: b"\x01\x02\x03"},
service_data={},
)
adv_with_service_data_uuid = AdvertisementData(
adv_with_service_data_uuid = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x01\x02\x03"},
)
adv_with_service_data_uuid_and_mfr_data = AdvertisementData(
adv_with_service_data_uuid_and_mfr_data = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={323: b"\x01\x02\x03"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x01\x02\x03"},
)
adv_with_service_data_uuid_and_mfr_data_and_service_uuid = AdvertisementData(
local_name="lock",
manufacturer_data={323: b"\x01\x02\x03"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x01\x02\x03"},
service_uuids=["0000fd3d-0000-1000-8000-00805f9b34fd"],
adv_with_service_data_uuid_and_mfr_data_and_service_uuid = (
generate_advertisement_data(
local_name="lock",
manufacturer_data={323: b"\x01\x02\x03"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"\x01\x02\x03"},
service_uuids=["0000fd3d-0000-1000-8000-00805f9b34fd"],
)
)
adv_with_service_uuid = AdvertisementData(
adv_with_service_uuid = generate_advertisement_data(
local_name="lock",
manufacturer_data={},
service_data={},
@ -790,18 +815,18 @@ async def test_discovery_match_by_service_data_uuid_when_format_changes(
assert len(mock_bleak_scanner_start.mock_calls) == 1
device = BLEDevice("44:44:33:11:23:45", "lock")
adv_without_service_data_uuid = AdvertisementData(
adv_without_service_data_uuid = generate_advertisement_data(
local_name="Qingping Temp RH M",
service_uuids=[],
manufacturer_data={},
)
xiaomi_format_adv = AdvertisementData(
xiaomi_format_adv = generate_advertisement_data(
local_name="Qingping Temp RH M",
service_data={
"0000fe95-0000-1000-8000-00805f9b34fb": b"0XH\x0b\x06\xa7%\x144-X\x08"
},
)
qingping_format_adv = AdvertisementData(
qingping_format_adv = generate_advertisement_data(
local_name="Qingping Temp RH M",
service_data={
"0000fdcd-0000-1000-8000-00805f9b34fb": b"\x08\x16\xa7%\x144-X\x01\x04\xdb\x00\xa6\x01\x02\x01d"
@ -871,12 +896,12 @@ async def test_discovery_match_first_by_service_uuid_and_then_manufacturer_id(
assert len(mock_bleak_scanner_start.mock_calls) == 1
device = BLEDevice("44:44:33:11:23:45", "lock")
adv_service_uuids = AdvertisementData(
adv_service_uuids = generate_advertisement_data(
local_name="lock",
service_uuids=["0000fd3d-0000-1000-8000-00805f9b34fc"],
manufacturer_data={},
)
adv_manufacturer_data = AdvertisementData(
adv_manufacturer_data = generate_advertisement_data(
local_name="lock",
service_uuids=[],
manufacturer_data={76: b"\x06\x02\x03\x99"},
@ -924,10 +949,10 @@ async def test_rediscovery(hass, mock_bleak_scanner_start, enable_bluetooth):
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={1: b"\x01"},
@ -958,8 +983,8 @@ async def test_async_discovered_device_api(
with patch(
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
), patch(
"bleak.BleakScanner.discovered_devices", # Must patch before we setup
[MagicMock(address="44:44:33:11:23:45")],
"bleak.BleakScanner.discovered_devices_and_advertisement_data", # Must patch before we setup
{"44:44:33:11:23:45": (MagicMock(address="44:44:33:11:23:45"), MagicMock())},
):
assert not bluetooth.async_discovered_service_info(hass)
assert not bluetooth.async_address_present(hass, "44:44:22:22:11:22")
@ -974,10 +999,14 @@ async def test_async_discovered_device_api(
assert not bluetooth.async_discovered_service_info(hass)
wrong_device = BLEDevice("44:44:33:11:23:42", "wrong_name")
wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
wrong_adv = generate_advertisement_data(
local_name="wrong_name", service_uuids=[]
)
inject_advertisement(hass, wrong_device, wrong_adv)
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[]
)
inject_advertisement(hass, switchbot_device, switchbot_adv)
wrong_device_went_unavailable = False
switchbot_device_went_unavailable = False
@ -1070,7 +1099,7 @@ async def test_register_callbacks(hass, mock_bleak_scanner_start, enable_bluetoo
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
@ -1080,13 +1109,13 @@ async def test_register_callbacks(hass, mock_bleak_scanner_start, enable_bluetoo
inject_advertisement(hass, switchbot_device, switchbot_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
@ -1138,7 +1167,7 @@ async def test_register_callbacks_raises_exception(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
@ -1197,7 +1226,7 @@ async def test_register_callback_by_address(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
@ -1207,13 +1236,13 @@ async def test_register_callback_by_address(
inject_advertisement(hass, switchbot_device, switchbot_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
# 3rd callback raises ValueError but is still tracked
inject_advertisement(hass, empty_device, empty_adv)
@ -1299,18 +1328,29 @@ async def test_register_callback_by_address_connectable_only(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_better_rssi = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
rssi=-30,
)
inject_advertisement_with_time_and_source_connectable(
hass, switchbot_device, switchbot_adv, time.monotonic(), "test", False
)
inject_advertisement_with_time_and_source_connectable(
hass, switchbot_device, switchbot_adv, time.monotonic(), "test", True
hass,
switchbot_device,
switchbot_adv_better_rssi,
time.monotonic(),
"test",
True,
)
cancel()
@ -1354,7 +1394,7 @@ async def test_register_callback_by_manufacturer_id(
assert len(mock_bleak_scanner_start.mock_calls) == 1
apple_device = BLEDevice("44:44:33:11:23:45", "rtx")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="rtx",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1362,7 +1402,7 @@ async def test_register_callback_by_manufacturer_id(
inject_advertisement(hass, apple_device, apple_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
@ -1409,7 +1449,7 @@ async def test_register_callback_by_connectable(
assert len(mock_bleak_scanner_start.mock_calls) == 1
apple_device = BLEDevice("44:44:33:11:23:45", "rtx")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="rtx",
manufacturer_data={7676: b"\xd8.\xad\xcd\r\x85"},
)
@ -1417,7 +1457,7 @@ async def test_register_callback_by_connectable(
inject_advertisement(hass, apple_device, apple_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
@ -1464,7 +1504,7 @@ async def test_not_filtering_wanted_apple_devices(
assert len(mock_bleak_scanner_start.mock_calls) == 1
ibeacon_device = BLEDevice("44:44:33:11:23:45", "rtx")
ibeacon_adv = AdvertisementData(
ibeacon_adv = generate_advertisement_data(
local_name="ibeacon",
manufacturer_data={76: b"\x02\x00\x00\x00"},
)
@ -1472,7 +1512,7 @@ async def test_not_filtering_wanted_apple_devices(
inject_advertisement(hass, ibeacon_device, ibeacon_adv)
homekit_device = BLEDevice("44:44:33:11:23:46", "rtx")
homekit_adv = AdvertisementData(
homekit_adv = generate_advertisement_data(
local_name="homekit",
manufacturer_data={76: b"\x06\x00\x00\x00"},
)
@ -1480,7 +1520,7 @@ async def test_not_filtering_wanted_apple_devices(
inject_advertisement(hass, homekit_device, homekit_adv)
apple_device = BLEDevice("44:44:33:11:23:47", "rtx")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="apple",
manufacturer_data={76: b"\x10\x00\x00\x00"},
)
@ -1524,7 +1564,7 @@ async def test_filtering_noisy_apple_devices(
assert len(mock_bleak_scanner_start.mock_calls) == 1
apple_device = BLEDevice("44:44:33:11:23:45", "rtx")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="noisy",
manufacturer_data={76: b"\xd8.\xad\xcd\r\x85"},
)
@ -1532,7 +1572,7 @@ async def test_filtering_noisy_apple_devices(
inject_advertisement(hass, apple_device, apple_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
@ -1574,7 +1614,7 @@ async def test_register_callback_by_address_connectable_manufacturer_id(
assert len(mock_bleak_scanner_start.mock_calls) == 1
apple_device = BLEDevice("44:44:33:11:23:45", "rtx")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="rtx",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1628,7 +1668,7 @@ async def test_register_callback_by_manufacturer_id_and_address(
assert len(mock_bleak_scanner_start.mock_calls) == 1
rtx_device = BLEDevice("44:44:33:11:23:45", "rtx")
rtx_adv = AdvertisementData(
rtx_adv = generate_advertisement_data(
local_name="rtx",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1636,7 +1676,7 @@ async def test_register_callback_by_manufacturer_id_and_address(
inject_advertisement(hass, rtx_device, rtx_adv)
yale_device = BLEDevice("44:44:33:11:23:45", "apple")
yale_adv = AdvertisementData(
yale_adv = generate_advertisement_data(
local_name="yale",
manufacturer_data={465: b"\xd8.\xad\xcd\r\x85"},
)
@ -1645,7 +1685,7 @@ async def test_register_callback_by_manufacturer_id_and_address(
await hass.async_block_till_done()
other_apple_device = BLEDevice("44:44:33:11:23:22", "apple")
other_apple_adv = AdvertisementData(
other_apple_adv = generate_advertisement_data(
local_name="apple",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1696,7 +1736,7 @@ async def test_register_callback_by_service_uuid_and_address(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_dev = BLEDevice("44:44:33:11:23:45", "switchbot")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="switchbot",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
)
@ -1704,7 +1744,7 @@ async def test_register_callback_by_service_uuid_and_address(
inject_advertisement(hass, switchbot_dev, switchbot_adv)
switchbot_missing_service_uuid_dev = BLEDevice("44:44:33:11:23:45", "switchbot")
switchbot_missing_service_uuid_adv = AdvertisementData(
switchbot_missing_service_uuid_adv = generate_advertisement_data(
local_name="switchbot",
)
@ -1714,7 +1754,7 @@ async def test_register_callback_by_service_uuid_and_address(
await hass.async_block_till_done()
service_uuid_wrong_address_dev = BLEDevice("44:44:33:11:23:22", "switchbot2")
service_uuid_wrong_address_adv = AdvertisementData(
service_uuid_wrong_address_adv = generate_advertisement_data(
local_name="switchbot2",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
)
@ -1765,7 +1805,7 @@ async def test_register_callback_by_service_data_uuid_and_address(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_dev = BLEDevice("44:44:33:11:23:45", "switchbot")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="switchbot",
service_data={"cba20d00-224d-11e6-9fb8-0002a5d5c51b": b"x"},
)
@ -1773,7 +1813,7 @@ async def test_register_callback_by_service_data_uuid_and_address(
inject_advertisement(hass, switchbot_dev, switchbot_adv)
switchbot_missing_service_uuid_dev = BLEDevice("44:44:33:11:23:45", "switchbot")
switchbot_missing_service_uuid_adv = AdvertisementData(
switchbot_missing_service_uuid_adv = generate_advertisement_data(
local_name="switchbot",
)
@ -1783,7 +1823,7 @@ async def test_register_callback_by_service_data_uuid_and_address(
await hass.async_block_till_done()
service_uuid_wrong_address_dev = BLEDevice("44:44:33:11:23:22", "switchbot2")
service_uuid_wrong_address_adv = AdvertisementData(
service_uuid_wrong_address_adv = generate_advertisement_data(
local_name="switchbot2",
service_data={"cba20d00-224d-11e6-9fb8-0002a5d5c51b": b"x"},
)
@ -1831,7 +1871,7 @@ async def test_register_callback_by_local_name(
assert len(mock_bleak_scanner_start.mock_calls) == 1
rtx_device = BLEDevice("44:44:33:11:23:45", "rtx")
rtx_adv = AdvertisementData(
rtx_adv = generate_advertisement_data(
local_name="rtx",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1839,12 +1879,12 @@ async def test_register_callback_by_local_name(
inject_advertisement(hass, rtx_device, rtx_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
rtx_device_2 = BLEDevice("44:44:33:11:23:45", "rtx")
rtx_adv_2 = AdvertisementData(
rtx_adv_2 = generate_advertisement_data(
local_name="rtx2",
manufacturer_data={21: b"\xd8.\xad\xcd\r\x85"},
)
@ -1927,7 +1967,7 @@ async def test_register_callback_by_service_data_uuid(
assert len(mock_bleak_scanner_start.mock_calls) == 1
apple_device = BLEDevice("44:44:33:11:23:45", "xiaomi")
apple_adv = AdvertisementData(
apple_adv = generate_advertisement_data(
local_name="xiaomi",
service_data={
"0000fe95-0000-1000-8000-00805f9b34fb": b"\xd8.\xad\xcd\r\x85"
@ -1937,7 +1977,7 @@ async def test_register_callback_by_service_data_uuid(
inject_advertisement(hass, apple_device, apple_adv)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
inject_advertisement(hass, empty_device, empty_adv)
await hass.async_block_till_done()
@ -1981,13 +2021,13 @@ async def test_register_callback_survives_reload(
assert len(mock_bleak_scanner_start.mock_calls) == 1
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["zba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["zba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
@ -2035,7 +2075,7 @@ async def test_process_advertisements_bail_on_good_advertisement(
while not done.done():
device = BLEDevice("aa:44:33:11:23:45", "wohand")
adv = AdvertisementData(
adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51a"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
@ -2060,13 +2100,13 @@ async def test_process_advertisements_ignore_bad_advertisement(
return_value = asyncio.Event()
device = BLEDevice("aa:44:33:11:23:45", "wohand")
adv = AdvertisementData(
adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51a"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fa": b""},
)
adv2 = AdvertisementData(
adv2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51a"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
@ -2142,20 +2182,20 @@ async def test_wrapped_instance_with_filter(
detected.append((device, advertisement_data))
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
assert _get_manager() is not None
scanner = models.HaBleakScannerWrapper(
@ -2214,20 +2254,20 @@ async def test_wrapped_instance_with_service_uuids(
detected.append((device, advertisement_data))
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
assert _get_manager() is not None
scanner = models.HaBleakScannerWrapper(
@ -2272,7 +2312,7 @@ async def test_wrapped_instance_with_broken_callbacks(
detected.append((device, advertisement_data))
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
@ -2313,20 +2353,20 @@ async def test_wrapped_instance_changes_uuids(
detected.append((device, advertisement_data))
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
assert _get_manager() is not None
scanner = models.HaBleakScannerWrapper()
@ -2368,20 +2408,20 @@ async def test_wrapped_instance_changes_filters(
detected.append((device, advertisement_data))
switchbot_device = BLEDevice("44:44:33:11:23:42", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
switchbot_adv_2 = AdvertisementData(
switchbot_adv_2 = generate_advertisement_data(
local_name="wohand",
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
manufacturer_data={89: b"\xd8.\xad\xcd\r\x84"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
)
empty_device = BLEDevice("11:22:33:44:55:62", "empty")
empty_adv = AdvertisementData(local_name="empty")
empty_adv = generate_advertisement_data(local_name="empty")
assert _get_manager() is not None
scanner = models.HaBleakScannerWrapper()
@ -2434,8 +2474,8 @@ async def test_async_ble_device_from_address(
with patch(
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
), patch(
"bleak.BleakScanner.discovered_devices", # Must patch before we setup
[MagicMock(address="44:44:33:11:23:45")],
"bleak.BleakScanner.discovered_devices_and_advertisement_data", # Must patch before we setup
{"44:44:33:11:23:45": (MagicMock(address="44:44:33:11:23:45"), MagicMock())},
):
assert not bluetooth.async_discovered_service_info(hass)
assert not bluetooth.async_address_present(hass, "44:44:22:22:11:22")
@ -2453,7 +2493,9 @@ async def test_async_ble_device_from_address(
assert not bluetooth.async_discovered_service_info(hass)
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[]
)
inject_advertisement(hass, switchbot_device, switchbot_adv)
await hass.async_block_till_done()

152
tests/components/bluetooth/test_manager.py
View File

@ -1,8 +1,9 @@
"""Tests for the Bluetooth integration manager."""
import time
from unittest.mock import AsyncMock, MagicMock, patch
from bleak.backends.scanner import AdvertisementData, BLEDevice
from bleak.backends.scanner import BLEDevice
from bluetooth_adapters import AdvertisementHistory
from homeassistant.components import bluetooth
@ -12,8 +13,10 @@ from homeassistant.components.bluetooth.manager import (
from homeassistant.setup import async_setup_component
from . import (
generate_advertisement_data,
inject_advertisement_with_source,
inject_advertisement_with_time_and_source,
inject_advertisement_with_time_and_source_connectable,
)
@ -25,7 +28,7 @@ async def test_advertisements_do_not_switch_adapters_for_no_reason(
address = "44:44:33:11:23:12"
switchbot_device_signal_100 = BLEDevice(address, "wohand_signal_100", rssi=-100)
switchbot_adv_signal_100 = AdvertisementData(
switchbot_adv_signal_100 = generate_advertisement_data(
local_name="wohand_signal_100", service_uuids=[]
)
inject_advertisement_with_source(
@ -38,7 +41,7 @@ async def test_advertisements_do_not_switch_adapters_for_no_reason(
)
switchbot_device_signal_99 = BLEDevice(address, "wohand_signal_99", rssi=-99)
switchbot_adv_signal_99 = AdvertisementData(
switchbot_adv_signal_99 = generate_advertisement_data(
local_name="wohand_signal_99", service_uuids=[]
)
inject_advertisement_with_source(
@ -51,7 +54,7 @@ async def test_advertisements_do_not_switch_adapters_for_no_reason(
)
switchbot_device_signal_98 = BLEDevice(address, "wohand_good_signal", rssi=-98)
switchbot_adv_signal_98 = AdvertisementData(
switchbot_adv_signal_98 = generate_advertisement_data(
local_name="wohand_good_signal", service_uuids=[]
)
inject_advertisement_with_source(
@ -70,9 +73,9 @@ async def test_switching_adapters_based_on_rssi(hass, enable_bluetooth):
address = "44:44:33:11:23:45"
switchbot_device_poor_signal = BLEDevice(address, "wohand_poor_signal", rssi=-100)
switchbot_adv_poor_signal = AdvertisementData(
local_name="wohand_poor_signal", service_uuids=[]
switchbot_device_poor_signal = BLEDevice(address, "wohand_poor_signal")
switchbot_adv_poor_signal = generate_advertisement_data(
local_name="wohand_poor_signal", service_uuids=[], rssi=-100
)
inject_advertisement_with_source(
hass, switchbot_device_poor_signal, switchbot_adv_poor_signal, "hci0"
@ -83,9 +86,9 @@ async def test_switching_adapters_based_on_rssi(hass, enable_bluetooth):
is switchbot_device_poor_signal
)
switchbot_device_good_signal = BLEDevice(address, "wohand_good_signal", rssi=-60)
switchbot_adv_good_signal = AdvertisementData(
local_name="wohand_good_signal", service_uuids=[]
switchbot_device_good_signal = BLEDevice(address, "wohand_good_signal")
switchbot_adv_good_signal = generate_advertisement_data(
local_name="wohand_good_signal", service_uuids=[], rssi=-60
)
inject_advertisement_with_source(
hass, switchbot_device_good_signal, switchbot_adv_good_signal, "hci1"
@ -105,11 +108,9 @@ async def test_switching_adapters_based_on_rssi(hass, enable_bluetooth):
)
# We should not switch adapters unless the signal hits the threshold
switchbot_device_similar_signal = BLEDevice(
address, "wohand_similar_signal", rssi=-62
)
switchbot_adv_similar_signal = AdvertisementData(
local_name="wohand_similar_signal", service_uuids=[]
switchbot_device_similar_signal = BLEDevice(address, "wohand_similar_signal")
switchbot_adv_similar_signal = generate_advertisement_data(
local_name="wohand_similar_signal", service_uuids=[], rssi=-62
)
inject_advertisement_with_source(
@ -126,9 +127,9 @@ async def test_switching_adapters_based_on_zero_rssi(hass, enable_bluetooth):
address = "44:44:33:11:23:45"
switchbot_device_no_rssi = BLEDevice(address, "wohand_poor_signal", rssi=0)
switchbot_adv_no_rssi = AdvertisementData(
local_name="wohand_no_rssi", service_uuids=[]
switchbot_device_no_rssi = BLEDevice(address, "wohand_poor_signal")
switchbot_adv_no_rssi = generate_advertisement_data(
local_name="wohand_no_rssi", service_uuids=[], rssi=0
)
inject_advertisement_with_source(
hass, switchbot_device_no_rssi, switchbot_adv_no_rssi, "hci0"
@ -139,9 +140,9 @@ async def test_switching_adapters_based_on_zero_rssi(hass, enable_bluetooth):
is switchbot_device_no_rssi
)
switchbot_device_good_signal = BLEDevice(address, "wohand_good_signal", rssi=-60)
switchbot_adv_good_signal = AdvertisementData(
local_name="wohand_good_signal", service_uuids=[]
switchbot_device_good_signal = BLEDevice(address, "wohand_good_signal")
switchbot_adv_good_signal = generate_advertisement_data(
local_name="wohand_good_signal", service_uuids=[], rssi=-60
)
inject_advertisement_with_source(
hass, switchbot_device_good_signal, switchbot_adv_good_signal, "hci1"
@ -161,11 +162,9 @@ async def test_switching_adapters_based_on_zero_rssi(hass, enable_bluetooth):
)
# We should not switch adapters unless the signal hits the threshold
switchbot_device_similar_signal = BLEDevice(
address, "wohand_similar_signal", rssi=-62
)
switchbot_adv_similar_signal = AdvertisementData(
local_name="wohand_similar_signal", service_uuids=[]
switchbot_device_similar_signal = BLEDevice(address, "wohand_similar_signal")
switchbot_adv_similar_signal = generate_advertisement_data(
local_name="wohand_similar_signal", service_uuids=[], rssi=-62
)
inject_advertisement_with_source(
@ -183,11 +182,9 @@ async def test_switching_adapters_based_on_stale(hass, enable_bluetooth):
address = "44:44:33:11:23:41"
start_time_monotonic = 50.0
switchbot_device_poor_signal_hci0 = BLEDevice(
address, "wohand_poor_signal_hci0", rssi=-100
)
switchbot_adv_poor_signal_hci0 = AdvertisementData(
local_name="wohand_poor_signal_hci0", service_uuids=[]
switchbot_device_poor_signal_hci0 = BLEDevice(address, "wohand_poor_signal_hci0")
switchbot_adv_poor_signal_hci0 = generate_advertisement_data(
local_name="wohand_poor_signal_hci0", service_uuids=[], rssi=-100
)
inject_advertisement_with_time_and_source(
hass,
@ -202,11 +199,9 @@ async def test_switching_adapters_based_on_stale(hass, enable_bluetooth):
is switchbot_device_poor_signal_hci0
)
switchbot_device_poor_signal_hci1 = BLEDevice(
address, "wohand_poor_signal_hci1", rssi=-99
)
switchbot_adv_poor_signal_hci1 = AdvertisementData(
local_name="wohand_poor_signal_hci1", service_uuids=[]
switchbot_device_poor_signal_hci1 = BLEDevice(address, "wohand_poor_signal_hci1")
switchbot_adv_poor_signal_hci1 = generate_advertisement_data(
local_name="wohand_poor_signal_hci1", service_uuids=[], rssi=-99
)
inject_advertisement_with_time_and_source(
hass,
@ -246,7 +241,7 @@ async def test_restore_history_from_dbus(hass, one_adapter):
ble_device = BLEDevice(address, "name")
history = {
address: AdvertisementHistory(
ble_device, AdvertisementData(local_name="name"), "hci0"
ble_device, generate_advertisement_data(local_name="name"), "hci0"
)
}
@ -258,3 +253,86 @@ async def test_restore_history_from_dbus(hass, one_adapter):
await hass.async_block_till_done()
assert bluetooth.async_ble_device_from_address(hass, address) is ble_device
async def test_switching_adapters_based_on_rssi_connectable_to_non_connectable(
hass, enable_bluetooth
):
"""Test switching adapters based on rssi from connectable to non connectable."""
address = "44:44:33:11:23:45"
now = time.monotonic()
switchbot_device_poor_signal = BLEDevice(address, "wohand_poor_signal")
switchbot_adv_poor_signal = generate_advertisement_data(
local_name="wohand_poor_signal", service_uuids=[], rssi=-100
)
inject_advertisement_with_time_and_source_connectable(
hass, switchbot_device_poor_signal, switchbot_adv_poor_signal, now, "hci0", True
)
assert (
bluetooth.async_ble_device_from_address(hass, address, False)
is switchbot_device_poor_signal
)
assert (
bluetooth.async_ble_device_from_address(hass, address, True)
is switchbot_device_poor_signal
)
switchbot_device_good_signal = BLEDevice(address, "wohand_good_signal")
switchbot_adv_good_signal = generate_advertisement_data(
local_name="wohand_good_signal", service_uuids=[], rssi=-60
)
inject_advertisement_with_time_and_source_connectable(
hass,
switchbot_device_good_signal,
switchbot_adv_good_signal,
now,
"hci1",
False,
)
assert (
bluetooth.async_ble_device_from_address(hass, address, False)
is switchbot_device_good_signal
)
assert (
bluetooth.async_ble_device_from_address(hass, address, True)
is switchbot_device_poor_signal
)
inject_advertisement_with_time_and_source_connectable(
hass,
switchbot_device_good_signal,
switchbot_adv_poor_signal,
now,
"hci0",
False,
)
assert (
bluetooth.async_ble_device_from_address(hass, address, False)
is switchbot_device_good_signal
)
assert (
bluetooth.async_ble_device_from_address(hass, address, True)
is switchbot_device_poor_signal
)
switchbot_device_excellent_signal = BLEDevice(address, "wohand_excellent_signal")
switchbot_adv_excellent_signal = generate_advertisement_data(
local_name="wohand_excellent_signal", service_uuids=[], rssi=-25
)
inject_advertisement_with_time_and_source_connectable(
hass,
switchbot_device_excellent_signal,
switchbot_adv_excellent_signal,
now,
"hci2",
False,
)
assert (
bluetooth.async_ble_device_from_address(hass, address, False)
is switchbot_device_excellent_signal
)
assert (
bluetooth.async_ble_device_from_address(hass, address, True)
is switchbot_device_poor_signal
)

82
tests/components/bluetooth/test_models.py
View File

@ -16,7 +16,12 @@ from homeassistant.components.bluetooth.models import (
HaBluetoothConnector,
)
from . import _get_manager, inject_advertisement, inject_advertisement_with_source
from . import (
_get_manager,
generate_advertisement_data,
inject_advertisement,
inject_advertisement_with_source,
)
class MockBleakClient(BleakClient):
@ -49,7 +54,7 @@ async def test_wrapped_bleak_scanner(hass, enable_bluetooth):
"""Test wrapped bleak scanner dispatches calls as expected."""
scanner = HaBleakScannerWrapper()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
inject_advertisement(hass, switchbot_device, switchbot_adv)
@ -84,7 +89,7 @@ async def test_wrapped_bleak_client_set_disconnected_callback_after_connected(
switchbot_device = BLEDevice(
"44:44:33:11:23:45", "wohand", {"path": "/org/bluez/hci0/dev_44_44_33_11_23_45"}
)
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
inject_advertisement(hass, switchbot_device, switchbot_adv)
@ -116,7 +121,7 @@ async def test_ble_device_with_proxy_client_out_of_connections(
},
rssi=-30,
)
switchbot_adv = AdvertisementData(
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
@ -153,6 +158,11 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
),
"path": "/org/bluez/hci0/dev_44_44_33_11_23_45",
},
)
switchbot_proxy_device_adv_no_connection_slot = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-30,
)
switchbot_proxy_device_has_connection_slot = BLEDevice(
@ -166,14 +176,19 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
},
rssi=-40,
)
switchbot_proxy_device_adv_has_connection_slot = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-40,
)
switchbot_device = BLEDevice(
"44:44:33:11:23:45",
"wohand",
{"path": "/org/bluez/hci0/dev_44_44_33_11_23_45"},
rssi=-100,
)
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}, rssi=-100
)
inject_advertisement_with_source(
@ -182,21 +197,28 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
inject_advertisement_with_source(
hass,
switchbot_proxy_device_has_connection_slot,
switchbot_adv,
switchbot_proxy_device_adv_has_connection_slot,
"esp32_has_connection_slot",
)
inject_advertisement_with_source(
hass,
switchbot_proxy_device_no_connection_slot,
switchbot_adv,
switchbot_proxy_device_adv_no_connection_slot,
"esp32_no_connection_slot",
)
class FakeScanner(BaseHaScanner):
@property
def discovered_devices(self) -> list[BLEDevice]:
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices."""
return [switchbot_proxy_device_has_connection_slot]
return {
switchbot_proxy_device_has_connection_slot.address: (
switchbot_proxy_device_has_connection_slot,
switchbot_proxy_device_adv_has_connection_slot,
)
}
async def async_get_device_by_address(self, address: str) -> BLEDevice | None:
"""Return a list of discovered devices."""
@ -237,7 +259,12 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
rssi=-30,
)
switchbot_proxy_device_no_connection_slot.metadata["delegate"] = 0
switchbot_proxy_device_no_connection_slot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-30,
)
switchbot_proxy_device_has_connection_slot = BLEDevice(
"44:44:33:11:23:45",
"wohand_has_connection_slot",
@ -247,9 +274,14 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
),
"path": "/org/bluez/hci0/dev_44_44_33_11_23_45",
},
rssi=-40,
)
switchbot_proxy_device_has_connection_slot.metadata["delegate"] = 0
switchbot_proxy_device_has_connection_slot_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-40,
)
switchbot_device = BLEDevice(
"44:44:33:11:23:45",
@ -258,31 +290,41 @@ async def test_ble_device_with_proxy_client_out_of_connections_uses_best_availab
rssi=-100,
)
switchbot_device.metadata["delegate"] = 0
switchbot_adv = AdvertisementData(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
switchbot_device_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-100,
)
inject_advertisement_with_source(
hass, switchbot_device, switchbot_adv, "00:00:00:00:00:01"
hass, switchbot_device, switchbot_device_adv, "00:00:00:00:00:01"
)
inject_advertisement_with_source(
hass,
switchbot_proxy_device_has_connection_slot,
switchbot_adv,
switchbot_proxy_device_has_connection_slot_adv,
"esp32_has_connection_slot",
)
inject_advertisement_with_source(
hass,
switchbot_proxy_device_no_connection_slot,
switchbot_adv,
switchbot_proxy_device_no_connection_slot_adv,
"esp32_no_connection_slot",
)
class FakeScanner(BaseHaScanner):
@property
def discovered_devices(self) -> list[BLEDevice]:
def discovered_devices_and_advertisement_data(
self,
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices."""
return [switchbot_proxy_device_has_connection_slot]
return {
switchbot_proxy_device_has_connection_slot.address: (
switchbot_proxy_device_has_connection_slot,
switchbot_proxy_device_has_connection_slot_adv,
)
}
async def async_get_device_by_address(self, address: str) -> BLEDevice | None:
"""Return a list of discovered devices."""

4
tests/components/bluetooth/test_passive_update_coordinator.py
View File

@ -127,8 +127,8 @@ async def test_unavailable_callbacks_mark_the_coordinator_unavailable(
):
"""Test that the coordinator goes unavailable when the bluetooth stack no longer sees the device."""
with patch(
"bleak.BleakScanner.discovered_devices", # Must patch before we setup
[MagicMock(address="44:44:33:11:23:45")],
"bleak.BleakScanner.discovered_devices_and_advertisement_data", # Must patch before we setup
{"44:44:33:11:23:45": (MagicMock(address="44:44:33:11:23:45"), MagicMock())},
):
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()

4
tests/components/bluetooth/test_passive_update_processor.py
View File

@ -201,8 +201,8 @@ async def test_unavailable_after_no_data(
):
"""Test that the coordinator is unavailable after no data for a while."""
with patch(
"bleak.BleakScanner.discovered_devices", # Must patch before we setup
[MagicMock(address="44:44:33:11:23:45")],
"bleak.BleakScanner.discovered_devices_and_advertisement_data", # Must patch before we setup
{"44:44:33:11:23:45": (MagicMock(address="44:44:33:11:23:45"), MagicMock())},
):
await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
await hass.async_block_till_done()

10
tests/components/bluetooth/test_scanner.py
View File

@ -4,11 +4,7 @@ import time
from unittest.mock import MagicMock, patch
from bleak import BleakError
from bleak.backends.scanner import (
AdvertisementData,
AdvertisementDataCallback,
BLEDevice,
)
from bleak.backends.scanner import AdvertisementDataCallback, BLEDevice
from dbus_fast import InvalidMessageError
import pytest
@ -22,7 +18,7 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
from homeassistant.util import dt as dt_util
from . import _get_manager, async_setup_with_one_adapter
from . import _get_manager, async_setup_with_one_adapter, generate_advertisement_data
from tests.common import async_fire_time_changed
@ -222,7 +218,7 @@ async def test_recovery_from_dbus_restart(hass, one_adapter):
):
_callback(
BLEDevice("44:44:33:11:23:42", "any_name"),
AdvertisementData(local_name="any_name"),
generate_advertisement_data(local_name="any_name"),
)
# Ensure we don't restart the scanner if we don't need to

13
tests/components/bluetooth_le_tracker/test_device_tracker.py
View File

@ -5,7 +5,7 @@ from datetime import timedelta
from unittest.mock import patch
from bleak import BleakError
from bleak.backends.scanner import AdvertisementData, BLEDevice
from bleak.backends.scanner import BLEDevice
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.bluetooth_le_tracker import device_tracker
@ -23,6 +23,7 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util, slugify
from tests.common import async_fire_time_changed
from tests.components.bluetooth import generate_advertisement_data
class MockBleakClient:
@ -89,7 +90,7 @@ async def test_preserve_new_tracked_device_name(
service_uuids=[],
source="local",
device=BLEDevice(address, None),
advertisement=AdvertisementData(local_name="empty"),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
@ -114,7 +115,7 @@ async def test_preserve_new_tracked_device_name(
service_uuids=[],
source="local",
device=BLEDevice(address, None),
advertisement=AdvertisementData(local_name="empty"),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
@ -158,7 +159,7 @@ async def test_tracking_battery_times_out(
service_uuids=[],
source="local",
device=BLEDevice(address, None),
advertisement=AdvertisementData(local_name="empty"),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
@ -224,7 +225,7 @@ async def test_tracking_battery_fails(hass, mock_bluetooth, mock_device_tracker_
service_uuids=[],
source="local",
device=BLEDevice(address, None),
advertisement=AdvertisementData(local_name="empty"),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
@ -292,7 +293,7 @@ async def test_tracking_battery_successful(
service_uuids=[],
source="local",
device=BLEDevice(address, None),
advertisement=AdvertisementData(local_name="empty"),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=True,
)

17
tests/components/bthome/__init__.py
View File

@ -1,10 +1,11 @@
"""Tests for the BTHome integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
TEMP_HUMI_SERVICE_INFO = BluetoothServiceInfoBleak(
name="ATC 8D18B2",
address="A4:C1:38:8D:18:B2",
@ -16,7 +17,7 @@ TEMP_HUMI_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -32,7 +33,7 @@ TEMP_HUMI_ENCRYPTED_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000181e-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -48,7 +49,7 @@ PRST_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="prst"),
advertisement=generate_advertisement_data(local_name="prst"),
time=0,
connectable=False,
)
@ -64,7 +65,7 @@ INVALID_PAYLOAD = BluetoothServiceInfoBleak(
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -78,7 +79,7 @@ NOT_BTHOME_SERVICE_INFO = BluetoothServiceInfoBleak(
service_data={},
service_uuids=[],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -97,7 +98,7 @@ def make_advertisement(address: str, payload: bytes) -> BluetoothServiceInfoBlea
},
service_uuids=["0000181c-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Test Device"),
advertisement=generate_advertisement_data(local_name="Test Device"),
time=0,
connectable=False,
)
@ -118,7 +119,7 @@ def make_encrypted_advertisement(
},
service_uuids=["0000181e-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="ATC 8F80A5"),
advertisement=generate_advertisement_data(local_name="ATC 8F80A5"),
time=0,
connectable=False,
)

5
tests/components/fjaraskupan/__init__.py
View File

@ -2,10 +2,11 @@
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
COOKER_SERVICE_INFO = BluetoothServiceInfoBleak(
name="COOKERHOOD_FJAR",
address="AA:BB:CC:DD:EE:FF",
@ -15,7 +16,7 @@ COOKER_SERVICE_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="COOKERHOOD_FJAR"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)

5
tests/components/keymitt_ble/__init__.py
View File

@ -2,11 +2,12 @@
from unittest.mock import patch
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.const import CONF_ADDRESS
from tests.components.bluetooth import generate_advertisement_data
DOMAIN = "keymitt_ble"
ENTRY_CONFIG = {
@ -38,7 +39,7 @@ SERVICE_INFO = BluetoothServiceInfoBleak(
service_data={},
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="mibp",
manufacturer_data={},
service_uuids=["0000abcd-0000-1000-8000-00805f9b34fb"],

9
tests/components/led_ble/__init__.py
View File

@ -1,9 +1,10 @@
"""Tests for the LED BLE Bluetooth integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="Triones:F30200000152C",
address="AA:BB:CC:DD:EE:FF",
@ -13,7 +14,7 @@ LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Triones:F30200000152C"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
@ -27,7 +28,7 @@ UNSUPPORTED_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="LEDnetWFF30200000152C"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
@ -45,7 +46,7 @@ NOT_LED_BLE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)

6
tests/components/melnor/conftest.py
View File

@ -5,7 +5,6 @@ from __future__ import annotations
from unittest.mock import AsyncMock, patch
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from melnor_bluetooth.device import Device
from homeassistant.components.bluetooth.models import BluetoothServiceInfoBleak
@ -14,6 +13,7 @@ from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.bluetooth import generate_advertisement_data
FAKE_ADDRESS_1 = "FAKE-ADDRESS-1"
FAKE_ADDRESS_2 = "FAKE-ADDRESS-2"
@ -30,7 +30,7 @@ FAKE_SERVICE_INFO_1 = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(FAKE_ADDRESS_1, None),
advertisement=AdvertisementData(local_name=""),
advertisement=generate_advertisement_data(local_name=""),
time=0,
connectable=True,
)
@ -46,7 +46,7 @@ FAKE_SERVICE_INFO_2 = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(FAKE_ADDRESS_2, None),
advertisement=AdvertisementData(local_name=""),
advertisement=generate_advertisement_data(local_name=""),
time=0,
connectable=True,
)

16
tests/components/switchbot/__init__.py
View File

@ -2,13 +2,13 @@
from unittest.mock import patch
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.const import CONF_ADDRESS
from homeassistant.core import HomeAssistant
from tests.common import MockConfigEntry
from tests.components.bluetooth import generate_advertisement_data
DOMAIN = "switchbot"
@ -62,7 +62,7 @@ WOHAND_SERVICE_INFO = BluetoothServiceInfoBleak(
address="AA:BB:CC:DD:EE:FF",
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="WoHand",
manufacturer_data={89: b"\xfd`0U\x92W"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x90\xd9"},
@ -82,7 +82,7 @@ WOHAND_SERVICE_INFO_NOT_CONNECTABLE = BluetoothServiceInfoBleak(
address="aa:bb:cc:dd:ee:ff",
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="WoHand",
manufacturer_data={89: b"\xfd`0U\x92W"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x90\xd9"},
@ -102,7 +102,7 @@ WOHAND_ENCRYPTED_SERVICE_INFO = BluetoothServiceInfoBleak(
address="798A8547-2A3D-C609-55FF-73FA824B923B",
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="WoHand",
manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"\xc8\x10\xcf"},
@ -122,7 +122,7 @@ WOHAND_SERVICE_ALT_ADDRESS_INFO = BluetoothServiceInfoBleak(
address="cc:cc:cc:cc:cc:cc",
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="WoHand",
manufacturer_data={89: b"\xfd`0U\x92W"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x90\xd9"},
@ -140,7 +140,7 @@ WOCURTAIN_SERVICE_INFO = BluetoothServiceInfoBleak(
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
local_name="WoCurtain",
manufacturer_data={89: b"\xc1\xc7'}U\xab"},
service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"c\xd0Y\x00\x11\x04"},
@ -159,7 +159,7 @@ WOSENSORTH_SERVICE_INFO = BluetoothServiceInfoBleak(
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"T\x00d\x00\x96\xac"},
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
manufacturer_data={2409: b"\xda,\x1e\xb1\x86Au\x03\x00\x96\xac"},
service_data={"0000fd3d-0000-1000-8000-00805f9b34fb": b"T\x00d\x00\x96\xac"},
),
@ -176,7 +176,7 @@ NOT_SWITCHBOT_INFO = BluetoothServiceInfoBleak(
service_data={},
rssi=-60,
source="local",
advertisement=AdvertisementData(
advertisement=generate_advertisement_data(
manufacturer_data={},
service_data={},
),

17
tests/components/xiaomi_ble/__init__.py
View File

@ -1,10 +1,11 @@
"""Tests for the SensorPush integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
NOT_SENSOR_PUSH_SERVICE_INFO = BluetoothServiceInfoBleak(
name="Not it",
address="00:00:00:00:00:00",
@ -14,7 +15,7 @@ NOT_SENSOR_PUSH_SERVICE_INFO = BluetoothServiceInfoBleak(
service_data={},
service_uuids=[],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -30,7 +31,7 @@ LYWSDCGQ_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -46,7 +47,7 @@ MMC_T201_1_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -62,7 +63,7 @@ JTYJGD03MI_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -78,7 +79,7 @@ YLKG07YL_SERVICE_INFO = BluetoothServiceInfoBleak(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -94,7 +95,7 @@ MISSING_PAYLOAD_ENCRYPTED = BluetoothServiceInfoBleak(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Not it"),
advertisement=generate_advertisement_data(local_name="Not it"),
time=0,
connectable=False,
)
@ -115,7 +116,7 @@ def make_advertisement(
},
service_uuids=["0000fe95-0000-1000-8000-00805f9b34fb"],
source="local",
advertisement=AdvertisementData(local_name="Test Device"),
advertisement=generate_advertisement_data(local_name="Test Device"),
time=0,
connectable=connectable,
)

11
tests/components/yalexs_ble/__init__.py
View File

@ -1,9 +1,10 @@
"""Tests for the Yale Access Bluetooth integration."""
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from tests.components.bluetooth import generate_advertisement_data
YALE_ACCESS_LOCK_DISCOVERY_INFO = BluetoothServiceInfoBleak(
name="M1012LU",
address="AA:BB:CC:DD:EE:FF",
@ -16,7 +17,7 @@ YALE_ACCESS_LOCK_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="M1012LU"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
@ -34,7 +35,7 @@ LOCK_DISCOVERY_INFO_UUID_ADDRESS = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="M1012LU"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
@ -51,7 +52,7 @@ OLD_FIRMWARE_LOCK_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)
@ -69,7 +70,7 @@ NOT_YALE_DISCOVERY_INFO = BluetoothServiceInfoBleak(
service_data={},
source="local",
device=BLEDevice(address="AA:BB:CC:DD:EE:FF", name="Aug"),
advertisement=AdvertisementData(),
advertisement=generate_advertisement_data(),
time=0,
connectable=True,
)