diff --git a/poetry.lock b/poetry.lock index 79210f0..2e4abc3 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.8.5 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand. [[package]] name = "aioesphomeapi" @@ -1260,6 +1260,24 @@ pluggy = ">=1.5,<2" [package.extras] dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"] +[[package]] +name = "pytest-asyncio" +version = "0.25.0" +description = "Pytest support for asyncio" +optional = false +python-versions = ">=3.9" +files = [ + {file = "pytest_asyncio-0.25.0-py3-none-any.whl", hash = "sha256:db5432d18eac6b7e28b46dcd9b69921b55c3b1086e85febfe04e70b18d9e81b3"}, + {file = "pytest_asyncio-0.25.0.tar.gz", hash = "sha256:8c0610303c9e0442a5db8604505fc0f545456ba1528824842b37b4a626cbf609"}, +] + +[package.dependencies] +pytest = ">=8.2,<9" + +[package.extras] +docs = ["sphinx (>=5.3)", "sphinx-rtd-theme (>=1)"] +testing = ["coverage (>=6.2)", "hypothesis (>=5.7.1)"] + [[package]] name = "pytest-codspeed" version = "3.1.0" @@ -2177,4 +2195,4 @@ ifaddr = ">=0.1.7" [metadata] lock-version = "2.0" python-versions = ">=3.11,<3.14" -content-hash = "209da2bd3753a6511f9ef29528f43880f0d36b8457f7eee88babcef6549eec00" +content-hash = "f30ec8057cccd6d61b19eaeb3a48ffd6618aad064e5ac76e1827a965cc217388" diff --git a/pyproject.toml b/pyproject.toml index 9ff22da..fa169e4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -34,6 +34,8 @@ lru-dict = ">=1.2.0" pytest = ">=7,<9" pytest-cov = ">=3,<7" pytest-codspeed = "^3.1.0" +zeroconf = "^0.136.2" +pytest-asyncio = "^0.25.0" [tool.poetry.group.docs] optional = true diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..38d4eda 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1,31 @@ +from typing import Any + +from bleak.backends.scanner import BLEDevice + +BLE_DEVICE_DEFAULTS = { + "name": None, + "rssi": -127, + "details": None, +} + + +def generate_ble_device( + address: str | None = None, + name: str | None = None, + details: Any | None = None, + rssi: int | None = None, + **kwargs: Any, +) -> BLEDevice: + """Generate a BLEDevice with defaults.""" + new = kwargs.copy() + if address is not None: + new["address"] = address + if name is not None: + new["name"] = name + if details is not None: + new["details"] = details + if rssi is not None: + new["rssi"] = rssi + for key, value in BLE_DEVICE_DEFAULTS.items(): + new.setdefault(key, value) + return BLEDevice(**new) diff --git a/tests/backend/test_client.py b/tests/backend/test_client.py new file mode 100644 index 0000000..092bb75 --- /dev/null +++ b/tests/backend/test_client.py @@ -0,0 +1,299 @@ +from unittest.mock import patch +from uuid import UUID + +import pytest +from aioesphomeapi import ( + APIClient, + APIVersion, + BluetoothGATTCharacteristic, + BluetoothGATTDescriptor, + BluetoothGATTService, + BluetoothProxyFeature, + DeviceInfo, + ESPHomeBluetoothGATTServices, +) +from bleak.exc import BleakError +from habluetooth import BaseHaRemoteScanner, HaBluetoothConnector +from pytest_asyncio import fixture as aio_fixture + +from bleak_esphome.backend.cache import ESPHomeBluetoothCache +from bleak_esphome.backend.client import ESPHomeClient, ESPHomeClientData +from bleak_esphome.backend.device import ESPHomeBluetoothDevice +from bleak_esphome.backend.scanner import ESPHomeScanner + +from .. import generate_ble_device + +ESP_MAC_ADDRESS = "AA:BB:CC:DD:EE:FF" +ESP_NAME = "proxy" + + +def split_uuid(uuid: str) -> list[int]: + int_128 = UUID(uuid).int + return [int_128 >> 64, int_128 & 0xFFFFFFFFFFFFFFFF] + + +@pytest.fixture +def esphome_bluetooth_gatt_services() -> ESPHomeBluetoothGATTServices: + + service1 = BluetoothGATTService( + uuid=split_uuid("00001800-0000-1000-8000-00805f9b34fb"), + handle=1, + characteristics=[], + ) + object.__setattr__( + service1, + "characteristics", + [ + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a00-0000-1000-8000-00805f9b34fb"), + handle=3, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a01-0000-1000-8000-00805f9b34fb"), + handle=5, + properties=2, + descriptors=[], + ), + ], + ) + service2 = BluetoothGATTService( + uuid=split_uuid("00001801-0000-1000-8000-00805f9b34fb"), + handle=6, + characteristics=[], + ) + service2_chars = [ + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a05-0000-1000-8000-00805f9b34fb"), + handle=8, + properties=32, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002b3a-0000-1000-8000-00805f9b34fb"), + handle=11, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002b29-0000-1000-8000-00805f9b34fb"), + handle=13, + properties=10, + descriptors=[], + ), + ] + object.__setattr__( + service2_chars[0], + "descriptors", + [ + BluetoothGATTDescriptor( + uuid=split_uuid("00002902-0000-1000-8000-00805f9b34fb"), + handle=9, + ) + ], + ) + object.__setattr__(service2, "characteristics", service2_chars) + + service3 = BluetoothGATTService( + uuid=split_uuid("d30a7847-e12b-09a8-b04b-8e0922a9abab"), + handle=14, + characteristics=[], + ) + service3_chars = [ + BluetoothGATTCharacteristic( + uuid=split_uuid("030b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=16, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("040b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=18, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("090b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=20, + properties=10, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("050b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=22, + properties=10, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("060b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=24, + properties=8, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("070b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=26, + properties=8, + descriptors=[], + ), + ] + object.__setattr__(service3, "characteristics", service3_chars) + service4 = BluetoothGATTService( + uuid=split_uuid("0000180a-0000-1000-8000-00805f9b34fb"), + handle=27, + characteristics=[], + ) + service4_chars = [ + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a29-0000-1000-8000-00805f9b34fb"), + handle=29, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a24-0000-1000-8000-00805f9b34fb"), + handle=31, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a25-0000-1000-8000-00805f9b34fb"), + handle=33, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a26-0000-1000-8000-00805f9b34fb"), + handle=35, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a27-0000-1000-8000-00805f9b34fb"), + handle=37, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("00002a28-0000-1000-8000-00805f9b34fb"), + handle=39, + properties=2, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("0a0b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=41, + properties=10, + descriptors=[], + ), + BluetoothGATTCharacteristic( + uuid=split_uuid("0b0b7847-e12b-09a8-b04b-8e0922a9abab"), + handle=43, + properties=10, + descriptors=[], + ), + ] + object.__setattr__(service4, "characteristics", service4_chars) + return ESPHomeBluetoothGATTServices( + address=57911560448430, + services=[service1, service2, service3, service4], + ) + + +def test_get_services() -> None: + connector = HaBluetoothConnector(ESPHomeClientData, ESP_MAC_ADDRESS, lambda: True) + scanner = ESPHomeScanner(ESP_MAC_ADDRESS, ESP_NAME, connector, True) + assert isinstance(scanner, BaseHaRemoteScanner) + + +@aio_fixture(name="client_data") +async def client_data_fixture(mock_client: APIClient) -> ESPHomeClientData: + """Return a client data fixture.""" + connector = HaBluetoothConnector(ESPHomeClientData, ESP_MAC_ADDRESS, lambda: True) + return ESPHomeClientData( + bluetooth_device=ESPHomeBluetoothDevice(ESP_NAME, ESP_MAC_ADDRESS), + cache=ESPHomeBluetoothCache(), + client=mock_client, + device_info=DeviceInfo( + mac_address=ESP_MAC_ADDRESS, + name=ESP_NAME, + bluetooth_proxy_feature_flags=BluetoothProxyFeature.PASSIVE_SCAN + | BluetoothProxyFeature.ACTIVE_CONNECTIONS + | BluetoothProxyFeature.REMOTE_CACHING + | BluetoothProxyFeature.PAIRING + | BluetoothProxyFeature.CACHE_CLEARING + | BluetoothProxyFeature.RAW_ADVERTISEMENTS, + ), + api_version=APIVersion(1, 9), + title=ESP_NAME, + scanner=ESPHomeScanner(ESP_MAC_ADDRESS, ESP_NAME, connector, True), + ) + + +@pytest.mark.asyncio +async def test_client_usage_while_not_connected(client_data: ESPHomeClientData) -> None: + """Test client usage while not connected.""" + ble_device = generate_ble_device( + "CC:BB:AA:DD:EE:FF", details={"source": ESP_MAC_ADDRESS, "address_type": 1} + ) + + client = ESPHomeClient(ble_device, client_data=client_data) + with pytest.raises( + BleakError, match=f"{ESP_NAME}.*{ESP_MAC_ADDRESS}.*not connected" + ): + await client.write_gatt_char("test", b"test") + + +@pytest.mark.asyncio +async def test_client_get_services_and_write( + client_data: ESPHomeClientData, + esphome_bluetooth_gatt_services: ESPHomeBluetoothGATTServices, +) -> None: + """Test getting client services and writing a GATT char.""" + ble_device = generate_ble_device( + "CC:BB:AA:DD:EE:FF", details={"source": ESP_MAC_ADDRESS, "address_type": 1} + ) + + client = ESPHomeClient(ble_device, client_data=client_data) + client._is_connected = True + with patch.object( + client._client, + "bluetooth_gatt_get_services", + return_value=esphome_bluetooth_gatt_services, + ): + services = await client.get_services() + + assert services is not None + + char = client._resolve_characteristic( + char_specifier="090b7847-e12b-09a8-b04b-8e0922a9abab", + ) + assert char is not None + assert char.uuid == "090b7847-e12b-09a8-b04b-8e0922a9abab" + assert char.properties == ["read", "write"] + assert char.handle == 20 + + char2 = services.get_characteristic("090b7847-e12b-09a8-b04b-8e0922a9abab") + assert char2 is not None + assert char2.uuid == "090b7847-e12b-09a8-b04b-8e0922a9abab" + assert char2.properties == ["read", "write"] + assert char2.handle == 20 + + char3 = services.get_characteristic(UUID("090b7847-e12b-09a8-b04b-8e0922a9abab")) + assert char3 is not None + assert char3.uuid == "090b7847-e12b-09a8-b04b-8e0922a9abab" + assert char3.properties == ["read", "write"] + assert char3.handle == 20 + + with patch.object( + client._client, + "bluetooth_gatt_write", + ) as mock_write: + await client.write_gatt_char( + "090b7847-e12b-09a8-b04b-8e0922a9abab", + b"test", + True, + ) + + mock_write.assert_called_once_with(225106397622015, 20, b"test", True) diff --git a/tests/conftest.py b/tests/conftest.py index fd5ca8c..b3a7aba 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,10 +1,19 @@ +from unittest.mock import AsyncMock, Mock, patch + import pytest +from aioesphomeapi import ( + APIClient, + APIVersion, + DeviceInfo, + ReconnectLogic, +) from bleak_retry_connector import BleakSlotManager from bluetooth_adapters import BluetoothAdapters from habluetooth import ( BluetoothManager, set_manager, ) +from zeroconf import Zeroconf @pytest.fixture(scope="session", autouse=True) @@ -12,3 +21,73 @@ def manager(): slot_manager = BleakSlotManager() bluetooth_adapters = BluetoothAdapters() set_manager(BluetoothManager(bluetooth_adapters, slot_manager)) + + +@pytest.fixture +def mock_device_info() -> DeviceInfo: + """Return the default mocked device info.""" + return DeviceInfo( + uses_password=False, + name="test", + legacy_bluetooth_proxy_version=0, + # ESPHome mac addresses are UPPER case + mac_address="11:22:33:44:55:AA", + esphome_version="1.0.0", + ) + + +class BaseMockReconnectLogic(ReconnectLogic): + """Mock ReconnectLogic.""" + + def stop_callback(self) -> None: + """Stop the reconnect logic.""" + # For the purposes of testing, we don't want to wait + # for the reconnect logic to finish trying to connect + self._cancel_connect("forced disconnect from test") + self._is_stopped = True + + async def stop(self) -> None: + """Stop the reconnect logic.""" + self.stop_callback() + + +@pytest.fixture +def mock_client(mock_device_info: DeviceInfo) -> APIClient: + """Mock APIClient.""" + mock_client = Mock(spec=APIClient) + + def mock_constructor( + address: str, + port: int, + password: str | None, + *, + client_info: str = "aioesphomeapi", + keepalive: float = 15.0, + zeroconf_instance: Zeroconf = None, + noise_psk: str | None = None, + expected_name: str | None = None, + ) -> APIClient: + """Fake the client constructor.""" + mock_client.host = address + mock_client.port = port + mock_client.password = password + mock_client.zeroconf_instance = zeroconf_instance + mock_client.noise_psk = noise_psk + return mock_client + + mock_client.side_effect = mock_constructor + mock_client.device_info = AsyncMock(return_value=mock_device_info) + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.list_entities_services = AsyncMock(return_value=([], [])) + mock_client.address = "127.0.0.1" + mock_client.api_version = APIVersion(99, 99) + + with ( + patch( + "aioesphomeapi.ReconnectLogic", + BaseMockReconnectLogic, + ), + patch("aioesphomeapi.APIClient", mock_client), + ): + yield mock_client