core/tests/components/bluetooth/test_init.py

"""Tests for the Bluetooth integration."""
from unittest.mock import MagicMock, patch

from bleak import BleakError
from bleak.backends.scanner import AdvertisementData, BLEDevice

from homeassistant.components import bluetooth
from homeassistant.components.bluetooth import (
    SOURCE_LOCAL,
    BluetoothChange,
    BluetoothServiceInfo,
    models,
)
from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_HOMEASSISTANT_STOP
from homeassistant.setup import async_setup_component


async def test_setup_and_stop(hass, mock_bleak_scanner_start):
    """Test we and setup and stop the scanner."""
    mock_bt = [
        {"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
    ]
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
    await hass.async_block_till_done()
    assert len(mock_bleak_scanner_start.mock_calls) == 1


async def test_setup_and_stop_no_bluetooth(hass, caplog):
    """Test we fail gracefully when bluetooth is not available."""
    mock_bt = [
        {"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
    ]
    with patch(
        "homeassistant.components.bluetooth.HaBleakScanner", side_effect=BleakError
    ) as mock_ha_bleak_scanner, patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(
        hass.config_entries.flow, "async_init"
    ):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
    await hass.async_block_till_done()
    assert len(mock_ha_bleak_scanner.mock_calls) == 1
    assert "Could not create bluetooth scanner" in caplog.text


async def test_calling_async_discovered_devices_no_bluetooth(hass, caplog):
    """Test we fail gracefully when asking for discovered devices and there is no blueooth."""
    mock_bt = []
    with patch(
        "homeassistant.components.bluetooth.HaBleakScanner", side_effect=BleakError
    ) as mock_ha_bleak_scanner, patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(
        hass.config_entries.flow, "async_init"
    ):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
    await hass.async_block_till_done()
    assert len(mock_ha_bleak_scanner.mock_calls) == 1
    assert "Could not create bluetooth scanner" in caplog.text
    assert not bluetooth.async_discovered_service_info(hass)
    assert not bluetooth.async_address_present(hass, "aa:bb:bb:dd:ee:ff")


async def test_discovery_match_by_service_uuid(hass, mock_bleak_scanner_start):
    """Test bluetooth discovery match by service_uuid."""
    mock_bt = [
        {"domain": "switchbot", "service_uuid": "cba20d00-224d-11e6-9fb8-0002a5d5c51b"}
    ]
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
        wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])

        models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 0

        switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
        switchbot_adv = AdvertisementData(
            local_name="wohand", service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
        )

        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 1
        assert mock_config_flow.mock_calls[0][1][0] == "switchbot"


async def test_discovery_match_by_local_name(hass, mock_bleak_scanner_start):
    """Test bluetooth discovery match by local_name."""
    mock_bt = [{"domain": "switchbot", "local_name": "wohand"}]
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        wrong_device = BLEDevice("44:44:33:11:23:45", "wrong_name")
        wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])

        models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 0

        switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
        switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])

        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 1
        assert mock_config_flow.mock_calls[0][1][0] == "switchbot"


async def test_discovery_match_by_manufacturer_id_and_first_byte(
    hass, mock_bleak_scanner_start
):
    """Test bluetooth discovery match by manufacturer_id and manufacturer_data_first_byte."""
    mock_bt = [
        {
            "domain": "homekit_controller",
            "manufacturer_id": 76,
            "manufacturer_data_first_byte": 0x06,
        }
    ]
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init") as mock_config_flow:
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        hkc_device = BLEDevice("44:44:33:11:23:45", "lock")
        hkc_adv = AdvertisementData(
            local_name="lock", service_uuids=[], manufacturer_data={76: b"\x06"}
        )

        models.HA_BLEAK_SCANNER._callback(hkc_device, hkc_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 1
        assert mock_config_flow.mock_calls[0][1][0] == "homekit_controller"
        mock_config_flow.reset_mock()

        # 2nd discovery should not generate another flow
        models.HA_BLEAK_SCANNER._callback(hkc_device, hkc_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 0

        mock_config_flow.reset_mock()
        not_hkc_device = BLEDevice("44:44:33:11:23:21", "lock")
        not_hkc_adv = AdvertisementData(
            local_name="lock", service_uuids=[], manufacturer_data={76: b"\x02"}
        )

        models.HA_BLEAK_SCANNER._callback(not_hkc_device, not_hkc_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 0
        not_apple_device = BLEDevice("44:44:33:11:23:23", "lock")
        not_apple_adv = AdvertisementData(
            local_name="lock", service_uuids=[], manufacturer_data={21: b"\x02"}
        )

        models.HA_BLEAK_SCANNER._callback(not_apple_device, not_apple_adv)
        await hass.async_block_till_done()

        assert len(mock_config_flow.mock_calls) == 0


async def test_async_discovered_device_api(hass, mock_bleak_scanner_start):
    """Test the async_discovered_device_api."""
    mock_bt = []
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch(
        "bleak.BleakScanner.discovered_devices",  # Must patch before we setup
        [MagicMock(address="44:44:33:11:23:45")],
    ):
        assert not bluetooth.async_discovered_service_info(hass)
        assert not bluetooth.async_address_present(hass, "44:44:22:22:11:22")

        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        assert not bluetooth.async_discovered_service_info(hass)

        wrong_device = BLEDevice("44:44:33:11:23:42", "wrong_name")
        wrong_adv = AdvertisementData(local_name="wrong_name", service_uuids=[])
        models.HA_BLEAK_SCANNER._callback(wrong_device, wrong_adv)
        switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
        switchbot_adv = AdvertisementData(local_name="wohand", service_uuids=[])
        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

        service_infos = bluetooth.async_discovered_service_info(hass)
        assert len(service_infos) == 1
        # wrong_name should not appear because bleak no longer sees it
        assert service_infos[0].name == "wohand"
        assert service_infos[0].source == SOURCE_LOCAL

        assert bluetooth.async_address_present(hass, "44:44:33:11:23:42") is False
        assert bluetooth.async_address_present(hass, "44:44:33:11:23:45") is True


async def test_register_callbacks(hass, mock_bleak_scanner_start):
    """Test registering a callback."""
    mock_bt = []
    callbacks = []

    def _fake_subscriber(
        service_info: BluetoothServiceInfo,
        change: BluetoothChange,
    ) -> None:
        """Fake subscriber for the BleakScanner."""
        callbacks.append((service_info, change))
        if len(callbacks) >= 3:
            raise ValueError

    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        cancel = bluetooth.async_register_callback(
            hass,
            _fake_subscriber,
            {"service_uuids": {"cba20d00-224d-11e6-9fb8-0002a5d5c51b"}},
        )

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
        switchbot_adv = AdvertisementData(
            local_name="wohand",
            service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
            manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
            service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
        )

        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)

        empty_device = BLEDevice("11:22:33:44:55:66", "empty")
        empty_adv = AdvertisementData(local_name="empty")

        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

        empty_device = BLEDevice("11:22:33:44:55:66", "empty")
        empty_adv = AdvertisementData(local_name="empty")

        # 3rd callback raises ValueError but is still tracked
        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

        cancel()

        # 4th callback should not be tracked since we canceled
        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

    assert len(callbacks) == 3

    service_info: BluetoothServiceInfo = callbacks[0][0]
    assert service_info.name == "wohand"
    assert service_info.source == SOURCE_LOCAL
    assert service_info.manufacturer == "Nordic Semiconductor ASA"
    assert service_info.manufacturer_id == 89

    service_info: BluetoothServiceInfo = callbacks[1][0]
    assert service_info.name == "empty"
    assert service_info.source == SOURCE_LOCAL
    assert service_info.manufacturer is None
    assert service_info.manufacturer_id is None

    service_info: BluetoothServiceInfo = callbacks[2][0]
    assert service_info.name == "empty"
    assert service_info.source == SOURCE_LOCAL
    assert service_info.manufacturer is None
    assert service_info.manufacturer_id is None


async def test_register_callback_by_address(hass, mock_bleak_scanner_start):
    """Test registering a callback by address."""
    mock_bt = []
    callbacks = []

    def _fake_subscriber(
        service_info: BluetoothServiceInfo, change: BluetoothChange
    ) -> None:
        """Fake subscriber for the BleakScanner."""
        callbacks.append((service_info, change))
        if len(callbacks) >= 3:
            raise ValueError

    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=mock_bt
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

        cancel = bluetooth.async_register_callback(
            hass,
            _fake_subscriber,
            {"address": "44:44:33:11:23:45"},
        )

        assert len(mock_bleak_scanner_start.mock_calls) == 1

        switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
        switchbot_adv = AdvertisementData(
            local_name="wohand",
            service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
            manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
            service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
        )

        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)

        empty_device = BLEDevice("11:22:33:44:55:66", "empty")
        empty_adv = AdvertisementData(local_name="empty")

        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

        empty_device = BLEDevice("11:22:33:44:55:66", "empty")
        empty_adv = AdvertisementData(local_name="empty")

        # 3rd callback raises ValueError but is still tracked
        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

        cancel()

        # 4th callback should not be tracked since we canceled
        models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
        await hass.async_block_till_done()

        # Now register again with a callback that fails to
        # make sure we do not perm fail
        cancel = bluetooth.async_register_callback(
            hass,
            _fake_subscriber,
            {"address": "44:44:33:11:23:45"},
        )
        cancel()

        # Now register again, since the 3rd callback
        # should fail but we should still record it
        cancel = bluetooth.async_register_callback(
            hass,
            _fake_subscriber,
            {"address": "44:44:33:11:23:45"},
        )
        cancel()

    assert len(callbacks) == 3

    for idx in range(3):
        service_info: BluetoothServiceInfo = callbacks[idx][0]
        assert service_info.name == "wohand"
        assert service_info.manufacturer == "Nordic Semiconductor ASA"
        assert service_info.manufacturer_id == 89


async def test_wrapped_instance_with_filter(hass, mock_bleak_scanner_start):
    """Test consumers can use the wrapped instance with a filter as if it was normal BleakScanner."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    detected = []

    def _device_detected(
        device: BLEDevice, advertisement_data: AdvertisementData
    ) -> None:
        """Handle a detected device."""
        detected.append((device, advertisement_data))

    switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
    switchbot_adv = AdvertisementData(
        local_name="wohand",
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
        manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
    )
    empty_device = BLEDevice("11:22:33:44:55:66", "empty")
    empty_adv = AdvertisementData(local_name="empty")

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper(
        filters={"UUIDs": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]}
    )
    scanner.register_detection_callback(_device_detected)

    mock_discovered = [MagicMock()]
    type(models.HA_BLEAK_SCANNER).discovered_devices = mock_discovered
    models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
    await hass.async_block_till_done()

    discovered = await scanner.discover(timeout=0)
    assert len(discovered) == 1
    assert discovered == mock_discovered
    assert len(detected) == 1

    scanner.register_detection_callback(_device_detected)
    # We should get a reply from the history when we register again
    assert len(detected) == 2
    scanner.register_detection_callback(_device_detected)
    # We should get a reply from the history when we register again
    assert len(detected) == 3

    type(models.HA_BLEAK_SCANNER).discovered_devices = []
    discovered = await scanner.discover(timeout=0)
    assert len(discovered) == 0
    assert discovered == []

    models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
    assert len(detected) == 4

    # The filter we created in the wrapped scanner with should be respected
    # and we should not get another callback
    models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
    assert len(detected) == 4


async def test_wrapped_instance_with_service_uuids(hass, mock_bleak_scanner_start):
    """Test consumers can use the wrapped instance with a service_uuids list as if it was normal BleakScanner."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    detected = []

    def _device_detected(
        device: BLEDevice, advertisement_data: AdvertisementData
    ) -> None:
        """Handle a detected device."""
        detected.append((device, advertisement_data))

    switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
    switchbot_adv = AdvertisementData(
        local_name="wohand",
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
        manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
    )
    empty_device = BLEDevice("11:22:33:44:55:66", "empty")
    empty_adv = AdvertisementData(local_name="empty")

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper(
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
    )
    scanner.register_detection_callback(_device_detected)

    type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
    for _ in range(2):
        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

    assert len(detected) == 2

    # The UUIDs list we created in the wrapped scanner with should be respected
    # and we should not get another callback
    models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
    assert len(detected) == 2


async def test_wrapped_instance_with_broken_callbacks(hass, mock_bleak_scanner_start):
    """Test broken callbacks do not cause the scanner to fail."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    detected = []

    def _device_detected(
        device: BLEDevice, advertisement_data: AdvertisementData
    ) -> None:
        """Handle a detected device."""
        if detected:
            raise ValueError
        detected.append((device, advertisement_data))

    switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
    switchbot_adv = AdvertisementData(
        local_name="wohand",
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
        manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
    )

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper(
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]
    )
    scanner.register_detection_callback(_device_detected)

    models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
    await hass.async_block_till_done()
    models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
    await hass.async_block_till_done()
    assert len(detected) == 1


async def test_wrapped_instance_changes_uuids(hass, mock_bleak_scanner_start):
    """Test consumers can use the wrapped instance can change the uuids later."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    detected = []

    def _device_detected(
        device: BLEDevice, advertisement_data: AdvertisementData
    ) -> None:
        """Handle a detected device."""
        detected.append((device, advertisement_data))

    switchbot_device = BLEDevice("44:44:33:11:23:45", "wohand")
    switchbot_adv = AdvertisementData(
        local_name="wohand",
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
        manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
    )
    empty_device = BLEDevice("11:22:33:44:55:66", "empty")
    empty_adv = AdvertisementData(local_name="empty")

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper()
    scanner.set_scanning_filter(service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"])
    scanner.register_detection_callback(_device_detected)

    type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
    for _ in range(2):
        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

    assert len(detected) == 2

    # The UUIDs list we created in the wrapped scanner with should be respected
    # and we should not get another callback
    models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
    assert len(detected) == 2


async def test_wrapped_instance_changes_filters(hass, mock_bleak_scanner_start):
    """Test consumers can use the wrapped instance can change the filter later."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    detected = []

    def _device_detected(
        device: BLEDevice, advertisement_data: AdvertisementData
    ) -> None:
        """Handle a detected device."""
        detected.append((device, advertisement_data))

    switchbot_device = BLEDevice("44:44:33:11:23:42", "wohand")
    switchbot_adv = AdvertisementData(
        local_name="wohand",
        service_uuids=["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
        manufacturer_data={89: b"\xd8.\xad\xcd\r\x85"},
        service_data={"00000d00-0000-1000-8000-00805f9b34fb": b"H\x10c"},
    )
    empty_device = BLEDevice("11:22:33:44:55:62", "empty")
    empty_adv = AdvertisementData(local_name="empty")

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper()
    scanner.set_scanning_filter(
        filters={"UUIDs": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"]}
    )
    scanner.register_detection_callback(_device_detected)

    type(models.HA_BLEAK_SCANNER).discovered_devices = [MagicMock()]
    for _ in range(2):
        models.HA_BLEAK_SCANNER._callback(switchbot_device, switchbot_adv)
        await hass.async_block_till_done()

    assert len(detected) == 2

    # The UUIDs list we created in the wrapped scanner with should be respected
    # and we should not get another callback
    models.HA_BLEAK_SCANNER._callback(empty_device, empty_adv)
    assert len(detected) == 2


async def test_wrapped_instance_unsupported_filter(
    hass, mock_bleak_scanner_start, caplog
):
    """Test we want when their filter is ineffective."""
    with patch(
        "homeassistant.components.bluetooth.async_get_bluetooth", return_value=[]
    ), patch.object(hass.config_entries.flow, "async_init"):
        assert await async_setup_component(
            hass, bluetooth.DOMAIN, {bluetooth.DOMAIN: {}}
        )
        hass.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
        await hass.async_block_till_done()

    assert models.HA_BLEAK_SCANNER is not None
    scanner = models.HaBleakScannerWrapper()
    scanner.set_scanning_filter(
        filters={
            "unsupported": ["cba20d00-224d-11e6-9fb8-0002a5d5c51b"],
            "DuplicateData": True,
        }
    )
    assert "Only UUIDs filters are supported" in caplog.text