mirror of
https://github.com/home-assistant/core.git
synced 2025-11-11 12:00:52 +00:00
2043 lines
66 KiB
Python
2043 lines
66 KiB
Python
"""Tests for the Bluetooth integration manager."""
|
|
|
|
from datetime import timedelta
|
|
import time
|
|
from typing import Any
|
|
from unittest.mock import patch
|
|
|
|
from bleak.backends.scanner import AdvertisementData, BLEDevice
|
|
from bluetooth_adapters import AdvertisementHistory
|
|
from freezegun import freeze_time
|
|
from habluetooth import BluetoothScanningMode, HaScanner
|
|
|
|
# pylint: disable-next=no-name-in-module
|
|
from habluetooth.advertisement_tracker import TRACKER_BUFFERING_WOBBLE_SECONDS
|
|
import pytest
|
|
|
|
from homeassistant import config_entries
|
|
from homeassistant.components import bluetooth
|
|
from homeassistant.components.bluetooth import (
|
|
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
|
|
MONOTONIC_TIME,
|
|
BaseHaRemoteScanner,
|
|
BluetoothChange,
|
|
BluetoothServiceInfo,
|
|
BluetoothServiceInfoBleak,
|
|
HaBluetoothConnector,
|
|
async_ble_device_from_address,
|
|
async_get_fallback_availability_interval,
|
|
async_get_learned_advertising_interval,
|
|
async_scanner_count,
|
|
async_set_fallback_availability_interval,
|
|
async_track_unavailable,
|
|
storage,
|
|
)
|
|
from homeassistant.components.bluetooth.const import (
|
|
SOURCE_LOCAL,
|
|
UNAVAILABLE_TRACK_SECONDS,
|
|
)
|
|
from homeassistant.components.bluetooth.manager import HomeAssistantBluetoothManager
|
|
from homeassistant.core import HomeAssistant, callback
|
|
from homeassistant.helpers import issue_registry as ir
|
|
from homeassistant.helpers.discovery_flow import DiscoveryKey
|
|
from homeassistant.setup import async_setup_component
|
|
from homeassistant.util import dt as dt_util
|
|
from homeassistant.util.dt import utcnow
|
|
from homeassistant.util.json import json_loads
|
|
|
|
from . import (
|
|
HCI0_SOURCE_ADDRESS,
|
|
HCI1_SOURCE_ADDRESS,
|
|
FakeRemoteScanner,
|
|
FakeScanner,
|
|
MockBleakClient,
|
|
_get_manager,
|
|
generate_advertisement_data,
|
|
generate_ble_device,
|
|
inject_advertisement_with_source,
|
|
inject_advertisement_with_time_and_source,
|
|
inject_advertisement_with_time_and_source_connectable,
|
|
patch_bluetooth_time,
|
|
)
|
|
|
|
from tests.common import (
|
|
MockConfigEntry,
|
|
MockModule,
|
|
async_call_logger_set_level,
|
|
async_fire_time_changed,
|
|
async_load_fixture,
|
|
mock_integration,
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_advertisements_do_not_switch_adapters_for_no_reason(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test we only switch adapters when needed."""
|
|
|
|
address = "44:44:33:11:23:12"
|
|
|
|
switchbot_device_signal_100 = generate_ble_device(address, "wohand_signal_100")
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100", service_uuids=[], rssi=-100
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_100, switchbot_adv_signal_100, HCI0_SOURCE_ADDRESS
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_signal_100
|
|
)
|
|
|
|
switchbot_device_signal_99 = generate_ble_device(address, "wohand_signal_99")
|
|
switchbot_adv_signal_99 = generate_advertisement_data(
|
|
local_name="wohand_signal_99", service_uuids=[], rssi=-99
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_99, switchbot_adv_signal_99, HCI0_SOURCE_ADDRESS
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_signal_99
|
|
)
|
|
|
|
switchbot_device_signal_98 = generate_ble_device(address, "wohand_good_signal")
|
|
switchbot_adv_signal_98 = generate_advertisement_data(
|
|
local_name="wohand_good_signal", service_uuids=[], rssi=-98
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_98, switchbot_adv_signal_98, HCI1_SOURCE_ADDRESS
|
|
)
|
|
|
|
# should not switch to hci1
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_signal_99
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_based_on_rssi(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test switching adapters based on rssi."""
|
|
|
|
address = "44:44:33:11:23:45"
|
|
|
|
switchbot_device_poor_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal
|
|
)
|
|
|
|
switchbot_device_good_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
inject_advertisement_with_source(
|
|
hass,
|
|
switchbot_device_good_signal,
|
|
switchbot_adv_poor_signal,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
# We should not switch adapters unless the signal hits the threshold
|
|
switchbot_device_similar_signal = generate_ble_device(
|
|
address, "wohand_similar_signal"
|
|
)
|
|
switchbot_adv_similar_signal = generate_advertisement_data(
|
|
local_name="wohand_similar_signal", service_uuids=[], rssi=-62
|
|
)
|
|
|
|
inject_advertisement_with_source(
|
|
hass,
|
|
switchbot_device_similar_signal,
|
|
switchbot_adv_similar_signal,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_based_on_zero_rssi(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test switching adapters based on zero rssi."""
|
|
|
|
address = "44:44:33:11:23:45"
|
|
|
|
switchbot_device_no_rssi = generate_ble_device(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_SOURCE_ADDRESS
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_no_rssi
|
|
)
|
|
|
|
switchbot_device_good_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_good_signal, switchbot_adv_no_rssi, HCI0_SOURCE_ADDRESS
|
|
)
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
# We should not switch adapters unless the signal hits the threshold
|
|
switchbot_device_similar_signal = generate_ble_device(
|
|
address, "wohand_similar_signal"
|
|
)
|
|
switchbot_adv_similar_signal = generate_advertisement_data(
|
|
local_name="wohand_similar_signal", service_uuids=[], rssi=-62
|
|
)
|
|
|
|
inject_advertisement_with_source(
|
|
hass,
|
|
switchbot_device_similar_signal,
|
|
switchbot_adv_similar_signal,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_based_on_stale(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test switching adapters based on the previous advertisement being stale."""
|
|
|
|
address = "44:44:33:11:23:41"
|
|
start_time_monotonic = 50.0
|
|
|
|
switchbot_device_poor_signal_hci0 = generate_ble_device(
|
|
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,
|
|
switchbot_device_poor_signal_hci0,
|
|
switchbot_adv_poor_signal_hci0,
|
|
start_time_monotonic,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci0
|
|
)
|
|
|
|
switchbot_device_poor_signal_hci1 = generate_ble_device(
|
|
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,
|
|
switchbot_device_poor_signal_hci1,
|
|
switchbot_adv_poor_signal_hci1,
|
|
start_time_monotonic,
|
|
HCI1_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# Should not switch adapters until the advertisement is stale
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci0
|
|
)
|
|
|
|
# Should switch to hci1 since the previous advertisement is stale
|
|
# even though the signal is poor because the device is now
|
|
# likely unreachable via hci0
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device_poor_signal_hci1,
|
|
switchbot_adv_poor_signal_hci1,
|
|
start_time_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1,
|
|
"hci1",
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci1
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_based_on_stale_with_discovered_interval(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test switching with discovered interval."""
|
|
|
|
address = "44:44:33:11:23:41"
|
|
start_time_monotonic = 50.0
|
|
|
|
switchbot_device_poor_signal_hci0 = generate_ble_device(
|
|
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,
|
|
switchbot_device_poor_signal_hci0,
|
|
switchbot_adv_poor_signal_hci0,
|
|
start_time_monotonic,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci0
|
|
)
|
|
|
|
bluetooth.async_set_fallback_availability_interval(hass, address, 10)
|
|
|
|
switchbot_device_poor_signal_hci1 = generate_ble_device(
|
|
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,
|
|
switchbot_device_poor_signal_hci1,
|
|
switchbot_adv_poor_signal_hci1,
|
|
start_time_monotonic,
|
|
HCI1_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# Should not switch adapters until the advertisement is stale
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci0
|
|
)
|
|
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device_poor_signal_hci1,
|
|
switchbot_adv_poor_signal_hci1,
|
|
start_time_monotonic + 10 + 1,
|
|
HCI1_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# Should not switch yet since we are not within the
|
|
# wobble period
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci0
|
|
)
|
|
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device_poor_signal_hci1,
|
|
switchbot_adv_poor_signal_hci1,
|
|
start_time_monotonic + 10 + TRACKER_BUFFERING_WOBBLE_SECONDS + 1,
|
|
HCI1_SOURCE_ADDRESS,
|
|
)
|
|
# Should switch to hci1 since the previous advertisement is stale
|
|
# even though the signal is poor because the device is now
|
|
# likely unreachable via hci0
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal_hci1
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_restore_history_from_dbus(
|
|
hass: HomeAssistant, disable_new_discovery_flows
|
|
) -> None:
|
|
"""Test we can restore history from dbus."""
|
|
address = "AA:BB:CC:CC:CC:FF"
|
|
|
|
ble_device = generate_ble_device(address, "name")
|
|
history = {
|
|
address: AdvertisementHistory(
|
|
ble_device,
|
|
generate_advertisement_data(local_name="name"),
|
|
"hci0",
|
|
)
|
|
}
|
|
|
|
with patch(
|
|
"bluetooth_adapters.systems.linux.LinuxAdapters.history",
|
|
history,
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert bluetooth.async_ble_device_from_address(hass, address) is ble_device
|
|
info = bluetooth.async_last_service_info(hass, address, False)
|
|
assert info.source == "00:00:00:00:00:01"
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_restore_history_from_dbus_and_remote_adapters(
|
|
hass: HomeAssistant,
|
|
hass_storage: dict[str, Any],
|
|
disable_new_discovery_flows,
|
|
) -> None:
|
|
"""Test we can restore history from dbus along with remote adapters."""
|
|
address = "AA:BB:CC:CC:CC:FF"
|
|
|
|
data = hass_storage[storage.REMOTE_SCANNER_STORAGE_KEY] = json_loads(
|
|
await async_load_fixture(hass, "bluetooth.remote_scanners", bluetooth.DOMAIN)
|
|
)
|
|
now = time.time()
|
|
timestamps = data["data"]["atom-bluetooth-proxy-ceaac4"][
|
|
"discovered_device_timestamps"
|
|
]
|
|
for address in timestamps:
|
|
timestamps[address] = now
|
|
|
|
ble_device = generate_ble_device(address, "name")
|
|
history = {
|
|
address: AdvertisementHistory(
|
|
ble_device,
|
|
generate_advertisement_data(local_name="name"),
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
}
|
|
|
|
with patch(
|
|
"bluetooth_adapters.systems.linux.LinuxAdapters.history",
|
|
history,
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert bluetooth.async_ble_device_from_address(hass, address) is not None
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, "EB:0B:36:35:6F:A4") is not None
|
|
)
|
|
assert disable_new_discovery_flows.call_count > 1
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_restore_history_from_dbus_and_corrupted_remote_adapters(
|
|
hass: HomeAssistant,
|
|
hass_storage: dict[str, Any],
|
|
disable_new_discovery_flows,
|
|
) -> None:
|
|
"""Test we can restore history from dbus when the remote adapters data is corrupted."""
|
|
address = "AA:BB:CC:CC:CC:FF"
|
|
|
|
data = hass_storage[storage.REMOTE_SCANNER_STORAGE_KEY] = json_loads(
|
|
await async_load_fixture(
|
|
hass, "bluetooth.remote_scanners.corrupt", bluetooth.DOMAIN
|
|
)
|
|
)
|
|
now = time.time()
|
|
timestamps = data["data"]["atom-bluetooth-proxy-ceaac4"][
|
|
"discovered_device_timestamps"
|
|
]
|
|
for address in timestamps:
|
|
timestamps[address] = now
|
|
|
|
ble_device = generate_ble_device(address, "name")
|
|
history = {
|
|
address: AdvertisementHistory(
|
|
ble_device,
|
|
generate_advertisement_data(local_name="name"),
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
}
|
|
|
|
with patch(
|
|
"bluetooth_adapters.systems.linux.LinuxAdapters.history",
|
|
history,
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert bluetooth.async_ble_device_from_address(hass, address) is not None
|
|
assert bluetooth.async_ble_device_from_address(hass, "EB:0B:36:35:6F:A4") is None
|
|
assert disable_new_discovery_flows.call_count >= 1
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_based_on_rssi_connectable_to_non_connectable(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""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 = generate_ble_device(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_SOURCE_ADDRESS,
|
|
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 = generate_ble_device(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 = generate_ble_device(
|
|
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
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_connectable_advertisement_can_be_retrieved_with_best_path_is_non_connectable(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test we can still get a connectable BLEDevice when the best path is non-connectable.
|
|
|
|
In this case the device is closer to a non-connectable scanner, but the
|
|
at least one connectable scanner has the device in range.
|
|
"""
|
|
|
|
address = "44:44:33:11:23:45"
|
|
now = time.monotonic()
|
|
switchbot_device_good_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
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 None
|
|
|
|
switchbot_device_poor_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
True,
|
|
)
|
|
|
|
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
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_when_one_goes_away(
|
|
hass: HomeAssistant, register_hci0_scanner: None
|
|
) -> None:
|
|
"""Test switching adapters when one goes away."""
|
|
cancel_hci2 = bluetooth.async_register_scanner(hass, FakeScanner("hci2", "hci2"))
|
|
|
|
address = "44:44:33:11:23:45"
|
|
|
|
switchbot_device_good_signal = generate_ble_device(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, "hci2"
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
switchbot_device_poor_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# We want to prefer the good signal when we have options
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
cancel_hci2()
|
|
|
|
inject_advertisement_with_source(
|
|
hass,
|
|
switchbot_device_poor_signal,
|
|
switchbot_adv_poor_signal,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# Now that hci2 is gone, we should prefer the poor signal
|
|
# since no poor signal is better than no signal
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal
|
|
)
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_switching_adapters_when_one_stop_scanning(
|
|
hass: HomeAssistant, register_hci0_scanner: None
|
|
) -> None:
|
|
"""Test switching adapters when stops scanning."""
|
|
hci2_scanner = FakeScanner("hci2", "hci2")
|
|
cancel_hci2 = bluetooth.async_register_scanner(hass, hci2_scanner)
|
|
|
|
address = "44:44:33:11:23:45"
|
|
|
|
switchbot_device_good_signal = generate_ble_device(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, "hci2"
|
|
)
|
|
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
switchbot_device_poor_signal = generate_ble_device(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_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# We want to prefer the good signal when we have options
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_good_signal
|
|
)
|
|
|
|
hci2_scanner.scanning = False
|
|
|
|
inject_advertisement_with_source(
|
|
hass,
|
|
switchbot_device_poor_signal,
|
|
switchbot_adv_poor_signal,
|
|
HCI0_SOURCE_ADDRESS,
|
|
)
|
|
|
|
# Now that hci2 has stopped scanning, we should prefer the poor signal
|
|
# since poor signal is better than no signal
|
|
assert (
|
|
bluetooth.async_ble_device_from_address(hass, address)
|
|
is switchbot_device_poor_signal
|
|
)
|
|
|
|
cancel_hci2()
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_bluetooth_adapters")
|
|
async def test_goes_unavailable_connectable_only_and_recovers(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test all connectable scanners go unavailable, and than recover when there is a non-connectable scanner."""
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert async_scanner_count(hass, connectable=True) == 0
|
|
assert async_scanner_count(hass, connectable=False) == 0
|
|
switchbot_device_connectable = generate_ble_device(
|
|
"44:44:33:11:23:45",
|
|
"wohand",
|
|
{},
|
|
)
|
|
switchbot_device_non_connectable = generate_ble_device(
|
|
"44:44:33:11:23:45",
|
|
"wohand",
|
|
{},
|
|
)
|
|
switchbot_device_adv = generate_advertisement_data(
|
|
local_name="wohand",
|
|
service_uuids=["050a021a-0000-1000-8000-00805f9b34fb"],
|
|
service_data={"050a021a-0000-1000-8000-00805f9b34fb": b"\n\xff"},
|
|
manufacturer_data={1: b"\x01"},
|
|
rssi=-100,
|
|
)
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo,
|
|
change: BluetoothChange,
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45", "connectable": True},
|
|
BluetoothScanningMode.ACTIVE,
|
|
)
|
|
|
|
class FakeScanner(BaseHaRemoteScanner):
|
|
def inject_advertisement(
|
|
self, device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Inject an advertisement."""
|
|
self._async_on_advertisement(
|
|
device.address,
|
|
advertisement_data.rssi,
|
|
device.name,
|
|
advertisement_data.service_uuids,
|
|
advertisement_data.service_data,
|
|
advertisement_data.manufacturer_data,
|
|
advertisement_data.tx_power,
|
|
{"scanner_specific_data": "test"},
|
|
MONOTONIC_TIME(),
|
|
)
|
|
|
|
connector = (
|
|
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
|
|
)
|
|
connectable_scanner = FakeScanner(
|
|
"connectable",
|
|
"connectable",
|
|
connector,
|
|
True,
|
|
)
|
|
unsetup_connectable_scanner = connectable_scanner.async_setup()
|
|
cancel_connectable_scanner = _get_manager().async_register_scanner(
|
|
connectable_scanner
|
|
)
|
|
connectable_scanner.inject_advertisement(
|
|
switchbot_device_connectable, switchbot_device_adv
|
|
)
|
|
assert async_ble_device_from_address(hass, "44:44:33:11:23:45") is not None
|
|
assert async_scanner_count(hass, connectable=True) == 1
|
|
assert len(callbacks) == 1
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
not_connectable_scanner = FakeScanner(
|
|
"not_connectable",
|
|
"not_connectable",
|
|
connector,
|
|
False,
|
|
)
|
|
unsetup_not_connectable_scanner = not_connectable_scanner.async_setup()
|
|
cancel_not_connectable_scanner = _get_manager().async_register_scanner(
|
|
not_connectable_scanner
|
|
)
|
|
not_connectable_scanner.inject_advertisement(
|
|
switchbot_device_non_connectable, switchbot_device_adv
|
|
)
|
|
assert async_scanner_count(hass, connectable=True) == 1
|
|
assert async_scanner_count(hass, connectable=False) == 2
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in not_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
unavailable_callbacks: list[BluetoothServiceInfoBleak] = []
|
|
|
|
@callback
|
|
def _unavailable_callback(service_info: BluetoothServiceInfoBleak) -> None:
|
|
"""Wrong device unavailable callback."""
|
|
nonlocal unavailable_callbacks
|
|
unavailable_callbacks.append(service_info.address)
|
|
|
|
cancel_unavailable = async_track_unavailable(
|
|
hass,
|
|
_unavailable_callback,
|
|
switchbot_device_connectable.address,
|
|
connectable=True,
|
|
)
|
|
|
|
assert async_scanner_count(hass, connectable=True) == 1
|
|
cancel_connectable_scanner()
|
|
unsetup_connectable_scanner()
|
|
assert async_scanner_count(hass, connectable=True) == 0
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
|
|
async_fire_time_changed(
|
|
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert "44:44:33:11:23:45" in unavailable_callbacks
|
|
cancel_unavailable()
|
|
|
|
connectable_scanner_2 = FakeScanner(
|
|
"connectable",
|
|
"connectable",
|
|
connector,
|
|
True,
|
|
)
|
|
unsetup_connectable_scanner_2 = connectable_scanner_2.async_setup()
|
|
cancel_connectable_scanner_2 = _get_manager().async_register_scanner(
|
|
connectable_scanner
|
|
)
|
|
connectable_scanner_2.inject_advertisement(
|
|
switchbot_device_connectable, switchbot_device_adv
|
|
)
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in connectable_scanner_2.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
# We should get another callback to make the device available again
|
|
assert len(callbacks) == 2
|
|
|
|
cancel()
|
|
cancel_connectable_scanner_2()
|
|
unsetup_connectable_scanner_2()
|
|
cancel_not_connectable_scanner()
|
|
unsetup_not_connectable_scanner()
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_bluetooth_adapters")
|
|
async def test_goes_unavailable_dismisses_discovery_and_makes_discoverable(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test that unavailable will dismiss any active discoveries and make device discoverable again."""
|
|
mock_bt = [
|
|
{
|
|
"domain": "switchbot",
|
|
"service_data_uuid": "050a021a-0000-1000-8000-00805f9b34fb",
|
|
"connectable": False,
|
|
},
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert async_scanner_count(hass, connectable=False) == 0
|
|
switchbot_device_non_connectable = generate_ble_device(
|
|
"44:44:33:11:23:45",
|
|
"wohand",
|
|
{},
|
|
)
|
|
switchbot_device_adv = generate_advertisement_data(
|
|
local_name="wohand",
|
|
service_uuids=["050a021a-0000-1000-8000-00805f9b34fb"],
|
|
service_data={"050a021a-0000-1000-8000-00805f9b34fb": b"\n\xff"},
|
|
manufacturer_data={1: b"\x01"},
|
|
rssi=-100,
|
|
)
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo,
|
|
change: BluetoothChange,
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45", "connectable": False},
|
|
BluetoothScanningMode.ACTIVE,
|
|
)
|
|
|
|
class FakeScanner(BaseHaRemoteScanner):
|
|
def inject_advertisement(
|
|
self, device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Inject an advertisement."""
|
|
self._async_on_advertisement(
|
|
device.address,
|
|
advertisement_data.rssi,
|
|
device.name,
|
|
advertisement_data.service_uuids,
|
|
advertisement_data.service_data,
|
|
advertisement_data.manufacturer_data,
|
|
advertisement_data.tx_power,
|
|
{"scanner_specific_data": "test"},
|
|
MONOTONIC_TIME(),
|
|
)
|
|
|
|
def clear_all_devices(self) -> None:
|
|
"""Clear all devices."""
|
|
self._previous_service_info.clear()
|
|
|
|
connector = (
|
|
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
|
|
)
|
|
non_connectable_scanner = FakeScanner(
|
|
"connectable",
|
|
"connectable",
|
|
connector,
|
|
False,
|
|
)
|
|
unsetup_connectable_scanner = non_connectable_scanner.async_setup()
|
|
cancel_connectable_scanner = _get_manager().async_register_scanner(
|
|
non_connectable_scanner
|
|
)
|
|
with patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
non_connectable_scanner.inject_advertisement(
|
|
switchbot_device_non_connectable, switchbot_device_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"
|
|
assert mock_config_flow.mock_calls[0][2]["context"] == {
|
|
"discovery_key": DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
"source": "bluetooth",
|
|
}
|
|
|
|
assert async_ble_device_from_address(hass, "44:44:33:11:23:45", False) is not None
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
assert len(callbacks) == 1
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in non_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
unavailable_callbacks: list[BluetoothServiceInfoBleak] = []
|
|
|
|
@callback
|
|
def _unavailable_callback(service_info: BluetoothServiceInfoBleak) -> None:
|
|
"""Wrong device unavailable callback."""
|
|
nonlocal unavailable_callbacks
|
|
unavailable_callbacks.append(service_info.address)
|
|
|
|
cancel_unavailable = async_track_unavailable(
|
|
hass,
|
|
_unavailable_callback,
|
|
switchbot_device_non_connectable.address,
|
|
connectable=False,
|
|
)
|
|
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
|
|
non_connectable_scanner.clear_all_devices()
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
not in non_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
monotonic_now = time.monotonic()
|
|
with (
|
|
patch.object(
|
|
hass.config_entries.flow,
|
|
"async_progress_by_init_data_type",
|
|
return_value=[{"flow_id": "mock_flow_id"}],
|
|
) as mock_async_progress_by_init_data_type,
|
|
patch.object(hass.config_entries.flow, "async_abort") as mock_async_abort,
|
|
patch_bluetooth_time(
|
|
monotonic_now + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
|
|
),
|
|
):
|
|
async_fire_time_changed(
|
|
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
|
|
)
|
|
await hass.async_block_till_done()
|
|
assert "44:44:33:11:23:45" in unavailable_callbacks
|
|
|
|
assert len(mock_async_progress_by_init_data_type.mock_calls) == 1
|
|
assert mock_async_abort.mock_calls[0][1][0] == "mock_flow_id"
|
|
|
|
# Test that if the device comes back online, it can be discovered again
|
|
with patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
new_switchbot_device_adv = generate_advertisement_data(
|
|
local_name="wohand",
|
|
service_uuids=["050a021a-0000-1000-8000-00805f9b34fb"],
|
|
service_data={"050a021a-0000-1000-8000-00805f9b34fb": b"\n\xff"},
|
|
manufacturer_data={1: b"\x01"},
|
|
rssi=-60,
|
|
)
|
|
non_connectable_scanner.inject_advertisement(
|
|
switchbot_device_non_connectable, new_switchbot_device_adv
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in non_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
|
|
assert mock_config_flow.mock_calls[0][2]["context"] == {
|
|
"discovery_key": DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
"source": "bluetooth",
|
|
}
|
|
|
|
cancel_unavailable()
|
|
|
|
cancel()
|
|
unsetup_connectable_scanner()
|
|
cancel_connectable_scanner()
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_debug_logging(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Test debug logging."""
|
|
assert await async_setup_component(hass, "logger", {"logger": {}})
|
|
async with async_call_logger_set_level(
|
|
"homeassistant.components.bluetooth", "DEBUG", hass=hass, caplog=caplog
|
|
):
|
|
address = "44:44:33:11:23:41"
|
|
start_time_monotonic = 50.0
|
|
|
|
switchbot_device_poor_signal_hci0 = generate_ble_device(
|
|
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,
|
|
switchbot_device_poor_signal_hci0,
|
|
switchbot_adv_poor_signal_hci0,
|
|
start_time_monotonic,
|
|
"hci0",
|
|
)
|
|
assert "wohand_poor_signal_hci0" in caplog.text
|
|
caplog.clear()
|
|
|
|
async with async_call_logger_set_level(
|
|
"homeassistant.components.bluetooth", "WARNING", hass=hass, caplog=caplog
|
|
):
|
|
switchbot_device_good_signal_hci0 = generate_ble_device(
|
|
address, "wohand_good_signal_hci0"
|
|
)
|
|
switchbot_adv_good_signal_hci0 = generate_advertisement_data(
|
|
local_name="wohand_good_signal_hci0", service_uuids=[], rssi=-33
|
|
)
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device_good_signal_hci0,
|
|
switchbot_adv_good_signal_hci0,
|
|
start_time_monotonic,
|
|
"hci0",
|
|
)
|
|
assert "wohand_good_signal_hci0" not in caplog.text
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth", "macos_adapter")
|
|
async def test_set_fallback_interval_small(hass: HomeAssistant) -> None:
|
|
"""Test we can set the fallback advertisement interval."""
|
|
assert async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
async_set_fallback_availability_interval(hass, "44:44:33:11:23:12", 2.0)
|
|
assert async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") == 2.0
|
|
|
|
start_monotonic_time = time.monotonic()
|
|
switchbot_device = generate_ble_device("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
|
|
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device,
|
|
switchbot_adv,
|
|
start_monotonic_time,
|
|
SOURCE_LOCAL,
|
|
)
|
|
|
|
@callback
|
|
def _switchbot_device_unavailable_callback(_address: str) -> None:
|
|
"""Switchbot device unavailable callback."""
|
|
nonlocal switchbot_device_went_unavailable
|
|
switchbot_device_went_unavailable = True
|
|
|
|
assert async_get_learned_advertising_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
switchbot_device_unavailable_cancel = async_track_unavailable(
|
|
hass,
|
|
_switchbot_device_unavailable_callback,
|
|
switchbot_device.address,
|
|
connectable=False,
|
|
)
|
|
|
|
monotonic_now = start_monotonic_time + 2
|
|
with patch_bluetooth_time(
|
|
monotonic_now + UNAVAILABLE_TRACK_SECONDS,
|
|
):
|
|
async_fire_time_changed(
|
|
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert switchbot_device_went_unavailable is True
|
|
switchbot_device_unavailable_cancel()
|
|
|
|
# We should forget fallback interval after it expires
|
|
assert async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth", "macos_adapter")
|
|
async def test_set_fallback_interval_big(hass: HomeAssistant) -> None:
|
|
"""Test we can set the fallback advertisement interval."""
|
|
assert async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
# Force the interval to be really big and check it doesn't expire using the default timeout (900)
|
|
|
|
async_set_fallback_availability_interval(hass, "44:44:33:11:23:12", 604800.0)
|
|
assert (
|
|
async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") == 604800.0
|
|
)
|
|
|
|
start_monotonic_time = time.monotonic()
|
|
switchbot_device = generate_ble_device("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
|
|
|
|
inject_advertisement_with_time_and_source(
|
|
hass,
|
|
switchbot_device,
|
|
switchbot_adv,
|
|
start_monotonic_time,
|
|
SOURCE_LOCAL,
|
|
)
|
|
|
|
@callback
|
|
def _switchbot_device_unavailable_callback(_address: str) -> None:
|
|
"""Switchbot device unavailable callback."""
|
|
nonlocal switchbot_device_went_unavailable
|
|
switchbot_device_went_unavailable = True
|
|
|
|
assert async_get_learned_advertising_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
switchbot_device_unavailable_cancel = async_track_unavailable(
|
|
hass,
|
|
_switchbot_device_unavailable_callback,
|
|
switchbot_device.address,
|
|
connectable=False,
|
|
)
|
|
|
|
# Check that device hasn't expired after a day
|
|
|
|
monotonic_now = start_monotonic_time + 86400
|
|
with patch_bluetooth_time(
|
|
monotonic_now + UNAVAILABLE_TRACK_SECONDS,
|
|
):
|
|
async_fire_time_changed(
|
|
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert switchbot_device_went_unavailable is False
|
|
|
|
# Try again after it has expired
|
|
|
|
monotonic_now = start_monotonic_time + 604800
|
|
with patch_bluetooth_time(
|
|
monotonic_now + UNAVAILABLE_TRACK_SECONDS,
|
|
):
|
|
async_fire_time_changed(
|
|
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
assert switchbot_device_went_unavailable is True
|
|
|
|
switchbot_device_unavailable_cancel()
|
|
|
|
# We should forget fallback interval after it expires
|
|
assert async_get_fallback_availability_interval(hass, "44:44:33:11:23:12") is None
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_bluetooth_adapters")
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"entry_domain",
|
|
"entry_discovery_keys",
|
|
),
|
|
[
|
|
# Matching discovery key
|
|
(
|
|
"switchbot",
|
|
{
|
|
"bluetooth": (
|
|
DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
)
|
|
},
|
|
),
|
|
# Matching discovery key
|
|
(
|
|
"switchbot",
|
|
{
|
|
"bluetooth": (
|
|
DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
),
|
|
"other": (DiscoveryKey(domain="other", key="blah", version=1),),
|
|
},
|
|
),
|
|
# Matching discovery key, other domain
|
|
# Note: Rediscovery is not currently restricted to the domain of the removed
|
|
# entry. Such a check can be added if needed.
|
|
(
|
|
"comp",
|
|
{
|
|
"bluetooth": (
|
|
DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
)
|
|
},
|
|
),
|
|
],
|
|
)
|
|
@pytest.mark.parametrize(
|
|
"entry_source",
|
|
[
|
|
config_entries.SOURCE_BLUETOOTH,
|
|
config_entries.SOURCE_IGNORE,
|
|
config_entries.SOURCE_USER,
|
|
],
|
|
)
|
|
async def test_bluetooth_rediscover(
|
|
hass: HomeAssistant,
|
|
entry_domain: str,
|
|
entry_discovery_keys: dict[str, tuple[DiscoveryKey, ...]],
|
|
entry_source: str,
|
|
) -> None:
|
|
"""Test we reinitiate flows when an ignored config entry is removed."""
|
|
mock_bt = [
|
|
{
|
|
"domain": "switchbot",
|
|
"service_data_uuid": "050a021a-0000-1000-8000-00805f9b34fb",
|
|
"connectable": False,
|
|
},
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert async_scanner_count(hass, connectable=False) == 0
|
|
switchbot_device_non_connectable = generate_ble_device(
|
|
"44:44:33:11:23:45",
|
|
"wohand",
|
|
{},
|
|
)
|
|
switchbot_device_adv = generate_advertisement_data(
|
|
local_name="wohand",
|
|
service_uuids=["050a021a-0000-1000-8000-00805f9b34fb"],
|
|
service_data={"050a021a-0000-1000-8000-00805f9b34fb": b"\n\xff"},
|
|
manufacturer_data={1: b"\x01"},
|
|
rssi=-100,
|
|
)
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo,
|
|
change: BluetoothChange,
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45", "connectable": False},
|
|
BluetoothScanningMode.ACTIVE,
|
|
)
|
|
|
|
class FakeScanner(BaseHaRemoteScanner):
|
|
def inject_advertisement(
|
|
self, device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Inject an advertisement."""
|
|
self._async_on_advertisement(
|
|
device.address,
|
|
advertisement_data.rssi,
|
|
device.name,
|
|
advertisement_data.service_uuids,
|
|
advertisement_data.service_data,
|
|
advertisement_data.manufacturer_data,
|
|
advertisement_data.tx_power,
|
|
{"scanner_specific_data": "test"},
|
|
MONOTONIC_TIME(),
|
|
)
|
|
|
|
def clear_all_devices(self) -> None:
|
|
"""Clear all devices."""
|
|
self._previous_service_info.clear()
|
|
|
|
connector = (
|
|
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
|
|
)
|
|
non_connectable_scanner = FakeScanner(
|
|
"connectable",
|
|
"connectable",
|
|
connector,
|
|
False,
|
|
)
|
|
unsetup_connectable_scanner = non_connectable_scanner.async_setup()
|
|
cancel_connectable_scanner = _get_manager().async_register_scanner(
|
|
non_connectable_scanner
|
|
)
|
|
with patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
non_connectable_scanner.inject_advertisement(
|
|
switchbot_device_non_connectable, switchbot_device_adv
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
expected_context = {
|
|
"discovery_key": DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
"source": "bluetooth",
|
|
}
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
|
|
assert mock_config_flow.mock_calls[0][2]["context"] == expected_context
|
|
|
|
hass.config.components.add(entry_domain)
|
|
mock_integration(hass, MockModule(entry_domain))
|
|
|
|
entry = MockConfigEntry(
|
|
domain=entry_domain,
|
|
discovery_keys=entry_discovery_keys,
|
|
unique_id="mock-unique-id",
|
|
state=config_entries.ConfigEntryState.LOADED,
|
|
source=entry_source,
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert (
|
|
async_ble_device_from_address(hass, "44:44:33:11:23:45", False) is not None
|
|
)
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
assert len(callbacks) == 1
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in non_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
await hass.config_entries.async_remove(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert (
|
|
async_ble_device_from_address(hass, "44:44:33:11:23:45", False) is not None
|
|
)
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
assert len(callbacks) == 1
|
|
|
|
assert len(mock_config_flow.mock_calls) == 2
|
|
assert mock_config_flow.mock_calls[1][1][0] == "switchbot"
|
|
assert mock_config_flow.mock_calls[1][2]["context"] == expected_context
|
|
|
|
cancel()
|
|
unsetup_connectable_scanner()
|
|
cancel_connectable_scanner()
|
|
|
|
|
|
@pytest.mark.usefixtures("mock_bluetooth_adapters")
|
|
@pytest.mark.parametrize(
|
|
(
|
|
"entry_domain",
|
|
"entry_discovery_keys",
|
|
"entry_source",
|
|
"entry_unique_id",
|
|
),
|
|
[
|
|
# Discovery key from other domain
|
|
(
|
|
"switchbot",
|
|
{
|
|
"zeroconf": (
|
|
DiscoveryKey(domain="zeroconf", key="44:44:33:11:23:45", version=1),
|
|
)
|
|
},
|
|
config_entries.SOURCE_IGNORE,
|
|
"mock-unique-id",
|
|
),
|
|
# Discovery key from the future
|
|
(
|
|
"switchbot",
|
|
{
|
|
"bluetooth": (
|
|
DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=2
|
|
),
|
|
)
|
|
},
|
|
config_entries.SOURCE_IGNORE,
|
|
"mock-unique-id",
|
|
),
|
|
],
|
|
)
|
|
async def test_bluetooth_rediscover_no_match(
|
|
hass: HomeAssistant,
|
|
entry_domain: str,
|
|
entry_discovery_keys: dict[str, tuple[DiscoveryKey, ...]],
|
|
entry_source: str,
|
|
entry_unique_id: str,
|
|
) -> None:
|
|
"""Test we don't reinitiate flows when a non matching config entry is removed."""
|
|
mock_bt = [
|
|
{
|
|
"domain": "switchbot",
|
|
"service_data_uuid": "050a021a-0000-1000-8000-00805f9b34fb",
|
|
"connectable": False,
|
|
},
|
|
]
|
|
with patch(
|
|
"homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
assert async_scanner_count(hass, connectable=False) == 0
|
|
switchbot_device_non_connectable = generate_ble_device(
|
|
"44:44:33:11:23:45",
|
|
"wohand",
|
|
{},
|
|
)
|
|
switchbot_device_adv = generate_advertisement_data(
|
|
local_name="wohand",
|
|
service_uuids=["050a021a-0000-1000-8000-00805f9b34fb"],
|
|
service_data={"050a021a-0000-1000-8000-00805f9b34fb": b"\n\xff"},
|
|
manufacturer_data={1: b"\x01"},
|
|
rssi=-100,
|
|
)
|
|
callbacks = []
|
|
|
|
def _fake_subscriber(
|
|
service_info: BluetoothServiceInfo,
|
|
change: BluetoothChange,
|
|
) -> None:
|
|
"""Fake subscriber for the BleakScanner."""
|
|
callbacks.append((service_info, change))
|
|
|
|
cancel = bluetooth.async_register_callback(
|
|
hass,
|
|
_fake_subscriber,
|
|
{"address": "44:44:33:11:23:45", "connectable": False},
|
|
BluetoothScanningMode.ACTIVE,
|
|
)
|
|
|
|
class FakeScanner(BaseHaRemoteScanner):
|
|
def inject_advertisement(
|
|
self, device: BLEDevice, advertisement_data: AdvertisementData
|
|
) -> None:
|
|
"""Inject an advertisement."""
|
|
self._async_on_advertisement(
|
|
device.address,
|
|
advertisement_data.rssi,
|
|
device.name,
|
|
advertisement_data.service_uuids,
|
|
advertisement_data.service_data,
|
|
advertisement_data.manufacturer_data,
|
|
advertisement_data.tx_power,
|
|
{"scanner_specific_data": "test"},
|
|
MONOTONIC_TIME(),
|
|
)
|
|
|
|
def clear_all_devices(self) -> None:
|
|
"""Clear all devices."""
|
|
self._previous_service_info.clear()
|
|
|
|
connector = (
|
|
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
|
|
)
|
|
non_connectable_scanner = FakeScanner(
|
|
"connectable",
|
|
"connectable",
|
|
connector,
|
|
False,
|
|
)
|
|
unsetup_connectable_scanner = non_connectable_scanner.async_setup()
|
|
cancel_connectable_scanner = _get_manager().async_register_scanner(
|
|
non_connectable_scanner
|
|
)
|
|
with patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
|
|
non_connectable_scanner.inject_advertisement(
|
|
switchbot_device_non_connectable, switchbot_device_adv
|
|
)
|
|
await hass.async_block_till_done()
|
|
|
|
expected_context = {
|
|
"discovery_key": DiscoveryKey(
|
|
domain="bluetooth", key="44:44:33:11:23:45", version=1
|
|
),
|
|
"source": "bluetooth",
|
|
}
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
assert mock_config_flow.mock_calls[0][1][0] == "switchbot"
|
|
assert mock_config_flow.mock_calls[0][2]["context"] == expected_context
|
|
|
|
hass.config.components.add(entry_domain)
|
|
mock_integration(hass, MockModule(entry_domain))
|
|
|
|
entry = MockConfigEntry(
|
|
domain=entry_domain,
|
|
discovery_keys=entry_discovery_keys,
|
|
unique_id=entry_unique_id,
|
|
state=config_entries.ConfigEntryState.LOADED,
|
|
source=entry_source,
|
|
)
|
|
entry.add_to_hass(hass)
|
|
|
|
assert (
|
|
async_ble_device_from_address(hass, "44:44:33:11:23:45", False) is not None
|
|
)
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
assert len(callbacks) == 1
|
|
|
|
assert (
|
|
"44:44:33:11:23:45"
|
|
in non_connectable_scanner.discovered_devices_and_advertisement_data
|
|
)
|
|
|
|
await hass.config_entries.async_remove(entry.entry_id)
|
|
await hass.async_block_till_done()
|
|
|
|
assert (
|
|
async_ble_device_from_address(hass, "44:44:33:11:23:45", False) is not None
|
|
)
|
|
assert async_scanner_count(hass, connectable=False) == 1
|
|
assert len(callbacks) == 1
|
|
assert len(mock_config_flow.mock_calls) == 1
|
|
|
|
cancel()
|
|
unsetup_connectable_scanner()
|
|
cancel_connectable_scanner()
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_async_register_disappeared_callback(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
) -> None:
|
|
"""Test bluetooth async_register_disappeared_callback handles failures."""
|
|
address = "44:44:33:11:23:12"
|
|
|
|
switchbot_device_signal_100 = generate_ble_device(address, "wohand_signal_100")
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100", service_uuids=[], rssi=-100
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_100, switchbot_adv_signal_100, "hci0"
|
|
)
|
|
|
|
failed_disappeared: list[str] = []
|
|
|
|
def _failing_callback(_address: str) -> None:
|
|
"""Failing callback."""
|
|
failed_disappeared.append(_address)
|
|
raise ValueError("This is a test")
|
|
|
|
ok_disappeared: list[str] = []
|
|
|
|
def _ok_callback(_address: str) -> None:
|
|
"""Ok callback."""
|
|
ok_disappeared.append(_address)
|
|
|
|
manager: HomeAssistantBluetoothManager = _get_manager()
|
|
cancel1 = manager.async_register_disappeared_callback(_failing_callback)
|
|
# Make sure the second callback still works if the first one fails and
|
|
# raises an exception
|
|
cancel2 = manager.async_register_disappeared_callback(_ok_callback)
|
|
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100",
|
|
manufacturer_data={123: b"abc"},
|
|
service_uuids=[],
|
|
rssi=-80,
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_100, switchbot_adv_signal_100, "hci1"
|
|
)
|
|
|
|
future_time = utcnow() + timedelta(seconds=3600)
|
|
future_monotonic_time = time.monotonic() + 3600
|
|
with (
|
|
freeze_time(future_time),
|
|
patch(
|
|
"habluetooth.manager.monotonic_time_coarse",
|
|
return_value=future_monotonic_time,
|
|
),
|
|
):
|
|
async_fire_time_changed(hass, future_time)
|
|
|
|
assert len(ok_disappeared) == 1
|
|
assert ok_disappeared[0] == address
|
|
assert len(failed_disappeared) == 1
|
|
assert failed_disappeared[0] == address
|
|
|
|
cancel1()
|
|
cancel2()
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_repair_issue_created_for_degraded_scanner_in_docker(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test repair issue is created when scanner is in degraded mode in Docker."""
|
|
await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
mock_adapters = {
|
|
"hci0": {
|
|
"address": "00:11:22:33:44:55",
|
|
"sw_version": "homeassistant",
|
|
"hw_version": "usb:v0A5Cp21E8",
|
|
"passive_scan": False,
|
|
"manufacturer": "Broadcom",
|
|
"product": "BCM20702A0",
|
|
"vendor_id": "0A5C",
|
|
"product_id": "21E8",
|
|
}
|
|
}
|
|
|
|
with (
|
|
patch("habluetooth.manager.IS_LINUX", True),
|
|
patch.object(type(manager), "is_operating_degraded", return_value=True),
|
|
patch(
|
|
"homeassistant.components.bluetooth.manager.is_docker_env",
|
|
return_value=True,
|
|
),
|
|
patch.object(manager._bluetooth_adapters, "adapters", mock_adapters),
|
|
):
|
|
manager.on_scanner_start(scanner)
|
|
|
|
issue_id = f"bluetooth_adapter_missing_permissions_{scanner.source}"
|
|
registry = ir.async_get(hass)
|
|
issue = registry.async_get_issue(bluetooth.DOMAIN, issue_id)
|
|
assert issue is not None
|
|
assert issue.severity == ir.IssueSeverity.WARNING
|
|
assert not issue.is_fixable
|
|
assert issue.translation_key == "bluetooth_adapter_missing_permissions"
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_repair_issue_deleted_when_scanner_not_degraded(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test repair issue is deleted when scanner is not in degraded mode."""
|
|
await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
registry = ir.async_get(hass)
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
mock_adapters = {
|
|
"hci0": {
|
|
"address": "00:11:22:33:44:55",
|
|
"sw_version": "homeassistant",
|
|
"hw_version": "usb:v0A5Cp21E8",
|
|
"passive_scan": False,
|
|
"manufacturer": "Broadcom",
|
|
"product": "BCM20702A0",
|
|
"vendor_id": "0A5C",
|
|
"product_id": "21E8",
|
|
}
|
|
}
|
|
|
|
issue_id = f"bluetooth_adapter_missing_permissions_{scanner.source}"
|
|
|
|
with (
|
|
patch("habluetooth.manager.IS_LINUX", True),
|
|
patch.object(type(manager), "is_operating_degraded", return_value=True),
|
|
patch(
|
|
"homeassistant.components.bluetooth.manager.is_docker_env",
|
|
return_value=True,
|
|
),
|
|
patch.object(manager._bluetooth_adapters, "adapters", mock_adapters),
|
|
):
|
|
manager.on_scanner_start(scanner)
|
|
|
|
assert registry.async_get_issue(bluetooth.DOMAIN, issue_id) is not None
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.bluetooth.manager.is_docker_env",
|
|
return_value=True,
|
|
),
|
|
patch.object(type(manager), "is_operating_degraded", return_value=False),
|
|
):
|
|
manager.on_scanner_start(scanner)
|
|
|
|
assert registry.async_get_issue(bluetooth.DOMAIN, issue_id) is None
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_no_repair_issue_when_not_docker(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test no repair issue is created when not running in Docker."""
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.bluetooth.manager.is_docker_env",
|
|
return_value=False,
|
|
),
|
|
patch.object(type(manager), "is_operating_degraded", return_value=True),
|
|
):
|
|
manager.on_scanner_start(scanner)
|
|
|
|
issue_id = f"bluetooth_adapter_missing_permissions_{scanner.source}"
|
|
registry = ir.async_get(hass)
|
|
assert registry.async_get_issue(bluetooth.DOMAIN, issue_id) is None
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_no_repair_issue_for_remote_scanner(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test no repair issue is created for remote scanners."""
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
connector = HaBluetoothConnector(MockBleakClient, "mock_connector", lambda: False)
|
|
scanner = FakeRemoteScanner("remote_scanner", "esp32", connector, True)
|
|
|
|
with (
|
|
patch(
|
|
"homeassistant.components.bluetooth.manager.is_docker_env",
|
|
return_value=True,
|
|
),
|
|
patch.object(type(manager), "is_operating_degraded", return_value=True),
|
|
):
|
|
manager.on_scanner_start(scanner)
|
|
|
|
registry = ir.async_get(hass)
|
|
issues = [
|
|
issue
|
|
for issue in registry.issues.values()
|
|
if issue.domain == bluetooth.DOMAIN
|
|
and "bluetooth_adapter_missing_permissions" in issue.issue_id
|
|
]
|
|
assert len(issues) == 0
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_repair_issue_created_for_passive_mode_fallback(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test repair issue is created when scanner falls back to passive mode."""
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
cancel = manager.async_register_scanner(scanner, connection_slots=1)
|
|
|
|
# Set scanner to passive mode when active was requested
|
|
scanner.set_requested_mode(BluetoothScanningMode.ACTIVE)
|
|
scanner.set_current_mode(BluetoothScanningMode.PASSIVE)
|
|
|
|
manager.on_scanner_start(scanner)
|
|
|
|
# Check repair issue is created
|
|
issue_id = f"bluetooth_adapter_passive_mode_{scanner.source}"
|
|
registry = ir.async_get(hass)
|
|
issue = registry.async_get_issue(bluetooth.DOMAIN, issue_id)
|
|
assert issue is not None
|
|
assert issue.severity == ir.IssueSeverity.WARNING
|
|
# Should default to USB translation key when adapter type is unknown
|
|
assert issue.translation_key == "bluetooth_adapter_passive_mode_usb"
|
|
assert not issue.is_fixable
|
|
|
|
cancel()
|
|
|
|
|
|
async def test_repair_issue_created_for_passive_mode_fallback_uart(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test repair issue is created with UART-specific message for UART adapters."""
|
|
with patch(
|
|
"bluetooth_adapters.systems.linux.LinuxAdapters.adapters",
|
|
{
|
|
"hci0": {
|
|
"address": "00:11:22:33:44:55",
|
|
"sw_version": "homeassistant",
|
|
"hw_version": "uart:bcm2711",
|
|
"passive_scan": False,
|
|
"manufacturer": "Raspberry Pi",
|
|
"product": "BCM2711",
|
|
"adapter_type": "uart", # UART adapter type
|
|
}
|
|
},
|
|
):
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
cancel = manager.async_register_scanner(scanner, connection_slots=1)
|
|
|
|
# Set scanner to passive mode when active was requested
|
|
scanner.set_requested_mode(BluetoothScanningMode.ACTIVE)
|
|
scanner.set_current_mode(BluetoothScanningMode.PASSIVE)
|
|
|
|
manager.on_scanner_start(scanner)
|
|
|
|
# Check repair issue is created with UART-specific translation key
|
|
issue_id = f"bluetooth_adapter_passive_mode_{scanner.source}"
|
|
registry = ir.async_get(hass)
|
|
issue = registry.async_get_issue(bluetooth.DOMAIN, issue_id)
|
|
assert issue is not None
|
|
assert issue.severity == ir.IssueSeverity.WARNING
|
|
assert issue.translation_key == "bluetooth_adapter_passive_mode_uart"
|
|
assert not issue.is_fixable
|
|
|
|
cancel()
|
|
|
|
|
|
@pytest.mark.usefixtures("one_adapter")
|
|
async def test_repair_issue_deleted_when_passive_mode_resolved(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test repair issue is deleted when scanner no longer in passive mode."""
|
|
assert await async_setup_component(hass, bluetooth.DOMAIN, {})
|
|
await hass.async_block_till_done()
|
|
|
|
manager = _get_manager()
|
|
|
|
scanner = HaScanner(
|
|
mode=BluetoothScanningMode.ACTIVE,
|
|
adapter="hci0",
|
|
address="00:11:22:33:44:55",
|
|
)
|
|
scanner.async_setup()
|
|
|
|
cancel = manager.async_register_scanner(scanner, connection_slots=1)
|
|
|
|
# Initially set scanner to passive mode when active was requested
|
|
scanner.set_requested_mode(BluetoothScanningMode.ACTIVE)
|
|
scanner.set_current_mode(BluetoothScanningMode.PASSIVE)
|
|
|
|
manager.on_scanner_start(scanner)
|
|
|
|
# Check repair issue is created
|
|
issue_id = f"bluetooth_adapter_passive_mode_{scanner.source}"
|
|
registry = ir.async_get(hass)
|
|
issue = registry.async_get_issue(bluetooth.DOMAIN, issue_id)
|
|
assert issue is not None
|
|
|
|
# Now simulate scanner recovering to active mode
|
|
scanner.set_current_mode(BluetoothScanningMode.ACTIVE)
|
|
manager.on_scanner_start(scanner)
|
|
|
|
# Check repair issue is deleted
|
|
issue = registry.async_get_issue(bluetooth.DOMAIN, issue_id)
|
|
assert issue is None
|
|
|
|
cancel()
|