mirror of
https://github.com/home-assistant/core.git
synced 2025-04-26 18:27:51 +00:00
Add slots to bluetooth manager (#95881)
This commit is contained in:
parent
d64ebbdc84
commit
bc28d7f33e
@ -18,6 +18,8 @@ TRACKER_BUFFERING_WOBBLE_SECONDS = 5
|
||||
class AdvertisementTracker:
|
||||
"""Tracker to determine the interval that a device is advertising."""
|
||||
|
||||
__slots__ = ("intervals", "sources", "_timings")
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the tracker."""
|
||||
self.intervals: dict[str, float] = {}
|
||||
|
@ -102,6 +102,28 @@ def _dispatch_bleak_callback(
|
||||
class BluetoothManager:
|
||||
"""Manage Bluetooth."""
|
||||
|
||||
__slots__ = (
|
||||
"hass",
|
||||
"_integration_matcher",
|
||||
"_cancel_unavailable_tracking",
|
||||
"_cancel_logging_listener",
|
||||
"_advertisement_tracker",
|
||||
"_unavailable_callbacks",
|
||||
"_connectable_unavailable_callbacks",
|
||||
"_callback_index",
|
||||
"_bleak_callbacks",
|
||||
"_all_history",
|
||||
"_connectable_history",
|
||||
"_non_connectable_scanners",
|
||||
"_connectable_scanners",
|
||||
"_adapters",
|
||||
"_sources",
|
||||
"_bluetooth_adapters",
|
||||
"storage",
|
||||
"slot_manager",
|
||||
"_debug",
|
||||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
|
@ -1,9 +1,11 @@
|
||||
"""Tests for the Bluetooth integration."""
|
||||
|
||||
|
||||
from contextlib import contextmanager
|
||||
import itertools
|
||||
import time
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from bleak import BleakClient
|
||||
from bleak.backends.scanner import AdvertisementData, BLEDevice
|
||||
@ -189,20 +191,46 @@ def inject_bluetooth_service_info(
|
||||
inject_advertisement(hass, device, advertisement_data)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def patch_all_discovered_devices(mock_discovered: list[BLEDevice]) -> None:
|
||||
"""Mock all the discovered devices from all the scanners."""
|
||||
return patch.object(
|
||||
_get_manager(),
|
||||
"_async_all_discovered_addresses",
|
||||
return_value={ble_device.address for ble_device in mock_discovered},
|
||||
manager = _get_manager()
|
||||
original_history = {}
|
||||
scanners = list(
|
||||
itertools.chain(
|
||||
manager._connectable_scanners, manager._non_connectable_scanners
|
||||
)
|
||||
)
|
||||
for scanner in scanners:
|
||||
data = scanner.discovered_devices_and_advertisement_data
|
||||
original_history[scanner] = data.copy()
|
||||
data.clear()
|
||||
if scanners:
|
||||
data = scanners[0].discovered_devices_and_advertisement_data
|
||||
data.clear()
|
||||
data.update(
|
||||
{device.address: (device, MagicMock()) for device in mock_discovered}
|
||||
)
|
||||
yield
|
||||
for scanner in scanners:
|
||||
data = scanner.discovered_devices_and_advertisement_data
|
||||
data.clear()
|
||||
data.update(original_history[scanner])
|
||||
|
||||
|
||||
@contextmanager
|
||||
def patch_discovered_devices(mock_discovered: list[BLEDevice]) -> None:
|
||||
"""Mock the combined best path to discovered devices from all the scanners."""
|
||||
return patch.object(
|
||||
_get_manager(), "async_discovered_devices", return_value=mock_discovered
|
||||
)
|
||||
manager = _get_manager()
|
||||
original_all_history = manager._all_history
|
||||
original_connectable_history = manager._connectable_history
|
||||
manager._connectable_history = {}
|
||||
manager._all_history = {
|
||||
device.address: MagicMock(device=device) for device in mock_discovered
|
||||
}
|
||||
yield
|
||||
manager._all_history = original_all_history
|
||||
manager._connectable_history = original_connectable_history
|
||||
|
||||
|
||||
async def async_setup_with_default_adapter(hass: HomeAssistant) -> MockConfigEntry:
|
||||
|
@ -15,7 +15,7 @@ from homeassistant.components.bluetooth.wrappers import (
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import _get_manager, generate_ble_device
|
||||
from . import generate_ble_device
|
||||
|
||||
MOCK_BLE_DEVICE = generate_ble_device(
|
||||
"00:00:00:00:00:00",
|
||||
@ -65,12 +65,7 @@ async def test_bleak_client_reports_with_address(
|
||||
"""Test we report when we pass an address to BleakClient."""
|
||||
install_multiple_bleak_catcher()
|
||||
|
||||
with patch.object(
|
||||
_get_manager(),
|
||||
"async_ble_device_from_address",
|
||||
return_value=MOCK_BLE_DEVICE,
|
||||
):
|
||||
instance = bleak.BleakClient("00:00:00:00:00:00")
|
||||
instance = bleak.BleakClient("00:00:00:00:00:00")
|
||||
|
||||
assert "BleakClient with an address instead of a BLEDevice" in caplog.text
|
||||
|
||||
@ -92,14 +87,7 @@ async def test_bleak_retry_connector_client_reports_with_address(
|
||||
"""Test we report when we pass an address to BleakClientWithServiceCache."""
|
||||
install_multiple_bleak_catcher()
|
||||
|
||||
with patch.object(
|
||||
_get_manager(),
|
||||
"async_ble_device_from_address",
|
||||
return_value=MOCK_BLE_DEVICE,
|
||||
):
|
||||
instance = bleak_retry_connector.BleakClientWithServiceCache(
|
||||
"00:00:00:00:00:00"
|
||||
)
|
||||
instance = bleak_retry_connector.BleakClientWithServiceCache("00:00:00:00:00:00")
|
||||
|
||||
assert "BleakClient with an address instead of a BLEDevice" in caplog.text
|
||||
|
||||
|
Loading…
x
Reference in New Issue
Block a user