Fix bluetooth_le_tracker reporting devices Home when they leave (#90641)

* fix bluetooth_le_tracker reporting devices Home when they leave

* refactor

* implement tests for BLE service_info.time check

* update bluetooth_le_tracker tests

* tweaks

---------

Co-authored-by: J. Nick Koston <nick@koston.org>
This commit is contained in:
Fabio De Simone 2023-04-05 02:59:57 +02:00 committed by GitHub
parent 03caf63ec2
commit 8495da1af0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 156 additions and 15 deletions

View file

@ -70,6 +70,7 @@ async def async_setup_scanner( # noqa: C901
yaml_path = hass.config.path(YAML_DEVICES)
devs_to_track: set[str] = set()
devs_no_track: set[str] = set()
devs_advertise_time: dict[str, float] = {}
devs_track_battery = {}
interval: timedelta = config.get(CONF_SCAN_INTERVAL, SCAN_INTERVAL)
# if track new devices is true discover new devices
@ -178,6 +179,7 @@ async def async_setup_scanner( # noqa: C901
"""Update from a ble callback."""
mac = service_info.address
if mac in devs_to_track:
devs_advertise_time[mac] = service_info.time
now = dt_util.utcnow()
hass.async_create_task(async_see_device(mac, service_info.name))
if (
@ -205,7 +207,9 @@ async def async_setup_scanner( # noqa: C901
# there have been no callbacks because the RSSI or
# other properties have not changed.
for service_info in bluetooth.async_discovered_service_info(hass, False):
_async_update_ble(service_info, bluetooth.BluetoothChange.ADVERTISEMENT)
# Only call _async_update_ble if the advertisement time has changed
if service_info.time != devs_advertise_time.get(service_info.address):
_async_update_ble(service_info, bluetooth.BluetoothChange.ADVERTISEMENT)
cancels = [
bluetooth.async_register_callback(

View file

@ -4,6 +4,7 @@ from datetime import timedelta
from unittest.mock import patch
from bleak import BleakError
from freezegun import freeze_time
from homeassistant.components.bluetooth import BluetoothServiceInfoBleak
from homeassistant.components.bluetooth_le_tracker import device_tracker
@ -12,6 +13,7 @@ from homeassistant.components.bluetooth_le_tracker.device_tracker import (
CONF_TRACK_BATTERY_INTERVAL,
)
from homeassistant.components.device_tracker import (
CONF_CONSIDER_HOME,
CONF_SCAN_INTERVAL,
CONF_TRACK_NEW,
DOMAIN,
@ -64,6 +66,150 @@ class MockBleakClientBattery5(MockBleakClient):
return b"\x05"
async def test_do_not_see_device_if_time_not_updated(
hass: HomeAssistant,
mock_bluetooth: None,
mock_device_tracker_conf: list[legacy.Device],
) -> None:
"""Test device going not_home after consider_home threshold from first scan if the subsequent scans have not incremented last seen time."""
address = "DE:AD:BE:EF:13:37"
name = "Mock device name"
entity_id = f"{DOMAIN}.{slugify(name)}"
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,
rssi=-19,
manufacturer_data={},
service_data={},
service_uuids=[],
source="local",
device=generate_ble_device(address, None),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
# Return with name with time = 0 for all the updates
mock_async_discovered_service_info.return_value = [device]
config = {
CONF_PLATFORM: "bluetooth_le_tracker",
CONF_SCAN_INTERVAL: timedelta(minutes=1),
CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(minutes=10),
}
result = await async_setup_component(hass, DOMAIN, {DOMAIN: config})
assert result
# Tick until device seen enough times for to be registered for tracking
for _ in range(device_tracker.MIN_SEEN_NEW):
async_fire_time_changed(
hass,
dt_util.utcnow() + config[CONF_SCAN_INTERVAL] + timedelta(seconds=1),
)
await hass.async_block_till_done()
# Advance time to trigger updates
time_after_consider_home = dt_util.utcnow() + config[CONF_CONSIDER_HOME] / 2
with freeze_time(time_after_consider_home):
async_fire_time_changed(hass, time_after_consider_home)
await hass.async_block_till_done()
# Advance time over the consider home threshold and trigger update after the threshold
time_after_consider_home = dt_util.utcnow() + config[CONF_CONSIDER_HOME]
with freeze_time(time_after_consider_home):
async_fire_time_changed(hass, time_after_consider_home)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "not_home"
async def test_see_device_if_time_updated(
hass: HomeAssistant,
mock_bluetooth: None,
mock_device_tracker_conf: list[legacy.Device],
) -> None:
"""Test device remaining home after consider_home threshold from first scan if the subsequent scans have incremented last seen time."""
address = "DE:AD:BE:EF:13:37"
name = "Mock device name"
entity_id = f"{DOMAIN}.{slugify(name)}"
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,
rssi=-19,
manufacturer_data={},
service_data={},
service_uuids=[],
source="local",
device=generate_ble_device(address, None),
advertisement=generate_advertisement_data(local_name="empty"),
time=0,
connectable=False,
)
# Return with name with time = 0 initially
mock_async_discovered_service_info.return_value = [device]
config = {
CONF_PLATFORM: "bluetooth_le_tracker",
CONF_SCAN_INTERVAL: timedelta(minutes=1),
CONF_TRACK_NEW: True,
CONF_CONSIDER_HOME: timedelta(minutes=10),
}
result = await async_setup_component(hass, DOMAIN, {DOMAIN: config})
assert result
# Tick until device seen enough times for to be registered for tracking
for _ in range(device_tracker.MIN_SEEN_NEW):
async_fire_time_changed(
hass,
dt_util.utcnow() + config[CONF_SCAN_INTERVAL] + timedelta(seconds=1),
)
await hass.async_block_till_done()
# Increment device time so it gets seen in the next update
device = BluetoothServiceInfoBleak(
name=name,
address=address,
rssi=-19,
manufacturer_data={},
service_data={},
service_uuids=[],
source="local",
device=generate_ble_device(address, None),
advertisement=generate_advertisement_data(local_name="empty"),
time=1,
connectable=False,
)
# Return with name with time = 0 initially
mock_async_discovered_service_info.return_value = [device]
# Advance time to trigger updates
time_after_consider_home = dt_util.utcnow() + config[CONF_CONSIDER_HOME] / 2
with freeze_time(time_after_consider_home):
async_fire_time_changed(hass, time_after_consider_home)
await hass.async_block_till_done()
# Advance time over the consider home threshold and trigger update after the threshold
time_after_consider_home = dt_util.utcnow() + config[CONF_CONSIDER_HOME]
with freeze_time(time_after_consider_home):
async_fire_time_changed(hass, time_after_consider_home)
await hass.async_block_till_done()
state = hass.states.get(entity_id)
assert state
assert state.state == "home"
async def test_preserve_new_tracked_device_name(
hass: HomeAssistant,
mock_bluetooth: None,
@ -77,9 +223,7 @@ async def test_preserve_new_tracked_device_name(
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info, patch.object(
device_tracker, "MIN_SEEN_NEW", 3
):
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,
@ -101,8 +245,7 @@ async def test_preserve_new_tracked_device_name(
CONF_SCAN_INTERVAL: timedelta(minutes=1),
CONF_TRACK_NEW: True,
}
result = await async_setup_component(hass, DOMAIN, {DOMAIN: config})
assert result
assert await async_setup_component(hass, DOMAIN, {DOMAIN: config})
# Seen once here; return without name when seen subsequent times
device = BluetoothServiceInfoBleak(
@ -147,9 +290,7 @@ async def test_tracking_battery_times_out(
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info, patch.object(
device_tracker, "MIN_SEEN_NEW", 3
):
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,
@ -216,9 +357,7 @@ async def test_tracking_battery_fails(
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info, patch.object(
device_tracker, "MIN_SEEN_NEW", 3
):
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,
@ -285,9 +424,7 @@ async def test_tracking_battery_successful(
with patch(
"homeassistant.components.bluetooth.async_discovered_service_info"
) as mock_async_discovered_service_info, patch.object(
device_tracker, "MIN_SEEN_NEW", 3
):
) as mock_async_discovered_service_info:
device = BluetoothServiceInfoBleak(
name=name,
address=address,