685 lines
26 KiB
Python
685 lines
26 KiB
Python
"""Tests for the Bluetooth integration."""
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from bleak import BleakError
|
|
from bleak.backends.scanner import AdvertisementData, BLEDevice
|
|
|
|
from homeassistant.components import bluetooth
|
|
from homeassistant.components.bluetooth import (
|
|
BluetoothChange,
|
|
BluetoothServiceInfo,
|
|
models,
|
|
)
|
|
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
|
|
from homeassistant.setup import async_setup_component
|
|
|
|
|
|
async def test_setup_and_stop(hass, mock_bleak_scanner_start):
|
|
"""Test we and setup and stop the scanner."""
|
|
mock_bt = [
|
|
{"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
await hass.async_block_till_done()
|
|
assert len(mock_bleak_scanner_start.mock_calls) == 1
|
|
|
|
|
|
async def test_setup_and_stop_no_bluetooth(hass, caplog):
|
|
"""Test we fail gracefully when bluetooth is not available."""
|
|
mock_bt = [
|
|
{"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.HaBleakScanner", side_effect=BleakError
|
|
) as mock_ha_bleak_scanner, patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(
|
|
hass.config_entries.flow, "async_init"
|
|
):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
await hass.async_block_till_done()
|
|
assert len(mock_ha_bleak_scanner.mock_calls) == 1
|
|
assert "Could not create bluetooth scanner" in caplog.text
|
|
|
|
|
|
async def test_calling_async_discovered_devices_no_bluetooth(hass, caplog):
|
|
"""Test we fail gracefully when asking for discovered devices and there is no blueooth."""
|
|
mock_bt = []
|
|
with patch(
|
|
"homeassistant.components.bluetooth.HaBleakScanner", side_effect=BleakError
|
|
) as mock_ha_bleak_scanner, patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(
|
|
hass.config_entries.flow, "async_init"
|
|
):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
|
|
await hass.async_block_till_done()
|
|
assert len(mock_ha_bleak_scanner.mock_calls) == 1
|
|
assert "Could not create bluetooth scanner" in caplog.text
|
|
assert not bluetooth.async_discovered_service_info(hass)
|
|
assert not bluetooth.async_address_present(hass, "aa:bb:bb:dd:ee:ff")
|
|
|
|
|
|
async def test_discovery_match_by_service_uuid(hass, mock_bleak_scanner_start):
|
|
"""Test bluetooth discovery match by service_uuid."""
|
|
mock_bt = [
|
|
{"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
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=[])
|
|
|
|
models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 0
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
|
|
|
|
|
|
async def test_discovery_match_by_local_name(hass, mock_bleak_scanner_start):
|
|
"""Test bluetooth discovery match by local_name."""
|
|
mock_bt = [{"domain": "switchbot", "local_name": "wohand"}]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
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=[])
|
|
|
|
models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 0
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
|
|
|
|
|
|
async def test_discovery_match_by_manufacturer_id_and_first_byte(
|
|
hass, mock_bleak_scanner_start
|
|
):
|
|
"""Test bluetooth discovery match by manufacturer_id and manufacturer_data_first_byte."""
|
|
mock_bt = [
|
|
{
|
|
"domain": "homekit_controller",
|
|
"manufacturer_id": 76,
|
|
"manufacturer_data_first_byte": 0x06,
|
|
}
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_bleak_scanner_start.mock_calls) == 1
|
|
|
|
hkc_device = BLEDevice("44:44:33:11:23:45", "lock")
|
|
hkc_adv = AdvertisementData(
|
|
local_name="lock", service_uuids=[], manufacturer_data={76: b"\x06"}
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(hkc_device, hkc_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "homekit_controller"
|
|
mock_config_flow.reset_mock()
|
|
|
|
# 2nd discovery should not generate another flow
|
|
models.HA_BLEAK_SCANNER._callback(hkc_device, hkc_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 0
|
|
|
|
mock_config_flow.reset_mock()
|
|
not_hkc_device = BLEDevice("44:44:33:11:23:21", "lock")
|
|
not_hkc_adv = AdvertisementData(
|
|
local_name="lock", service_uuids=[], manufacturer_data={76: b"\x02"}
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(not_hkc_device, not_hkc_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 0
|
|
not_apple_device = BLEDevice("44:44:33:11:23:23", "lock")
|
|
not_apple_adv = AdvertisementData(
|
|
local_name="lock", service_uuids=[], manufacturer_data={21: b"\x02"}
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(not_apple_device, not_apple_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_config_flow.mock_calls) == 0
|
|
|
|
|
|
async def test_async_discovered_device_api(hass, mock_bleak_scanner_start):
|
|
"""Test the async_discovered_device_api."""
|
|
mock_bt = []
|
|
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")],
|
|
):
|
|
assert not bluetooth.async_discovered_service_info(hass)
|
|
assert not bluetooth.async_address_present(hass, "44:44:22:22:11:22")
|
|
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(mock_bleak_scanner_start.mock_calls) == 1
|
|
|
|
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=[])
|
|
models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
service_infos = bluetooth.async_discovered_service_info(hass)
|
|
assert len(service_infos) == 1
|
|
# wrong_name should not appear because bleak no longer sees it
|
|
assert service_infos[0].name == "wohand"
|
|
|
|
assert bluetooth.async_address_present(hass, "44:44:33:11:23:42") is False
|
|
assert bluetooth.async_address_present(hass, "44:44:33:11:23:45") is True
|
|
|
|
|
|
async def test_register_callbacks(hass, mock_bleak_scanner_start):
|
|
"""Test registering a callback."""
|
|
mock_bt = []
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo, change: BluetoothChange
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
if len(callbacks) >= 3:
|
|
raise ValueError
|
|
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"service_uuids": {"cba20d00-224d-11e6-9fb8-0002a5d5c51b"}},
|
|
)
|
|
|
|
assert len(mock_bleak_scanner_start.mock_calls) == 1
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
|
|
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
models.HA_BLEAK_SCANNER._callback(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")
|
|
|
|
# 3rd callback raises ValueError but is still tracked
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
cancel()
|
|
|
|
# 4th callback should not be tracked since we canceled
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(callbacks) == 3
|
|
|
|
service_info: BluetoothServiceInfo = callbacks[0][0]
|
|
assert service_info.name == "wohand"
|
|
assert service_info.manufacturer == "Nordic Semiconductor ASA"
|
|
assert service_info.manufacturer_id == 89
|
|
|
|
service_info: BluetoothServiceInfo = callbacks[1][0]
|
|
assert service_info.name == "empty"
|
|
assert service_info.manufacturer is None
|
|
assert service_info.manufacturer_id is None
|
|
|
|
service_info: BluetoothServiceInfo = callbacks[2][0]
|
|
assert service_info.name == "empty"
|
|
assert service_info.manufacturer is None
|
|
assert service_info.manufacturer_id is None
|
|
|
|
|
|
async def test_register_callback_by_address(hass, mock_bleak_scanner_start):
|
|
"""Test registering a callback by address."""
|
|
mock_bt = []
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo, change: BluetoothChange
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
if len(callbacks) >= 3:
|
|
raise ValueError
|
|
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45"},
|
|
)
|
|
|
|
assert len(mock_bleak_scanner_start.mock_calls) == 1
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
|
|
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
models.HA_BLEAK_SCANNER._callback(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")
|
|
|
|
# 3rd callback raises ValueError but is still tracked
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
cancel()
|
|
|
|
# 4th callback should not be tracked since we canceled
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
# Now register again with a callback that fails to
|
|
# make sure we do not perm fail
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45"},
|
|
)
|
|
cancel()
|
|
|
|
# Now register again, since the 3rd callback
|
|
# should fail but we should still record it
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45"},
|
|
)
|
|
cancel()
|
|
|
|
assert len(callbacks) == 3
|
|
|
|
for idx in range(3):
|
|
service_info: BluetoothServiceInfo = callbacks[idx][0]
|
|
assert service_info.name == "wohand"
|
|
assert service_info.manufacturer == "Nordic Semiconductor ASA"
|
|
assert service_info.manufacturer_id == 89
|
|
|
|
|
|
async def test_wrapped_instance_with_filter(hass, mock_bleak_scanner_start):
|
|
"""Test consumers can use the wrapped instance with a filter as if it was normal BleakScanner."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
detected = []
|
|
|
|
def _device_detected(
|
|
device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Handle a detected device."""
|
|
detected.append((device, advertisement_data))
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper(
|
|
filters={"UUIDs": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]}
|
|
)
|
|
scanner.register_detection_callback(_device_detected)
|
|
|
|
mock_discovered = [MagicMock()]
|
|
type(models.HA_BLEAK_SCANNER).discovered_devices = mock_discovered
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
discovered = await scanner.discover(timeout=0)
|
|
assert len(discovered) == 1
|
|
assert discovered == mock_discovered
|
|
assert len(detected) == 1
|
|
|
|
scanner.register_detection_callback(_device_detected)
|
|
# We should get a reply from the history when we register again
|
|
assert len(detected) == 2
|
|
scanner.register_detection_callback(_device_detected)
|
|
# We should get a reply from the history when we register again
|
|
assert len(detected) == 3
|
|
|
|
type(models.HA_BLEAK_SCANNER).discovered_devices = []
|
|
discovered = await scanner.discover(timeout=0)
|
|
assert len(discovered) == 0
|
|
assert discovered == []
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
assert len(detected) == 4
|
|
|
|
# The filter we created in the wrapped scanner with should be respected
|
|
# and we should not get another callback
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
assert len(detected) == 4
|
|
|
|
|
|
async def test_wrapped_instance_with_service_uuids(hass, mock_bleak_scanner_start):
|
|
"""Test consumers can use the wrapped instance with a service_uuids list as if it was normal BleakScanner."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
detected = []
|
|
|
|
def _device_detected(
|
|
device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Handle a detected device."""
|
|
detected.append((device, advertisement_data))
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper(
|
|
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
|
|
)
|
|
scanner.register_detection_callback(_device_detected)
|
|
|
|
type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
|
|
for _ in range(2):
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(detected) == 2
|
|
|
|
# The UUIDs list we created in the wrapped scanner with should be respected
|
|
# and we should not get another callback
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
assert len(detected) == 2
|
|
|
|
|
|
async def test_wrapped_instance_with_broken_callbacks(hass, mock_bleak_scanner_start):
|
|
"""Test broken callbacks do not cause the scanner to fail."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
detected = []
|
|
|
|
def _device_detected(
|
|
device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Handle a detected device."""
|
|
if detected:
|
|
raise ValueError
|
|
detected.append((device, advertisement_data))
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper(
|
|
service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
|
|
)
|
|
scanner.register_detection_callback(_device_detected)
|
|
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
assert len(detected) == 1
|
|
|
|
|
|
async def test_wrapped_instance_changes_uuids(hass, mock_bleak_scanner_start):
|
|
"""Test consumers can use the wrapped instance can change the uuids later."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
detected = []
|
|
|
|
def _device_detected(
|
|
device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Handle a detected device."""
|
|
detected.append((device, advertisement_data))
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
empty_device = BLEDevice("11:22:33:44:55:66", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper()
|
|
scanner.set_scanning_filter(service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"])
|
|
scanner.register_detection_callback(_device_detected)
|
|
|
|
type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
|
|
for _ in range(2):
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(detected) == 2
|
|
|
|
# The UUIDs list we created in the wrapped scanner with should be respected
|
|
# and we should not get another callback
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
assert len(detected) == 2
|
|
|
|
|
|
async def test_wrapped_instance_changes_filters(hass, mock_bleak_scanner_start):
|
|
"""Test consumers can use the wrapped instance can change the filter later."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
detected = []
|
|
|
|
def _device_detected(
|
|
device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Handle a detected device."""
|
|
detected.append((device, advertisement_data))
|
|
|
|
switchbot_device = BLEDevice("44:44:33:11:23:42", "wohand")
|
|
switchbot_adv = AdvertisementData(
|
|
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"},
|
|
)
|
|
empty_device = BLEDevice("11:22:33:44:55:62", "empty")
|
|
empty_adv = AdvertisementData(local_name="empty")
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper()
|
|
scanner.set_scanning_filter(
|
|
filters={"UUIDs": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]}
|
|
)
|
|
scanner.register_detection_callback(_device_detected)
|
|
|
|
type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
|
|
for _ in range(2):
|
|
models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
|
|
await hass.async_block_till_done()
|
|
|
|
assert len(detected) == 2
|
|
|
|
# The UUIDs list we created in the wrapped scanner with should be respected
|
|
# and we should not get another callback
|
|
models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
|
|
assert len(detected) == 2
|
|
|
|
|
|
async def test_wrapped_instance_unsupported_filter(
|
|
hass, mock_bleak_scanner_start, caplog
|
|
):
|
|
"""Test we want when their filter is ineffective."""
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
|
|
), patch.object(hass.config_entries.flow, "async_init"):
|
|
assert await async_setup_component(
|
|
hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
|
|
)
|
|
hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
|
|
await hass.async_block_till_done()
|
|
|
|
assert models.HA_BLEAK_SCANNER is not None
|
|
scanner = models.HaBleakScannerWrapper()
|
|
scanner.set_scanning_filter(
|
|
filters={"unsupported": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]}
|
|
)
|
|
assert "Only UUIDs filters are supported" in caplog.text
|