fix patching time

This commit is contained in:
J. Nick Koston
2023-12-05 13:07:51 -10:00
parent 1008165b54
commit a1bc9a239f
20 changed files with 129 additions and 194 deletions

View File

@@ -21,6 +21,7 @@ from bluetooth_adapters import (
adapter_unique_name, adapter_unique_name,
get_adapters, get_adapters,
) )
from bluetooth_data_tools import monotonic_time_coarse as MONOTONIC_TIME
from habluetooth import ( from habluetooth import (
BaseHaScanner, BaseHaScanner,
BluetoothScannerDevice, BluetoothScannerDevice,
@@ -80,7 +81,7 @@ from .const import (
LINUX_FIRMWARE_LOAD_FALLBACK_SECONDS, LINUX_FIRMWARE_LOAD_FALLBACK_SECONDS,
SOURCE_LOCAL, SOURCE_LOCAL,
) )
from .manager import MONOTONIC_TIME, HomeAssistantBluetoothManager from .manager import HomeAssistantBluetoothManager
from .match import BluetoothCallbackMatcher, IntegrationMatcher from .match import BluetoothCallbackMatcher, IntegrationMatcher
from .models import BluetoothCallback, BluetoothChange from .models import BluetoothCallback, BluetoothChange
from .storage import BluetoothStorage from .storage import BluetoothStorage

View File

@@ -4,11 +4,10 @@ from __future__ import annotations
from collections.abc import Callable, Iterable from collections.abc import Callable, Iterable
import itertools import itertools
import logging import logging
from typing import Final
from bleak_retry_connector import BleakSlotManager from bleak_retry_connector import BleakSlotManager
from bluetooth_adapters import BluetoothAdapters from bluetooth_adapters import BluetoothAdapters
from habluetooth import BluetoothManager, manager from habluetooth import BluetoothManager
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.const import EVENT_LOGGING_CHANGED from homeassistant.const import EVENT_LOGGING_CHANGED
@@ -34,8 +33,6 @@ from .models import BluetoothCallback, BluetoothChange, BluetoothServiceInfoBlea
from .storage import BluetoothStorage from .storage import BluetoothStorage
from .util import async_load_history_from_system from .util import async_load_history_from_system
MONOTONIC_TIME: Final = manager.MONOTONIC_TIME
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)

View File

@@ -3,9 +3,8 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from enum import Enum from enum import Enum
from typing import TYPE_CHECKING, Final from typing import TYPE_CHECKING
from bluetooth_data_tools import monotonic_time_coarse
from home_assistant_bluetooth import BluetoothServiceInfoBleak from home_assistant_bluetooth import BluetoothServiceInfoBleak
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -14,8 +13,6 @@ if TYPE_CHECKING:
MANAGER: BluetoothManager | None = None MANAGER: BluetoothManager | None = None
MONOTONIC_TIME: Final = monotonic_time_coarse
BluetoothChange = Enum("BluetoothChange", "ADVERTISEMENT") BluetoothChange = Enum("BluetoothChange", "ADVERTISEMENT")
BluetoothCallback = Callable[[BluetoothServiceInfoBleak, BluetoothChange], None] BluetoothCallback = Callable[[BluetoothServiceInfoBleak, BluetoothChange], None]

View File

@@ -5,12 +5,12 @@ from contextlib import contextmanager
import itertools import itertools
import time import time
from typing import Any from typing import Any
from unittest.mock import MagicMock from unittest.mock import MagicMock, patch
from bleak import BleakClient from bleak import BleakClient
from bleak.backends.scanner import AdvertisementData, BLEDevice from bleak.backends.scanner import AdvertisementData, BLEDevice
from bluetooth_adapters import DEFAULT_ADDRESS from bluetooth_adapters import DEFAULT_ADDRESS
from habluetooth import BaseHaScanner, BluetoothManager from habluetooth import BaseHaScanner, BluetoothManager, get_manager
from homeassistant.components.bluetooth import ( from homeassistant.components.bluetooth import (
DOMAIN, DOMAIN,
@@ -18,7 +18,6 @@ from homeassistant.components.bluetooth import (
BluetoothServiceInfo, BluetoothServiceInfo,
BluetoothServiceInfoBleak, BluetoothServiceInfoBleak,
async_get_advertisement_callback, async_get_advertisement_callback,
models,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
@@ -36,6 +35,7 @@ __all__ = (
"generate_advertisement_data", "generate_advertisement_data",
"generate_ble_device", "generate_ble_device",
"MockBleakClient", "MockBleakClient",
"patch_bluetooth_time",
) )
ADVERTISEMENT_DATA_DEFAULTS = { ADVERTISEMENT_DATA_DEFAULTS = {
@@ -55,6 +55,19 @@ BLE_DEVICE_DEFAULTS = {
} }
@contextmanager
def patch_bluetooth_time(mock_time: float) -> None:
"""Patch the bluetooth time."""
with patch(
"homeassistant.components.bluetooth.MONOTONIC_TIME", return_value=mock_time
), patch(
"habluetooth.base_scanner.monotonic_time_coarse", return_value=mock_time
), patch(
"habluetooth.manager.monotonic_time_coarse", return_value=mock_time
), patch("habluetooth.scanner.monotonic_time_coarse", return_value=mock_time):
yield
def generate_advertisement_data(**kwargs: Any) -> AdvertisementData: def generate_advertisement_data(**kwargs: Any) -> AdvertisementData:
"""Generate advertisement data with defaults.""" """Generate advertisement data with defaults."""
new = kwargs.copy() new = kwargs.copy()
@@ -87,7 +100,7 @@ def generate_ble_device(
def _get_manager() -> BluetoothManager: def _get_manager() -> BluetoothManager:
"""Return the bluetooth manager.""" """Return the bluetooth manager."""
return models.MANAGER return get_manager()
def inject_advertisement( def inject_advertisement(

View File

@@ -1,7 +1,6 @@
"""Tests for the Bluetooth integration advertisement tracking.""" """Tests for the Bluetooth integration advertisement tracking."""
from datetime import timedelta from datetime import timedelta
import time import time
from unittest.mock import patch
from habluetooth.advertisement_tracker import ADVERTISING_TIMES_NEEDED from habluetooth.advertisement_tracker import ADVERTISING_TIMES_NEEDED
import pytest import pytest
@@ -25,6 +24,7 @@ from . import (
generate_ble_device, generate_ble_device,
inject_advertisement_with_time_and_source, inject_advertisement_with_time_and_source,
inject_advertisement_with_time_and_source_connectable, inject_advertisement_with_time_and_source_connectable,
patch_bluetooth_time,
) )
from tests.common import async_fire_time_changed from tests.common import async_fire_time_changed
@@ -70,9 +70,8 @@ async def test_advertisment_interval_shorter_than_adapter_stack_timeout(
) )
monotonic_now = start_monotonic_time + ((ADVERTISING_TIMES_NEEDED - 1) * 2) monotonic_now = start_monotonic_time + ((ADVERTISING_TIMES_NEEDED - 1) * 2)
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -123,9 +122,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_connectab
monotonic_now = start_monotonic_time + ( monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS (ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
) )
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -189,9 +187,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
monotonic_now = start_monotonic_time + ( monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS (ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
) )
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -245,9 +242,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_not_conne
monotonic_now = start_monotonic_time + ( monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS (ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
) )
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -321,9 +317,8 @@ async def test_advertisment_interval_shorter_than_adapter_stack_timeout_adapter_
monotonic_now = start_monotonic_time + ( monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS (ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
) )
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -402,9 +397,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
monotonic_now = start_monotonic_time + ( monotonic_now = start_monotonic_time + (
(ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS (ADVERTISING_TIMES_NEEDED - 1) * ONE_HOUR_SECONDS
) )
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -415,9 +409,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
cancel_scanner() cancel_scanner()
# Now that the scanner is gone we should go back to the stack default timeout # Now that the scanner is gone we should go back to the stack default timeout
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -427,9 +420,8 @@ async def test_advertisment_interval_longer_than_adapter_stack_timeout_adapter_c
assert switchbot_device_went_unavailable is False assert switchbot_device_went_unavailable is False
# Now that the scanner is gone we should go back to the stack default timeout # Now that the scanner is gone we should go back to the stack default timeout
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
@@ -484,9 +476,8 @@ async def test_advertisment_interval_longer_increasing_than_adapter_stack_timeou
) )
monotonic_now = start_monotonic_time + UNAVAILABLE_TRACK_SECONDS + 1 monotonic_now = start_monotonic_time + UNAVAILABLE_TRACK_SECONDS + 1
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)

View File

@@ -35,6 +35,7 @@ from . import (
_get_manager, _get_manager,
generate_advertisement_data, generate_advertisement_data,
generate_ble_device, generate_ble_device,
patch_bluetooth_time,
) )
from tests.common import async_fire_time_changed, load_fixture from tests.common import async_fire_time_changed, load_fixture
@@ -212,10 +213,7 @@ async def test_remote_scanner_expires_connectable(
expire_utc = dt_util.utcnow() + timedelta( expire_utc = dt_util.utcnow() + timedelta(
seconds=CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 seconds=CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
) )
with patch( with patch_bluetooth_time(expire_monotonic):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=expire_monotonic,
):
async_fire_time_changed(hass, expire_utc) async_fire_time_changed(hass, expire_utc)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -295,10 +293,7 @@ async def test_remote_scanner_expires_non_connectable(
expire_utc = dt_util.utcnow() + timedelta( expire_utc = dt_util.utcnow() + timedelta(
seconds=CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 seconds=CONNECTABLE_FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
) )
with patch( with patch_bluetooth_time(expire_monotonic):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=expire_monotonic,
):
async_fire_time_changed(hass, expire_utc) async_fire_time_changed(hass, expire_utc)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -311,10 +306,7 @@ async def test_remote_scanner_expires_non_connectable(
expire_utc = dt_util.utcnow() + timedelta( expire_utc = dt_util.utcnow() + timedelta(
seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
) )
with patch( with patch_bluetooth_time(expire_monotonic):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=expire_monotonic,
):
async_fire_time_changed(hass, expire_utc) async_fire_time_changed(hass, expire_utc)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -512,10 +504,7 @@ async def test_device_with_ten_minute_advertising_interval(
connectable=False, connectable=False,
) )
with patch( with patch_bluetooth_time(new_time):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=new_time,
):
scanner.inject_advertisement(bparasite_device, bparasite_device_adv) scanner.inject_advertisement(bparasite_device, bparasite_device_adv)
original_device = scanner.discovered_devices_and_advertisement_data[ original_device = scanner.discovered_devices_and_advertisement_data[
@@ -525,10 +514,7 @@ async def test_device_with_ten_minute_advertising_interval(
for _ in range(1, 20): for _ in range(1, 20):
new_time += advertising_interval new_time += advertising_interval
with patch( with patch_bluetooth_time(new_time):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=new_time,
):
scanner.inject_advertisement(bparasite_device, bparasite_device_adv) scanner.inject_advertisement(bparasite_device, bparasite_device_adv)
# Make sure the BLEDevice object gets updated # Make sure the BLEDevice object gets updated
@@ -543,10 +529,7 @@ async def test_device_with_ten_minute_advertising_interval(
bluetooth.async_address_present(hass, bparasite_device.address, False) is True bluetooth.async_address_present(hass, bparasite_device.address, False) is True
) )
assert bparasite_device_went_unavailable is False assert bparasite_device_went_unavailable is False
with patch( with patch_bluetooth_time(new_time):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=new_time,
):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=future_time)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=future_time))
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -556,13 +539,7 @@ async def test_device_with_ten_minute_advertising_interval(
future_time + advertising_interval + TRACKER_BUFFERING_WOBBLE_SECONDS + 1 future_time + advertising_interval + TRACKER_BUFFERING_WOBBLE_SECONDS + 1
) )
with patch( with patch_bluetooth_time(missed_advertisement_future_time):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=missed_advertisement_future_time,
), patch(
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=missed_advertisement_future_time,
):
# Fire once for the scanner to expire the device # Fire once for the scanner to expire the device
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -626,10 +603,7 @@ async def test_scanner_stops_responding(
+ SCANNER_WATCHDOG_INTERVAL.total_seconds() + SCANNER_WATCHDOG_INTERVAL.total_seconds()
) )
# We hit the timer with no detections, so we reset the adapter and restart the scanner # We hit the timer with no detections, so we reset the adapter and restart the scanner
with patch( with patch_bluetooth_time(failure_reached_time):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=failure_reached_time,
):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -650,10 +624,7 @@ async def test_scanner_stops_responding(
failure_reached_time += 1 failure_reached_time += 1
with patch( with patch_bluetooth_time(failure_reached_time):
"habluetooth.base_scanner.MONOTONIC_TIME",
return_value=failure_reached_time,
):
scanner.inject_advertisement(bparasite_device, bparasite_device_adv) scanner.inject_advertisement(bparasite_device, bparasite_device_adv)
# As soon as we get a detection, we know the scanner is working again # As soon as we get a detection, we know the scanner is working again

View File

@@ -46,6 +46,7 @@ from . import (
inject_advertisement_with_source, inject_advertisement_with_source,
inject_advertisement_with_time_and_source, inject_advertisement_with_time_and_source,
inject_advertisement_with_time_and_source_connectable, inject_advertisement_with_time_and_source_connectable,
patch_bluetooth_time,
) )
from tests.common import async_fire_time_changed, load_fixture from tests.common import async_fire_time_changed, load_fixture
@@ -960,9 +961,8 @@ async def test_goes_unavailable_dismisses_discovery_and_makes_discoverable(
return_value=[{"flow_id": "mock_flow_id"}], return_value=[{"flow_id": "mock_flow_id"}],
) as mock_async_progress_by_init_data_type, patch.object( ) as mock_async_progress_by_init_data_type, patch.object(
hass.config_entries.flow, "async_abort" hass.config_entries.flow, "async_abort"
) as mock_async_abort, patch( ) as mock_async_abort, patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
return_value=monotonic_now + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -1103,9 +1103,8 @@ async def test_set_fallback_interval_small(
) )
monotonic_now = start_monotonic_time + 2 monotonic_now = start_monotonic_time + 2
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -1168,9 +1167,8 @@ async def test_set_fallback_interval_big(
# Check that device hasn't expired after a day # Check that device hasn't expired after a day
monotonic_now = start_monotonic_time + 86400 monotonic_now = start_monotonic_time + 86400
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -1182,9 +1180,8 @@ async def test_set_fallback_interval_big(
# Try again after it has expired # Try again after it has expired
monotonic_now = start_monotonic_time + 604800 monotonic_now = start_monotonic_time + 604800
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now + UNAVAILABLE_TRACK_SECONDS,
return_value=monotonic_now + UNAVAILABLE_TRACK_SECONDS,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)

View File

@@ -22,7 +22,11 @@ from homeassistant.helpers.service_info.bluetooth import BluetoothServiceInfo
from homeassistant.setup import async_setup_component from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
from . import inject_bluetooth_service_info, patch_all_discovered_devices from . import (
inject_bluetooth_service_info,
patch_all_discovered_devices,
patch_bluetooth_time,
)
from tests.common import async_fire_time_changed from tests.common import async_fire_time_changed
@@ -159,10 +163,9 @@ async def test_unavailable_callbacks_mark_the_coordinator_unavailable(
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", [MagicMock(address="44:44:33:11:23:45")]
return_value=monotonic_now, ):
), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()
@@ -176,9 +179,8 @@ async def test_unavailable_callbacks_mark_the_coordinator_unavailable(
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 2 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 2
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now,
return_value=monotonic_now,
), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]): ), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,

View File

@@ -48,6 +48,7 @@ from . import (
inject_bluetooth_service_info, inject_bluetooth_service_info,
inject_bluetooth_service_info_bleak, inject_bluetooth_service_info_bleak,
patch_all_discovered_devices, patch_all_discovered_devices,
patch_bluetooth_time,
) )
from tests.common import ( from tests.common import (
@@ -471,9 +472,8 @@ async def test_unavailable_after_no_data(
assert processor.available is True assert processor.available is True
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now,
return_value=monotonic_now,
), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]): ), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)
@@ -490,9 +490,8 @@ async def test_unavailable_after_no_data(
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 2 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 2
with patch( with patch_bluetooth_time(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", monotonic_now,
return_value=monotonic_now,
), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]): ), patch_all_discovered_devices([MagicMock(address="44:44:33:11:23:45")]):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS) hass, dt_util.utcnow() + timedelta(seconds=UNAVAILABLE_TRACK_SECONDS)

View File

@@ -25,6 +25,7 @@ from . import (
async_setup_with_one_adapter, async_setup_with_one_adapter,
generate_advertisement_data, generate_advertisement_data,
generate_ble_device, generate_ble_device,
patch_bluetooth_time,
) )
from tests.common import MockConfigEntry, async_fire_time_changed from tests.common import MockConfigEntry, async_fire_time_changed
@@ -226,9 +227,8 @@ async def test_recovery_from_dbus_restart(
mock_discovered = [MagicMock()] mock_discovered = [MagicMock()]
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 10,
return_value=start_time_monotonic + 10,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -236,9 +236,8 @@ async def test_recovery_from_dbus_restart(
assert called_start == 1 assert called_start == 1
# Fire a callback to reset the timer # Fire a callback to reset the timer
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic,
return_value=start_time_monotonic,
): ):
_callback( _callback(
generate_ble_device("44:44:33:11:23:42", "any_name"), generate_ble_device("44:44:33:11:23:42", "any_name"),
@@ -246,9 +245,8 @@ async def test_recovery_from_dbus_restart(
) )
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 20,
return_value=start_time_monotonic + 20,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -256,9 +254,8 @@ async def test_recovery_from_dbus_restart(
assert called_start == 1 assert called_start == 1
# We hit the timer, so we restart the scanner # We hit the timer, so we restart the scanner
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + SCANNER_WATCHDOG_TIMEOUT + 20,
return_value=start_time_monotonic + SCANNER_WATCHDOG_TIMEOUT + 20,
): ):
async_fire_time_changed( async_fire_time_changed(
hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL + timedelta(seconds=20) hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL + timedelta(seconds=20)
@@ -301,9 +298,8 @@ async def test_adapter_recovery(hass: HomeAssistant, one_adapter: None) -> None:
scanner = MockBleakScanner() scanner = MockBleakScanner()
start_time_monotonic = time.monotonic() start_time_monotonic = time.monotonic()
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic,
return_value=start_time_monotonic,
), patch( ), patch(
"habluetooth.scanner.OriginalBleakScanner", "habluetooth.scanner.OriginalBleakScanner",
return_value=scanner, return_value=scanner,
@@ -316,9 +312,8 @@ async def test_adapter_recovery(hass: HomeAssistant, one_adapter: None) -> None:
mock_discovered = [MagicMock()] mock_discovered = [MagicMock()]
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 10,
return_value=start_time_monotonic + 10,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -326,9 +321,8 @@ async def test_adapter_recovery(hass: HomeAssistant, one_adapter: None) -> None:
assert called_start == 1 assert called_start == 1
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 20,
return_value=start_time_monotonic + 20,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -336,9 +330,8 @@ async def test_adapter_recovery(hass: HomeAssistant, one_adapter: None) -> None:
assert called_start == 1 assert called_start == 1
# We hit the timer with no detections, so we reset the adapter and restart the scanner # We hit the timer with no detections, so we reset the adapter and restart the scanner
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic
return_value=start_time_monotonic
+ SCANNER_WATCHDOG_TIMEOUT + SCANNER_WATCHDOG_TIMEOUT
+ SCANNER_WATCHDOG_INTERVAL.total_seconds(), + SCANNER_WATCHDOG_INTERVAL.total_seconds(),
), patch( ), patch(
@@ -390,9 +383,8 @@ async def test_adapter_scanner_fails_to_start_first_time(
scanner = MockBleakScanner() scanner = MockBleakScanner()
start_time_monotonic = time.monotonic() start_time_monotonic = time.monotonic()
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic,
return_value=start_time_monotonic,
), patch( ), patch(
"habluetooth.scanner.OriginalBleakScanner", "habluetooth.scanner.OriginalBleakScanner",
return_value=scanner, return_value=scanner,
@@ -405,9 +397,8 @@ async def test_adapter_scanner_fails_to_start_first_time(
mock_discovered = [MagicMock()] mock_discovered = [MagicMock()]
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 10,
return_value=start_time_monotonic + 10,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -415,9 +406,8 @@ async def test_adapter_scanner_fails_to_start_first_time(
assert called_start == 1 assert called_start == 1
# Ensure we don't restart the scanner if we don't need to # Ensure we don't restart the scanner if we don't need to
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic + 20,
return_value=start_time_monotonic + 20,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL) async_fire_time_changed(hass, dt_util.utcnow() + SCANNER_WATCHDOG_INTERVAL)
await hass.async_block_till_done() await hass.async_block_till_done()
@@ -425,9 +415,8 @@ async def test_adapter_scanner_fails_to_start_first_time(
assert called_start == 1 assert called_start == 1
# We hit the timer with no detections, so we reset the adapter and restart the scanner # We hit the timer with no detections, so we reset the adapter and restart the scanner
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic
return_value=start_time_monotonic
+ SCANNER_WATCHDOG_TIMEOUT + SCANNER_WATCHDOG_TIMEOUT
+ SCANNER_WATCHDOG_INTERVAL.total_seconds(), + SCANNER_WATCHDOG_INTERVAL.total_seconds(),
), patch( ), patch(
@@ -441,9 +430,8 @@ async def test_adapter_scanner_fails_to_start_first_time(
# We hit the timer again the previous start call failed, make sure # We hit the timer again the previous start call failed, make sure
# we try again # we try again
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic
return_value=start_time_monotonic
+ SCANNER_WATCHDOG_TIMEOUT + SCANNER_WATCHDOG_TIMEOUT
+ SCANNER_WATCHDOG_INTERVAL.total_seconds(), + SCANNER_WATCHDOG_INTERVAL.total_seconds(),
), patch( ), patch(
@@ -504,9 +492,8 @@ async def test_adapter_fails_to_start_and_takes_a_bit_to_init(
with patch( with patch(
"habluetooth.scanner.ADAPTER_INIT_TIME", "habluetooth.scanner.ADAPTER_INIT_TIME",
0, 0,
), patch( ), patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic,
return_value=start_time_monotonic,
), patch( ), patch(
"habluetooth.scanner.OriginalBleakScanner", "habluetooth.scanner.OriginalBleakScanner",
return_value=scanner, return_value=scanner,
@@ -555,9 +542,8 @@ async def test_restart_takes_longer_than_watchdog_time(
with patch( with patch(
"habluetooth.scanner.ADAPTER_INIT_TIME", "habluetooth.scanner.ADAPTER_INIT_TIME",
0, 0,
), patch( ), patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic,
return_value=start_time_monotonic,
), patch( ), patch(
"habluetooth.scanner.OriginalBleakScanner", "habluetooth.scanner.OriginalBleakScanner",
return_value=scanner, return_value=scanner,
@@ -568,9 +554,8 @@ async def test_restart_takes_longer_than_watchdog_time(
# Now force a recover adapter 2x # Now force a recover adapter 2x
for _ in range(2): for _ in range(2):
with patch( with patch_bluetooth_time(
"habluetooth.base_scanner.MONOTONIC_TIME", start_time_monotonic
return_value=start_time_monotonic
+ SCANNER_WATCHDOG_TIMEOUT + SCANNER_WATCHDOG_TIMEOUT
+ SCANNER_WATCHDOG_INTERVAL.total_seconds(), + SCANNER_WATCHDOG_INTERVAL.total_seconds(),
): ):

View File

@@ -2,7 +2,6 @@
from datetime import timedelta from datetime import timedelta
import logging import logging
import time import time
from unittest.mock import patch
import pytest import pytest
@@ -25,6 +24,7 @@ from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.bluetooth import ( from tests.components.bluetooth import (
inject_bluetooth_service_info, inject_bluetooth_service_info,
patch_all_discovered_devices, patch_all_discovered_devices,
patch_bluetooth_time,
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -236,10 +236,7 @@ async def test_unavailable(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()
@@ -290,10 +287,7 @@ async def test_sleepy_device(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()
@@ -344,10 +338,7 @@ async def test_sleepy_device_restores_state(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()

View File

@@ -2,7 +2,6 @@
from datetime import timedelta from datetime import timedelta
import logging import logging
import time import time
from unittest.mock import patch
import pytest import pytest
@@ -25,6 +24,7 @@ from tests.common import MockConfigEntry, async_fire_time_changed
from tests.components.bluetooth import ( from tests.components.bluetooth import (
inject_bluetooth_service_info, inject_bluetooth_service_info,
patch_all_discovered_devices, patch_all_discovered_devices,
patch_bluetooth_time,
) )
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -1150,10 +1150,7 @@ async def test_unavailable(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()
@@ -1206,10 +1203,7 @@ async def test_sleepy_device(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()
@@ -1262,10 +1256,7 @@ async def test_sleepy_device_restore_state(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch_bluetooth_time(monotonic_now), patch_all_discovered_devices([]):
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME",
return_value=monotonic_now,
), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() dt_util.utcnow()

View File

@@ -113,7 +113,7 @@ async def test_gvh5178_multi_sensor(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -140,7 +140,7 @@ async def test_gvh5178_multi_sensor(hass: HomeAssistant) -> None:
# Fastforward time without BLE advertisements # Fastforward time without BLE advertisements
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -64,7 +64,7 @@ async def test_sensors(
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -115,7 +115,7 @@ async def test_sensors_io_series_4(
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -71,7 +71,7 @@ async def async_inject_broadcast(
async def async_move_time_forwards(hass: HomeAssistant, offset: float): async def async_move_time_forwards(hass: HomeAssistant, offset: float):
"""Mock time advancing from now to now+offset.""" """Mock time advancing from now to now+offset."""
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=time.monotonic() + offset, return_value=time.monotonic() + offset,
): ):
async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=offset)) async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=offset))

View File

@@ -73,7 +73,7 @@ async def test_binary_sensor_restore_state(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -83,7 +83,7 @@ async def test_binary_sensor_restore_state(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -56,7 +56,7 @@ async def test_sensors(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -295,7 +295,7 @@ async def test_unavailable(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -348,7 +348,7 @@ async def test_sleepy_device(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -401,7 +401,7 @@ async def test_sleepy_device_restore_state(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(

View File

@@ -693,7 +693,7 @@ async def test_unavailable(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -740,7 +740,7 @@ async def test_sleepy_device(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(
@@ -789,7 +789,7 @@ async def test_sleepy_device_restore_state(hass: HomeAssistant) -> None:
monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1
with patch( with patch(
"homeassistant.components.bluetooth.manager.MONOTONIC_TIME", "homeassistant.components.bluetooth.MONOTONIC_TIME",
return_value=monotonic_now, return_value=monotonic_now,
), patch_all_discovered_devices([]): ), patch_all_discovered_devices([]):
async_fire_time_changed( async_fire_time_changed(