"""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_start.""" mock_bt = [ { "domain": "homekit_controller", "manufacturer_id": 76, "manufacturer_data_start": [0x06, 0x02, 0x03], } ] 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\x02\x03\x99"}, ) 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 isinstance(service_infos[0].device, BLEDevice) assert isinstance(service_infos[0].advertisement, AdvertisementData) 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 async def test_async_ble_device_from_address(hass, mock_bleak_scanner_start): """Test the async_ble_device_from_address 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 ( bluetooth.async_ble_device_from_address(hass, "44:44:33:11:23:45") is None ) 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) 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 ( bluetooth.async_ble_device_from_address(hass, "44:44:33:11:23:45") is switchbot_device ) assert ( bluetooth.async_ble_device_from_address(hass, "00:66:33:22:11:22") is None )