evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Restore remote discovered devices between remote scanner restarts (#83699)

Browse source
This commit is contained in:
J. Nick Koston 2022-12-11 09:02:55 -10:00 • committed by GitHub
parent fbab7413a5
commit 9008006ac8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
19 changed files with 3323 additions and 110 deletions
Download patch file Download diff file Expand all files Collapse all files

74
tests/components/bluetooth/test_manager.py
View file

@ -8,10 +8,12 @@ from bluetooth_adapters import AdvertisementHistory
import pytest
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import storage
from homeassistant.components.bluetooth.manager import (
FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.json import json_loads
from homeassistant.setup import async_setup_component
from . import (
@ -22,6 +24,8 @@ from . import (
inject_advertisement_with_time_and_source_connectable,
)
from tests.common import load_fixture
@pytest.fixture
def register_hci0_scanner(hass: HomeAssistant) -> None:
@ -282,6 +286,76 @@ async def test_restore_history_from_dbus(hass, one_adapter):
assert bluetooth.async_ble_device_from_address(hass, address) is ble_device
async def test_restore_history_from_dbus_and_remote_adapters(
hass, one_adapter, hass_storage
):
"""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(
load_fixture("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 = BLEDevice(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 not None
assert (
bluetooth.async_ble_device_from_address(hass, "EB:0B:36:35:6F:A4") is not None
)
async def test_restore_history_from_dbus_and_corrupted_remote_adapters(
hass, one_adapter, hass_storage
):
"""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(
load_fixture("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 = BLEDevice(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 not None
assert bluetooth.async_ble_device_from_address(hass, "EB:0B:36:35:6F:A4") is None
async def test_switching_adapters_based_on_rssi_connectable_to_non_connectable(
hass, enable_bluetooth, register_hci0_scanner, register_hci1_scanner
):