mirror of
https://github.com/home-assistant/core.git
synced 2025-11-10 11:29:46 +00:00
567 lines
17 KiB
Python
567 lines
17 KiB
Python
"""The tests for the bluetooth WebSocket API."""
|
|
|
|
import asyncio
|
|
from datetime import timedelta
|
|
import time
|
|
from unittest.mock import ANY, patch
|
|
|
|
from bleak_retry_connector import Allocations
|
|
from freezegun import freeze_time
|
|
from habluetooth import BluetoothScanningMode
|
|
import pytest
|
|
|
|
from homeassistant.components.bluetooth import DOMAIN
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.util.dt import utcnow
|
|
|
|
from . import (
|
|
HCI0_SOURCE_ADDRESS,
|
|
HCI1_SOURCE_ADDRESS,
|
|
NON_CONNECTABLE_REMOTE_SOURCE_ADDRESS,
|
|
FakeScanner,
|
|
_get_manager,
|
|
generate_advertisement_data,
|
|
generate_ble_device,
|
|
inject_advertisement_with_source,
|
|
inject_advertisement_with_time_and_source_connectable,
|
|
)
|
|
|
|
from tests.common import MockConfigEntry, async_fire_time_changed
|
|
from tests.typing import WebSocketGenerator
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_advertisements(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_advertisements."""
|
|
address = "44:44:33:11:23:12"
|
|
|
|
switchbot_device_signal_100 = generate_ble_device(address, "wohand_signal_100")
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100", service_uuids=[], rssi=-100
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_100, switchbot_adv_signal_100, HCI0_SOURCE_ADDRESS
|
|
)
|
|
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_advertisements",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"add": [
|
|
{
|
|
"address": "44:44:33:11:23:12",
|
|
"connectable": True,
|
|
"manufacturer_data": {},
|
|
"name": "wohand_signal_100",
|
|
"rssi": -100,
|
|
"service_data": {},
|
|
"service_uuids": [],
|
|
"source": HCI0_SOURCE_ADDRESS,
|
|
"time": ANY,
|
|
"tx_power": -127,
|
|
"raw": None,
|
|
}
|
|
]
|
|
}
|
|
adv_time = response["event"]["add"][0]["time"]
|
|
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100",
|
|
manufacturer_data={123: b"abc"},
|
|
service_uuids=[],
|
|
rssi=-80,
|
|
)
|
|
# Inject with raw bytes data
|
|
inject_advertisement_with_time_and_source_connectable(
|
|
hass,
|
|
switchbot_device_signal_100,
|
|
switchbot_adv_signal_100,
|
|
time.monotonic(),
|
|
HCI1_SOURCE_ADDRESS,
|
|
True,
|
|
raw=b"\x02\x01\x06\x03\x03\x0f\x18",
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"add": [
|
|
{
|
|
"address": "44:44:33:11:23:12",
|
|
"connectable": True,
|
|
"manufacturer_data": {"123": "616263"},
|
|
"name": "wohand_signal_100",
|
|
"rssi": -80,
|
|
"service_data": {},
|
|
"service_uuids": [],
|
|
"source": HCI1_SOURCE_ADDRESS,
|
|
"time": ANY,
|
|
"tx_power": -127,
|
|
"raw": "02010603030f18",
|
|
}
|
|
]
|
|
}
|
|
new_time = response["event"]["add"][0]["time"]
|
|
assert new_time > adv_time
|
|
future_time = utcnow() + timedelta(seconds=3600)
|
|
future_monotonic_time = time.monotonic() + 3600
|
|
with (
|
|
freeze_time(future_time),
|
|
patch(
|
|
"habluetooth.manager.monotonic_time_coarse",
|
|
return_value=future_monotonic_time,
|
|
),
|
|
):
|
|
async_fire_time_changed(hass, future_time)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {"remove": [{"address": "44:44:33:11:23:12"}]}
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_connection_allocations(
|
|
hass: HomeAssistant,
|
|
register_hci0_scanner: None,
|
|
register_hci1_scanner: None,
|
|
register_non_connectable_scanner: None,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_connection_allocations."""
|
|
address = "44:44:33:11:23:12"
|
|
|
|
switchbot_device_signal_100 = generate_ble_device(address, "wohand_signal_100")
|
|
switchbot_adv_signal_100 = generate_advertisement_data(
|
|
local_name="wohand_signal_100", service_uuids=[], rssi=-100
|
|
)
|
|
inject_advertisement_with_source(
|
|
hass, switchbot_device_signal_100, switchbot_adv_signal_100, HCI0_SOURCE_ADDRESS
|
|
)
|
|
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_connection_allocations",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
|
|
assert response["event"] == [
|
|
{
|
|
"allocated": [],
|
|
"free": 5,
|
|
"slots": 5,
|
|
"source": "00:00:00:00:00:01",
|
|
},
|
|
{
|
|
"allocated": [],
|
|
"free": 5,
|
|
"slots": 5,
|
|
"source": HCI0_SOURCE_ADDRESS,
|
|
},
|
|
{
|
|
"allocated": [],
|
|
"free": 5,
|
|
"slots": 5,
|
|
"source": HCI1_SOURCE_ADDRESS,
|
|
},
|
|
{
|
|
"allocated": [],
|
|
"free": 0,
|
|
"slots": 0,
|
|
"source": NON_CONNECTABLE_REMOTE_SOURCE_ADDRESS,
|
|
},
|
|
]
|
|
|
|
manager = _get_manager()
|
|
manager.async_on_allocation_changed(
|
|
Allocations(
|
|
adapter="hci1", # Will be translated to source
|
|
slots=5,
|
|
free=4,
|
|
allocated=["AA:BB:CC:DD:EE:EE"],
|
|
)
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == [
|
|
{
|
|
"allocated": ["AA:BB:CC:DD:EE:EE"],
|
|
"free": 4,
|
|
"slots": 5,
|
|
"source": "AA:BB:CC:DD:EE:11",
|
|
},
|
|
]
|
|
manager.async_on_allocation_changed(
|
|
Allocations(
|
|
adapter="hci1", # Will be translated to source
|
|
slots=5,
|
|
free=5,
|
|
allocated=[],
|
|
)
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == [
|
|
{"allocated": [], "free": 5, "slots": 5, "source": HCI1_SOURCE_ADDRESS}
|
|
]
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_connection_allocations_specific_scanner(
|
|
hass: HomeAssistant,
|
|
register_non_connectable_scanner: None,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_connection_allocations for a specific source address."""
|
|
entry = MockConfigEntry(
|
|
domain=DOMAIN, unique_id=NON_CONNECTABLE_REMOTE_SOURCE_ADDRESS
|
|
)
|
|
entry.add_to_hass(hass)
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_connection_allocations",
|
|
"config_entry_id": entry.entry_id,
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
|
|
assert response["event"] == [
|
|
{
|
|
"allocated": [],
|
|
"free": 0,
|
|
"slots": 0,
|
|
"source": NON_CONNECTABLE_REMOTE_SOURCE_ADDRESS,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_connection_allocations_invalid_config_entry_id(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_connection_allocations for an invalid config entry id."""
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_connection_allocations",
|
|
"config_entry_id": "non_existent",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert not response["success"]
|
|
assert response["error"]["code"] == "invalid_config_entry_id"
|
|
assert response["error"]["message"] == "Config entry non_existent not found"
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_connection_allocations_invalid_scanner(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_connection_allocations for an invalid source address."""
|
|
entry = MockConfigEntry(domain=DOMAIN, unique_id="invalid")
|
|
entry.add_to_hass(hass)
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_connection_allocations",
|
|
"config_entry_id": entry.entry_id,
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert not response["success"]
|
|
assert response["error"]["code"] == "invalid_source"
|
|
assert response["error"]["message"] == "Source invalid not found"
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_details(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_connection_allocations."""
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_details",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
|
|
assert response["event"] == {
|
|
"add": [
|
|
{
|
|
"adapter": "hci0",
|
|
"connectable": False,
|
|
"name": "hci0 (00:00:00:00:00:01)",
|
|
"source": "00:00:00:00:00:01",
|
|
"scanner_type": "unknown",
|
|
}
|
|
]
|
|
}
|
|
|
|
manager = _get_manager()
|
|
hci3_scanner = FakeScanner("AA:BB:CC:DD:EE:33", "hci3")
|
|
cancel_hci3 = manager.async_register_hass_scanner(hci3_scanner)
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"add": [
|
|
{
|
|
"adapter": "hci3",
|
|
"connectable": False,
|
|
"name": "hci3 (AA:BB:CC:DD:EE:33)",
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"scanner_type": "unknown",
|
|
}
|
|
]
|
|
}
|
|
cancel_hci3()
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"remove": [
|
|
{
|
|
"adapter": "hci3",
|
|
"connectable": False,
|
|
"name": "hci3 (AA:BB:CC:DD:EE:33)",
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"scanner_type": "unknown",
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_details_specific_scanner(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_scanner_details for a specific source address."""
|
|
entry = MockConfigEntry(domain=DOMAIN, unique_id="AA:BB:CC:DD:EE:33")
|
|
entry.add_to_hass(hass)
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_details",
|
|
"config_entry_id": entry.entry_id,
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
manager = _get_manager()
|
|
hci3_scanner = FakeScanner("AA:BB:CC:DD:EE:33", "hci3")
|
|
cancel_hci3 = manager.async_register_hass_scanner(hci3_scanner)
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"add": [
|
|
{
|
|
"adapter": "hci3",
|
|
"connectable": False,
|
|
"name": "hci3 (AA:BB:CC:DD:EE:33)",
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"scanner_type": "unknown",
|
|
}
|
|
]
|
|
}
|
|
cancel_hci3()
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"remove": [
|
|
{
|
|
"adapter": "hci3",
|
|
"connectable": False,
|
|
"name": "hci3 (AA:BB:CC:DD:EE:33)",
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"scanner_type": "unknown",
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_details_invalid_config_entry_id(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_scanner_details for an invalid config entry id."""
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_details",
|
|
"config_entry_id": "non_existent",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert not response["success"]
|
|
assert response["error"]["code"] == "invalid_config_entry_id"
|
|
assert response["error"]["message"] == "Config entry non_existent not found"
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_state(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_scanner_state."""
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_state",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
# Should receive initial state for existing scanner
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"source": "00:00:00:00:00:01",
|
|
"adapter": "hci0",
|
|
"current_mode": "active",
|
|
"requested_mode": "active",
|
|
}
|
|
|
|
# Register a new scanner
|
|
manager = _get_manager()
|
|
hci3_scanner = FakeScanner("AA:BB:CC:DD:EE:33", "hci3")
|
|
cancel_hci3 = manager.async_register_hass_scanner(hci3_scanner)
|
|
|
|
# Simulate a mode change
|
|
hci3_scanner.current_mode = BluetoothScanningMode.ACTIVE
|
|
hci3_scanner.requested_mode = BluetoothScanningMode.ACTIVE
|
|
manager.scanner_mode_changed(hci3_scanner)
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"adapter": "hci3",
|
|
"current_mode": "active",
|
|
"requested_mode": "active",
|
|
}
|
|
|
|
cancel_hci3()
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_state_specific_scanner(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_scanner_state for a specific source address."""
|
|
# Register the scanner first
|
|
manager = _get_manager()
|
|
hci3_scanner = FakeScanner("AA:BB:CC:DD:EE:33", "hci3")
|
|
cancel_hci3 = manager.async_register_hass_scanner(hci3_scanner)
|
|
|
|
entry = MockConfigEntry(domain=DOMAIN, unique_id="AA:BB:CC:DD:EE:33")
|
|
entry.add_to_hass(hass)
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_state",
|
|
"config_entry_id": entry.entry_id,
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["success"]
|
|
|
|
# Should receive initial state
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"adapter": "hci3",
|
|
"current_mode": None,
|
|
"requested_mode": None,
|
|
}
|
|
|
|
# Simulate a mode change
|
|
hci3_scanner.current_mode = BluetoothScanningMode.PASSIVE
|
|
hci3_scanner.requested_mode = BluetoothScanningMode.ACTIVE
|
|
manager.scanner_mode_changed(hci3_scanner)
|
|
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert response["event"] == {
|
|
"source": "AA:BB:CC:DD:EE:33",
|
|
"adapter": "hci3",
|
|
"current_mode": "passive",
|
|
"requested_mode": "active",
|
|
}
|
|
|
|
cancel_hci3()
|
|
|
|
|
|
@pytest.mark.usefixtures("enable_bluetooth")
|
|
async def test_subscribe_scanner_state_invalid_config_entry_id(
|
|
hass: HomeAssistant,
|
|
hass_ws_client: WebSocketGenerator,
|
|
) -> None:
|
|
"""Test bluetooth subscribe_scanner_state for an invalid config entry id."""
|
|
client = await hass_ws_client()
|
|
await client.send_json(
|
|
{
|
|
"id": 1,
|
|
"type": "bluetooth/subscribe_scanner_state",
|
|
"config_entry_id": "non_existent",
|
|
}
|
|
)
|
|
async with asyncio.timeout(1):
|
|
response = await client.receive_json()
|
|
assert not response["success"]
|
|
assert response["error"]["code"] == "invalid_config_entry_id"
|
|
assert response["error"]["message"] == "Config entry non_existent not found"
|