jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-13 16:27:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Bluetooth config entries for remote scanners (#135543)

Browse Source
This commit is contained in:
J. Nick Koston 2025-01-16 09:52:52 -10:00 committed by GitHub
parent 93b3d76ee2
commit 6e255060c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 353 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
homeassistant/components/bluetooth/__init__.py
View File

@ -78,6 +78,9 @@ from .const import (
CONF_ADAPTER,
CONF_DETAILS,
CONF_PASSIVE,
CONF_SOURCE_CONFIG_ENTRY_ID,
CONF_SOURCE_DOMAIN,
CONF_SOURCE_MODEL,
DOMAIN,
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
LINUX_FIRMWARE_LOAD_FALLBACK_SECONDS,
@ -315,6 +318,32 @@ async def async_update_device(
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up a config entry for a bluetooth scanner."""
if source_entry_id := entry.data.get(CONF_SOURCE_CONFIG_ENTRY_ID):
if not (source_entry := hass.config_entries.async_get_entry(source_entry_id)):
# Cleanup the orphaned entry using a call_soon to ensure
# we can return before the entry is removed
hass.loop.call_soon(
hass_callback(
lambda: hass.async_create_task(
hass.config_entries.async_remove(entry.entry_id),
"remove orphaned bluetooth entry {entry.entry_id}",
)
)
)
address = entry.unique_id
assert address is not None
assert source_entry is not None
await async_update_device(
hass,
entry,
source_entry.title,
AdapterDetails(
address=address,
product=entry.data.get(CONF_SOURCE_MODEL),
manufacturer=entry.data[CONF_SOURCE_DOMAIN],
),
)
return True
manager = _get_manager(hass)
address = entry.unique_id
assert address is not None

7
homeassistant/components/bluetooth/api.py
View File

@ -178,9 +178,14 @@ def async_register_scanner(
hass: HomeAssistant,
scanner: BaseHaScanner,
connection_slots: int | None = None,
source_domain: str | None = None,
source_model: str | None = None,
source_config_entry_id: str | None = None,
) -> CALLBACK_TYPE:
"""Register a BleakScanner."""
return _get_manager(hass).async_register_scanner(scanner, connection_slots)
return _get_manager(hass).async_register_hass_scanner(
scanner, connection_slots, source_domain, source_model, source_config_entry_id
)
@hass_callback

52
homeassistant/components/bluetooth/config_flow.py
View File

@ -18,7 +18,12 @@ from habluetooth import get_manager
import voluptuous as vol
from homeassistant.components import onboarding
from homeassistant.config_entries import ConfigEntry, ConfigFlow, ConfigFlowResult
from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.core import callback
from homeassistant.helpers.schema_config_entry_flow import (
SchemaFlowFormStep,
@ -26,7 +31,16 @@ from homeassistant.helpers.schema_config_entry_flow import (
)
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import CONF_ADAPTER, CONF_DETAILS, CONF_PASSIVE, DOMAIN
from .const import (
CONF_ADAPTER,
CONF_DETAILS,
CONF_PASSIVE,
CONF_SOURCE,
CONF_SOURCE_CONFIG_ENTRY_ID,
CONF_SOURCE_DOMAIN,
CONF_SOURCE_MODEL,
DOMAIN,
)
from .util import adapter_title
OPTIONS_SCHEMA = vol.Schema(
@ -63,6 +77,8 @@ class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
self, discovery_info: DiscoveryInfoType
) -> ConfigFlowResult:
"""Handle a flow initialized by discovery."""
if discovery_info and CONF_SOURCE in discovery_info:
return await self.async_step_external_scanner(discovery_info)
self._adapter = cast(str, discovery_info[CONF_ADAPTER])
self._details = cast(AdapterDetails, discovery_info[CONF_DETAILS])
await self.async_set_unique_id(self._details[ADAPTER_ADDRESS])
@ -167,6 +183,24 @@ class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
),
)
async def async_step_external_scanner(
self, user_input: dict[str, Any]
) -> ConfigFlowResult:
"""Handle a flow initialized by an external scanner."""
source = user_input[CONF_SOURCE]
await self.async_set_unique_id(source)
data = {
CONF_SOURCE: source,
CONF_SOURCE_MODEL: user_input[CONF_SOURCE_MODEL],
CONF_SOURCE_DOMAIN: user_input[CONF_SOURCE_DOMAIN],
CONF_SOURCE_CONFIG_ENTRY_ID: user_input[CONF_SOURCE_CONFIG_ENTRY_ID],
}
self._abort_if_unique_id_configured(updates=data)
manager = get_manager()
scanner = manager.async_scanner_by_source(source)
assert scanner is not None
return self.async_create_entry(title=scanner.name, data=data)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
@ -177,8 +211,10 @@ class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
) -> SchemaOptionsFlowHandler:
) -> SchemaOptionsFlowHandler | RemoteAdapterOptionsFlowHandler:
"""Get the options flow for this handler."""
if CONF_SOURCE in config_entry.data:
return RemoteAdapterOptionsFlowHandler()
return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)
@classmethod
@ -186,3 +222,13 @@ class BluetoothConfigFlow(ConfigFlow, domain=DOMAIN):
def async_supports_options_flow(cls, config_entry: ConfigEntry) -> bool:
"""Return options flow support for this handler."""
return bool((manager := get_manager()) and manager.supports_passive_scan)
class RemoteAdapterOptionsFlowHandler(OptionsFlow):
"""Handle a option flow for remote adapters."""
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle options flow."""
return self.async_abort(reason="remote_adapters_not_supported")

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

@ -18,6 +18,12 @@ CONF_DETAILS = "details"
CONF_PASSIVE = "passive"
CONF_SOURCE: Final = "source"
CONF_SOURCE_DOMAIN: Final = "source_domain"
CONF_SOURCE_MODEL: Final = "source_model"
CONF_SOURCE_CONFIG_ENTRY_ID: Final = "source_config_entry_id"
SOURCE_LOCAL: Final = "local"
DATA_MANAGER: Final = "bluetooth_manager"

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

@ -22,7 +22,13 @@ from homeassistant.core import (
from homeassistant.helpers import discovery_flow
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from .const import DOMAIN
from .const import (
CONF_SOURCE,
CONF_SOURCE_CONFIG_ENTRY_ID,
CONF_SOURCE_DOMAIN,
CONF_SOURCE_MODEL,
DOMAIN,
)
from .match import (
ADDRESS,
CALLBACK,
@ -240,6 +246,39 @@ class HomeAssistantBluetoothManager(BluetoothManager):
unregister()
self._async_save_scanner_history(scanner)
@hass_callback
def async_register_hass_scanner(
self,
scanner: BaseHaScanner,
connection_slots: int | None = None,
source_domain: str | None = None,
source_model: str | None = None,
source_config_entry_id: str | None = None,
) -> CALLBACK_TYPE:
"""Register a scanner."""
cancel = self.async_register_scanner(scanner, connection_slots)
if (
isinstance(scanner, BaseHaRemoteScanner)
and source_domain
and source_config_entry_id
and not self.hass.config_entries.async_entry_for_domain_unique_id(
DOMAIN, scanner.source
)
):
self.hass.async_create_task(
self.hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
CONF_SOURCE: scanner.source,
CONF_SOURCE_DOMAIN: source_domain,
CONF_SOURCE_MODEL: source_model,
CONF_SOURCE_CONFIG_ENTRY_ID: source_config_entry_id,
},
)
)
return cancel
def async_register_scanner(
self,
scanner: BaseHaScanner,
@ -257,6 +296,13 @@ class HomeAssistantBluetoothManager(BluetoothManager):
def async_remove_scanner(self, source: str) -> None:
"""Remove a scanner."""
self.storage.async_remove_advertisement_history(source)
if entry := self.hass.config_entries.async_entry_for_domain_unique_id(
DOMAIN, source
):
self.hass.async_create_task(
self.hass.config_entries.async_remove(entry.entry_id),
f"Removing {source} Bluetooth config entry",
)
@hass_callback
def _handle_config_entry_removed(

3
homeassistant/components/bluetooth/strings.json
View File

@ -33,6 +33,9 @@
"passive": "Passive scanning"
}
}
},
"abort": {
"remote_adapters_not_supported": "Bluetooth configuration for remote adapters is not supported."
}
}
}

9
homeassistant/components/esphome/bluetooth.py
View File

@ -11,6 +11,7 @@ from bleak_esphome import connect_scanner
from homeassistant.components.bluetooth import async_register_scanner
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback
from .const import DOMAIN
from .entry_data import RuntimeEntryData
@ -38,7 +39,13 @@ def async_connect_scanner(
return partial(
_async_unload,
[
async_register_scanner(hass, scanner),
async_register_scanner(
hass,
scanner,
source_domain=DOMAIN,
source_model=device_info.model,
source_config_entry_id=entry_data.entry_id,
),
scanner.async_setup(),
],
)

8
homeassistant/components/shelly/bluetooth/__init__.py
View File

@ -28,7 +28,13 @@ async def async_connect_scanner(
source = format_mac(coordinator.mac).upper()
scanner = create_scanner(source, entry.title)
unload_callbacks = [
async_register_scanner(hass, scanner),
async_register_scanner(
hass,
scanner,
source_domain=entry.domain,
source_model=coordinator.model,
source_config_entry_id=entry.entry_id,
),
scanner.async_setup(),
coordinator.async_subscribe_events(scanner.async_on_event),
]

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

@ -14,7 +14,9 @@ from habluetooth import BaseHaScanner, BluetoothManager, get_manager
from homeassistant.components.bluetooth import (
DOMAIN,
MONOTONIC_TIME,
SOURCE_LOCAL,
BaseHaRemoteScanner,
BluetoothServiceInfo,
BluetoothServiceInfoBleak,
async_get_advertisement_callback,
@ -324,3 +326,26 @@ class FakeScanner(FakeScannerMixin, BaseHaScanner):
) -> dict[str, tuple[BLEDevice, AdvertisementData]]:
"""Return a list of discovered devices and their advertisement data."""
return {}
class FakeRemoteScanner(BaseHaRemoteScanner):
"""Fake remote scanner."""
def inject_advertisement(
self,
device: BLEDevice,
advertisement_data: AdvertisementData,
now: float | None = None,
) -> 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"},
now or MONOTONIC_TIME(),
)

81
tests/components/bluetooth/test_base_scanner.py
View File

@ -7,16 +7,12 @@ import time
from typing import Any
from unittest.mock import patch
from bleak.backends.device import BLEDevice
from bleak.backends.scanner import AdvertisementData
# pylint: disable-next=no-name-in-module
from habluetooth.advertisement_tracker import TRACKER_BUFFERING_WOBBLE_SECONDS
import pytest
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import (
MONOTONIC_TIME,
BaseHaRemoteScanner,
HaBluetoothConnector,
storage,
@ -28,12 +24,14 @@ from homeassistant.components.bluetooth.const import (
SCANNER_WATCHDOG_TIMEOUT,
UNAVAILABLE_TRACK_SECONDS,
)
from homeassistant.components.bluetooth.manager import HomeAssistantBluetoothManager
from homeassistant.core import HomeAssistant, callback
from homeassistant.setup import async_setup_component
import homeassistant.util.dt as dt_util
from homeassistant.util.json import json_loads
from . import (
FakeRemoteScanner as FakeScanner,
MockBleakClient,
_get_manager,
generate_advertisement_data,
@ -41,30 +39,7 @@ from . import (
patch_bluetooth_time,
)
from tests.common import async_fire_time_changed, load_fixture
class FakeScanner(BaseHaRemoteScanner):
"""Fake scanner."""
def inject_advertisement(
self,
device: BLEDevice,
advertisement_data: AdvertisementData,
now: float | None = None,
) -> 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"},
now or MONOTONIC_TIME(),
)
from tests.common import MockConfigEntry, async_fire_time_changed, load_fixture
@pytest.mark.parametrize("name_2", [None, "w"])
@ -545,3 +520,53 @@ async def test_scanner_stops_responding(hass: HomeAssistant) -> None:
cancel()
unsetup()
@pytest.mark.usefixtures("enable_bluetooth")
async def test_remote_scanner_bluetooth_config_entry(hass: HomeAssistant) -> None:
"""Test the remote scanner gets a bluetooth config entry."""
manager: HomeAssistantBluetoothManager = _get_manager()
switchbot_device = generate_ble_device(
"44:44:33:11:23:45",
"wohand",
{},
rssi=-100,
)
switchbot_device_adv = generate_advertisement_data(
local_name="wohand",
service_uuids=[],
manufacturer_data={1: b"\x01"},
rssi=-100,
)
connector = (
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
)
scanner = FakeScanner("esp32", "esp32", connector, True)
unsetup = scanner.async_setup()
entry = MockConfigEntry(domain="test")
entry.add_to_hass(hass)
cancel = manager.async_register_hass_scanner(
scanner,
source_domain="test",
source_model="test",
source_config_entry_id=entry.entry_id,
)
await hass.async_block_till_done()
scanner.inject_advertisement(switchbot_device, switchbot_device_adv)
assert len(scanner.discovered_devices) == 1
cancel()
unsetup()
assert hass.config_entries.async_entry_for_domain_unique_id(
"bluetooth", scanner.source
)
manager.async_remove_scanner(scanner.source)
await hass.async_block_till_done()
assert not hass.config_entries.async_entry_for_domain_unique_id(
"bluetooth", scanner.source
)

83
tests/components/bluetooth/test_config_flow.py
View File

@ -6,16 +6,23 @@ from bluetooth_adapters import DEFAULT_ADDRESS, AdapterDetails
import pytest
from homeassistant import config_entries
from homeassistant.components.bluetooth import HaBluetoothConnector
from homeassistant.components.bluetooth.const import (
CONF_ADAPTER,
CONF_DETAILS,
CONF_PASSIVE,
CONF_SOURCE,
CONF_SOURCE_CONFIG_ENTRY_ID,
CONF_SOURCE_DOMAIN,
CONF_SOURCE_MODEL,
DOMAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.setup import async_setup_component
from . import FakeRemoteScanner, MockBleakClient, _get_manager
from tests.common import MockConfigEntry
from tests.typing import WebSocketGenerator
@ -450,6 +457,36 @@ async def test_options_flow_enabled_linux(
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.usefixtures(
"one_adapter", "mock_bleak_scanner_start", "mock_bluetooth_adapters"
)
async def test_options_flow_remote_adapter(hass: HomeAssistant) -> None:
"""Test options are not available for remote adapters."""
source_entry = MockConfigEntry(
domain="test",
)
source_entry.add_to_hass(hass)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_SOURCE: "BB:BB:BB:BB:BB:BB",
CONF_SOURCE_DOMAIN: "test",
CONF_SOURCE_MODEL: "test",
CONF_SOURCE_CONFIG_ENTRY_ID: source_entry.entry_id,
},
options={},
unique_id="BB:BB:BB:BB:BB:BB",
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
result = await hass.config_entries.options.async_init(entry.entry_id)
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "remote_adapters_not_supported"
@pytest.mark.usefixtures("one_adapter")
async def test_async_step_user_linux_adapter_is_ignored(hass: HomeAssistant) -> None:
"""Test we give a hint that the adapter is ignored."""
@ -467,3 +504,49 @@ async def test_async_step_user_linux_adapter_is_ignored(hass: HomeAssistant) ->
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "no_adapters"
assert result["description_placeholders"] == {"ignored_adapters": "1"}
@pytest.mark.usefixtures("enable_bluetooth")
async def test_async_step_integration_discovery_remote_adapter(
hass: HomeAssistant,
) -> None:
"""Test remote adapter configuration via integration discovery."""
entry = MockConfigEntry(domain="test")
connector = (
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
)
scanner = FakeRemoteScanner("esp32", "esp32", connector, True)
manager = _get_manager()
cancel_scanner = manager.async_register_scanner(scanner)
entry.add_to_hass(hass)
with (
patch("homeassistant.components.bluetooth.async_setup", return_value=True),
patch(
"homeassistant.components.bluetooth.async_setup_entry", return_value=True
) as mock_setup_entry,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_INTEGRATION_DISCOVERY},
data={
CONF_SOURCE: scanner.source,
CONF_SOURCE_DOMAIN: "test",
CONF_SOURCE_MODEL: "test",
CONF_SOURCE_CONFIG_ENTRY_ID: entry.entry_id,
},
)
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == "esp32"
assert result["data"] == {
CONF_SOURCE: scanner.source,
CONF_SOURCE_DOMAIN: "test",
CONF_SOURCE_MODEL: "test",
CONF_SOURCE_CONFIG_ENTRY_ID: entry.entry_id,
}
assert len(mock_setup_entry.mock_calls) == 1
await hass.async_block_till_done()
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
cancel_scanner()
await hass.async_block_till_done()

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

@ -18,6 +18,7 @@ from homeassistant.components.bluetooth import (
BluetoothChange,
BluetoothScanningMode,
BluetoothServiceInfo,
HaBluetoothConnector,
async_process_advertisements,
async_rediscover_address,
async_track_unavailable,
@ -25,6 +26,10 @@ from homeassistant.components.bluetooth import (
from homeassistant.components.bluetooth.const import (
BLUETOOTH_DISCOVERY_COOLDOWN_SECONDS,
CONF_PASSIVE,
CONF_SOURCE,
CONF_SOURCE_CONFIG_ENTRY_ID,
CONF_SOURCE_DOMAIN,
CONF_SOURCE_MODEL,
DOMAIN,
LINUX_FIRMWARE_LOAD_FALLBACK_SECONDS,
SOURCE_LOCAL,
@ -47,7 +52,9 @@ from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from . import (
FakeRemoteScanner,
FakeScanner,
MockBleakClient,
_get_manager,
async_setup_with_default_adapter,
async_setup_with_one_adapter,
@ -3263,3 +3270,33 @@ async def test_title_updated_if_mac_address(
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.title == "ACME Bluetooth Adapter 5.0 (00:00:00:00:00:01)"
@pytest.mark.usefixtures("enable_bluetooth")
async def test_cleanup_orphened_remote_scanner_config_entry(
hass: HomeAssistant,
) -> None:
"""Test the remote scanner config entries get cleaned up when orphened."""
connector = (
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
)
scanner = FakeRemoteScanner("esp32", "esp32", connector, True)
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_SOURCE: scanner.source,
CONF_SOURCE_DOMAIN: "test",
CONF_SOURCE_MODEL: "test",
CONF_SOURCE_CONFIG_ENTRY_ID: "no_longer_exists",
},
unique_id=scanner.source,
)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
# Orphened remote scanner config entry should be cleaned up
assert not hass.config_entries.async_entry_for_domain_unique_id(
"bluetooth", scanner.source
)