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

Improve remote bluetooth scanner diagnostics and add missing test cover (#83796)

Browse Source
This commit is contained in:
J. Nick Koston 2022-12-11 21:33:30 -10:00 committed by GitHub
parent ec47f7b6ff
commit 80a8d5443d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 283 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
homeassistant/components/bluetooth/base_scanner.py
View File

@ -46,17 +46,25 @@ class BaseHaScanner(ABC):
"_connecting",
"name",
"scanning",
"_last_detection",
)
def __init__(self, hass: HomeAssistant, source: str, adapter: str) -> None:
def __init__(
self,
hass: HomeAssistant,
source: str,
adapter: str,
connector: HaBluetoothConnector | None = None,
) -> None:
"""Initialize the scanner."""
self.hass = hass
self.connectable = False
self.source = source
self.connector: HaBluetoothConnector | None = None
self.connector = connector
self._connecting = 0
self.name = adapter_human_name(adapter, source) if adapter != source else source
self.scanning = True
self._last_detection = 0.0
@contextmanager
def connecting(self) -> Generator[None, None, None]:
@ -84,7 +92,12 @@ class BaseHaScanner(ABC):
async def async_diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the scanner."""
return {
"name": self.name,
"source": self.source,
"scanning": self.scanning,
"type": self.__class__.__name__,
"last_detection": self._last_detection,
"monotonic_time": MONOTONIC_TIME(),
"discovered_devices_and_advertisement_data": [
{
"name": device_adv[0].name,
@ -120,14 +133,13 @@ class BaseHaRemoteScanner(BaseHaScanner):
connectable: bool,
) -> None:
"""Initialize the scanner."""
super().__init__(hass, scanner_id, name)
super().__init__(hass, scanner_id, name, connector)
self._new_info_callback = new_info_callback
self._discovered_device_advertisement_datas: dict[
str, tuple[BLEDevice, AdvertisementData]
] = {}
self._discovered_device_timestamps: dict[str, float] = {}
self.connectable = connectable
self.connector = connector
self._details: dict[str, str | HaBluetoothConnector] = {"source": scanner_id}
self._expire_seconds = FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS
assert models.MANAGER is not None
@ -216,6 +228,7 @@ class BaseHaRemoteScanner(BaseHaScanner):
) -> None:
"""Call the registered callback."""
now = MONOTONIC_TIME()
self._last_detection = now
if prev_discovery := self._discovered_device_advertisement_datas.get(address):
# Merge the new data with the old data
# to function the same as BlueZ which
@ -278,16 +291,15 @@ class BaseHaRemoteScanner(BaseHaScanner):
async def async_diagnostics(self) -> dict[str, Any]:
"""Return diagnostic information about the scanner."""
now = MONOTONIC_TIME()
return await super().async_diagnostics() | {
"type": self.__class__.__name__,
"discovered_devices_and_advertisement_data": [
{
"name": device_adv[0].name,
"address": device_adv[0].address,
"rssi": device_adv[0].rssi,
"advertisement_data": device_adv[1],
"details": device_adv[0].details,
}
for device_adv in self.discovered_devices_and_advertisement_data.values()
],
"storage": self._storage.async_get_advertisement_history_as_dict(
self.source
),
"connectable": self.connectable,
"discovered_device_timestamps": self._discovered_device_timestamps,
"time_since_last_device_detection": {
address: now - timestamp
for address, timestamp in self._discovered_device_timestamps.items()
},
}

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

@ -137,7 +137,6 @@ class HaScanner(BaseHaScanner):
self.adapter = adapter
self._start_stop_lock = asyncio.Lock()
self._cancel_watchdog: CALLBACK_TYPE | None = None
self._last_detection = 0.0
self._start_time = 0.0
self._new_info_callback = new_info_callback
self.scanning = False
@ -166,9 +165,6 @@ class HaScanner(BaseHaScanner):
base_diag = await super().async_diagnostics()
return base_diag | {
"adapter": self.adapter,
"source": self.source,
"name": self.name,
"last_detection": self._last_detection,
"start_time": self._start_time,
}

8
homeassistant/components/bluetooth/storage.py
View File

@ -3,6 +3,7 @@ from __future__ import annotations
from bluetooth_adapters import (
DiscoveredDeviceAdvertisementData,
DiscoveredDeviceAdvertisementDataDict,
DiscoveryStorageType,
discovered_device_advertisement_data_from_dict,
discovered_device_advertisement_data_to_dict,
@ -45,6 +46,13 @@ class BluetoothStorage:
return None
return discovered_device_advertisement_data_from_dict(scanner_data)
@callback
def async_get_advertisement_history_as_dict(
self, scanner: str
) -> DiscoveredDeviceAdvertisementDataDict | None:
"""Get discovered devices by scanner as a dict."""
return self._data.get(scanner)
@callback
def _async_get_data(self) -> DiscoveryStorageType:
"""Get data to save to disk."""

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

@ -3,12 +3,18 @@
from unittest.mock import ANY, patch
from bleak.backends.scanner import BLEDevice
from bleak.backends.scanner import AdvertisementData, BLEDevice
from bluetooth_adapters import DEFAULT_ADDRESS
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import BaseHaRemoteScanner, HaBluetoothConnector
from . import generate_advertisement_data, inject_advertisement
from . import (
MockBleakClient,
_get_manager,
generate_advertisement_data,
inject_advertisement,
)
from tests.common import MockConfigEntry
from tests.components.diagnostics import get_diagnostics_for_config_entry
@ -153,13 +159,15 @@ async def test_diagnostics(
-127,
[[]],
],
"details": None,
"name": "x",
"rssi": -60,
"details": None,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "hci0 (00:00:00:00:00:01)",
"scanning": True,
"source": "00:00:00:00:00:01",
"start_time": ANY,
"type": "HaScanner",
@ -178,13 +186,15 @@ async def test_diagnostics(
-127,
[[]],
],
"details": None,
"name": "x",
"rssi": -60,
"details": None,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "hci0 (00:00:00:00:00:01)",
"scanning": True,
"source": "00:00:00:00:00:01",
"start_time": ANY,
"type": "HaScanner",
@ -203,13 +213,15 @@ async def test_diagnostics(
-127,
[[]],
],
"details": None,
"name": "x",
"rssi": -60,
"details": None,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "hci1 (00:00:00:00:00:02)",
"scanning": True,
"source": "00:00:00:00:00:02",
"start_time": ANY,
"type": "HaScanner",
@ -222,7 +234,7 @@ async def test_diagnostics(
async def test_diagnostics_macos(
hass, hass_client, mock_bleak_scanner_start, mock_bluetooth_adapters, macos_adapter
):
"""Test we can setup and unsetup bluetooth with multiple adapters."""
"""Test diagnostics for macos."""
# Normally we do not want to patch our classes, but since bleak will import
# a different scanner based on the operating system, we need to patch here
# because we cannot import the scanner class directly without it throwing an
@ -367,13 +379,15 @@ async def test_diagnostics_macos(
-127,
[[]],
],
"details": None,
"name": "x",
"rssi": -60,
"details": None,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "Core Bluetooth",
"scanning": True,
"source": "Core Bluetooth",
"start_time": ANY,
"type": "HaScanner",
@ -381,3 +395,219 @@ async def test_diagnostics_macos(
],
},
}
async def test_diagnostics_remote_adapter(
hass,
hass_client,
mock_bleak_scanner_start,
mock_bluetooth_adapters,
enable_bluetooth,
one_adapter,
):
"""Test diagnostics for remote adapter."""
manager = _get_manager()
switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
switchbot_adv = generate_advertisement_data(
local_name="wohand", service_uuids=[], manufacturer_data={1: b"\x01"}
)
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"},
)
with patch(
"homeassistant.components.bluetooth.diagnostics.platform.system",
return_value="Linux",
), patch(
"homeassistant.components.bluetooth.diagnostics.get_dbus_managed_objects",
return_value={},
):
entry1 = MockConfigEntry(
domain=bluetooth.DOMAIN, data={}, unique_id="00:00:00:00:00:01"
)
entry1.add_to_hass(hass)
assert await hass.config_entries.async_setup(entry1.entry_id)
await hass.async_block_till_done()
new_info_callback = manager.scanner_adv_received
connector = (
HaBluetoothConnector(MockBleakClient, "mock_bleak_client", lambda: False),
)
scanner = FakeScanner(
hass, "esp32", "esp32", new_info_callback, connector, False
)
unsetup = scanner.async_setup()
cancel = manager.async_register_scanner(scanner, True)
scanner.inject_advertisement(switchbot_device, switchbot_adv)
inject_advertisement(hass, switchbot_device, switchbot_adv)
diag = await get_diagnostics_for_config_entry(hass, hass_client, entry1)
assert diag == {
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",
"hw_version": "usb:v1D6Bp0246d053F",
"manufacturer": "ACME",
"passive_scan": False,
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"sw_version": "homeassistant",
"vendor_id": "cc01",
}
},
"dbus": {},
"manager": {
"adapters": {
"hci0": {
"address": "00:00:00:00:00:01",
"hw_version": "usb:v1D6Bp0246d053F",
"manufacturer": "ACME",
"passive_scan": False,
"product": "Bluetooth Adapter 5.0",
"product_id": "aa01",
"sw_version": "homeassistant",
"vendor_id": "cc01",
}
},
"advertisement_tracker": {
"intervals": {},
"sources": {"44:44:33:11:23:45": "esp32"},
"timings": {"44:44:33:11:23:45": [ANY]},
},
"all_history": [
{
"address": "44:44:33:11:23:45",
"advertisement": [
"wohand",
{"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}},
{},
[],
-127,
-127,
[],
],
"connectable": False,
"device": {
"__type": "<class " "'bleak.backends.device.BLEDevice'>",
"repr": "BLEDevice(44:44:33:11:23:45, " "wohand)",
},
"manufacturer_data": {
"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}
},
"name": "wohand",
"rssi": -127,
"service_data": {},
"service_uuids": [],
"source": "esp32",
"time": ANY,
}
],
"connectable_history": [
{
"address": "44:44:33:11:23:45",
"advertisement": [
"wohand",
{"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}},
{},
[],
-127,
-127,
[[]],
],
"connectable": True,
"device": {
"__type": "<class " "'bleak.backends.device.BLEDevice'>",
"repr": "BLEDevice(44:44:33:11:23:45, " "wohand)",
},
"manufacturer_data": {
"1": {"__type": "<class " "'bytes'>", "repr": "b'\\x01'"}
},
"name": "wohand",
"rssi": -127,
"service_data": {},
"service_uuids": [],
"source": "local",
"time": ANY,
}
],
"scanners": [
{
"adapter": "hci0",
"discovered_devices_and_advertisement_data": [],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "hci0 (00:00:00:00:00:01)",
"scanning": True,
"source": "00:00:00:00:00:01",
"start_time": ANY,
"type": "HaScanner",
},
{
"adapter": "hci0",
"discovered_devices_and_advertisement_data": [],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "hci0 (00:00:00:00:00:01)",
"scanning": True,
"source": "00:00:00:00:00:01",
"start_time": ANY,
"type": "HaScanner",
},
{
"connectable": False,
"discovered_device_timestamps": {"44:44:33:11:23:45": ANY},
"time_since_last_device_detection": {"44:44:33:11:23:45": ANY},
"discovered_devices_and_advertisement_data": [
{
"address": "44:44:33:11:23:45",
"advertisement_data": [
"wohand",
{
"1": {
"__type": "<class " "'bytes'>",
"repr": "b'\\x01'",
}
},
{},
[],
-127,
-127,
[],
],
"details": {
"scanner_specific_data": "test",
"source": "esp32",
},
"name": "wohand",
"rssi": -127,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "esp32",
"scanning": True,
"source": "esp32",
"storage": None,
"type": "FakeScanner",
},
],
},
}
cancel()
unsetup()

11
tests/components/shelly/test_diagnostics.py
View File

@ -1,4 +1,6 @@
"""Tests for Shelly diagnostics platform."""
from unittest.mock import ANY
from aiohttp import ClientSession
from aioshelly.ble.const import BLE_SCAN_RESULT_EVENT
@ -92,6 +94,8 @@ async def test_rpc_config_entry_diagnostics(
"entry": entry_dict,
"bluetooth": {
"scanner": {
"connectable": False,
"discovered_device_timestamps": {"AA:BB:CC:DD:EE:FF": ANY},
"discovered_devices_and_advertisement_data": [
{
"address": "AA:BB:CC:DD:EE:FF",
@ -119,6 +123,13 @@ async def test_rpc_config_entry_diagnostics(
"rssi": -62,
}
],
"last_detection": ANY,
"monotonic_time": ANY,
"name": "Mock Title (12:34:56:78:9A:BC)",
"scanning": True,
"source": "12:34:56:78:9A:BC",
"storage": None,
"time_since_last_device_detection": {"AA:BB:CC:DD:EE:FF": ANY},
"type": "ShellyBLEScanner",
}
},