From 66a99cf51c27f740df5e7ef855b4e7788b44fd72 Mon Sep 17 00:00:00 2001 From: Jamie Magee Date: Wed, 26 Aug 2026 06:12:38 -0700 Subject: [PATCH] Add Specialized Turbo e-bike BLE integration (#168444) --- .strict-typing | 1 + CODEOWNERS | 2 + .../components/specialized_turbo/__init__.py | 78 + .../specialized_turbo/config_flow.py | 424 +++++ .../components/specialized_turbo/const.py | 11 + .../specialized_turbo/coordinator.py | 282 ++++ .../components/specialized_turbo/icons.json | 57 + .../specialized_turbo/manifest.json | 53 + .../specialized_turbo/quality_scale.yaml | 89 + .../components/specialized_turbo/sensor.py | 351 ++++ .../components/specialized_turbo/strings.json | 149 ++ homeassistant/generated/bluetooth.py | 66 + homeassistant/generated/config_flows.py | 1 + homeassistant/generated/integrations.json | 6 + mypy.ini | 10 + requirements_all.txt | 3 + .../components/specialized_turbo/__init__.py | 25 + .../components/specialized_turbo/conftest.py | 248 +++ .../snapshots/test_sensor.ambr | 1480 +++++++++++++++++ .../specialized_turbo/test_config_flow.py | 807 +++++++++ .../components/specialized_turbo/test_init.py | 307 ++++ .../specialized_turbo/test_sensor.py | 242 +++ 22 files changed, 4692 insertions(+) create mode 100644 homeassistant/components/specialized_turbo/__init__.py create mode 100644 homeassistant/components/specialized_turbo/config_flow.py create mode 100644 homeassistant/components/specialized_turbo/const.py create mode 100644 homeassistant/components/specialized_turbo/coordinator.py create mode 100644 homeassistant/components/specialized_turbo/icons.json create mode 100644 homeassistant/components/specialized_turbo/manifest.json create mode 100644 homeassistant/components/specialized_turbo/quality_scale.yaml create mode 100644 homeassistant/components/specialized_turbo/sensor.py create mode 100644 homeassistant/components/specialized_turbo/strings.json create mode 100644 tests/components/specialized_turbo/__init__.py create mode 100644 tests/components/specialized_turbo/conftest.py create mode 100644 tests/components/specialized_turbo/snapshots/test_sensor.ambr create mode 100644 tests/components/specialized_turbo/test_config_flow.py create mode 100644 tests/components/specialized_turbo/test_init.py create mode 100644 tests/components/specialized_turbo/test_sensor.py diff --git a/.strict-typing b/.strict-typing index 2fe9686f69ce..e6f86a0703d7 100644 --- a/.strict-typing +++ b/.strict-typing @@ -548,6 +548,7 @@ homeassistant.components.snooz.* homeassistant.components.solarlog.* homeassistant.components.sonarr.* homeassistant.components.spaceapi.* +homeassistant.components.specialized_turbo.* homeassistant.components.speedtestdotnet.* homeassistant.components.spotify.* homeassistant.components.sql.* diff --git a/CODEOWNERS b/CODEOWNERS index cc03bc894041..a93fb7d21b67 100644 --- a/CODEOWNERS +++ b/CODEOWNERS @@ -1752,6 +1752,8 @@ CLAUDE.md @home-assistant/core /tests/components/soundtouch/ @kroimon /homeassistant/components/spaceapi/ @fabaff /tests/components/spaceapi/ @fabaff +/homeassistant/components/specialized_turbo/ @JamieMagee +/tests/components/specialized_turbo/ @JamieMagee /homeassistant/components/speedtestdotnet/ @rohankapoorcom @engrbm87 /tests/components/speedtestdotnet/ @rohankapoorcom @engrbm87 /homeassistant/components/splunk/ @Bre77 diff --git a/homeassistant/components/specialized_turbo/__init__.py b/homeassistant/components/specialized_turbo/__init__.py new file mode 100644 index 000000000000..3af3bbb51151 --- /dev/null +++ b/homeassistant/components/specialized_turbo/__init__.py @@ -0,0 +1,78 @@ +"""Specialized Turbo BLE integration for Home Assistant.""" + +import logging + +from specialized_turbo import BikeAdvertisement, BLEProfile, ProtocolEncryptionMethod + +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_ADDRESS, Platform +from homeassistant.core import HomeAssistant + +from .const import CONF_HMI_HARDWARE, CONF_HMI_SERIAL, CONF_WRAPPED_KEY +from .coordinator import SpecializedTurboCoordinator + +_LOGGER = logging.getLogger(__name__) + +PLATFORMS: list[Platform] = [Platform.SENSOR] + +type SpecializedTurboConfigEntry = ConfigEntry[SpecializedTurboCoordinator] + + +async def async_setup_entry( + hass: HomeAssistant, entry: SpecializedTurboConfigEntry +) -> bool: + """Set up Specialized Turbo from a config entry.""" + address: str = entry.data[CONF_ADDRESS] + wrapped_key: str | None = entry.data.get(CONF_WRAPPED_KEY) + hmi_hardware: str | None = entry.data.get(CONF_HMI_HARDWARE) + hmi_serial: str | None = entry.data.get(CONF_HMI_SERIAL) + advertisement = ( + BikeAdvertisement( + generation=BLEProfile.TCX, + encryption=ProtocolEncryptionMethod.AES_CTR, + hmi_hardware=hmi_hardware, + hmi_serial=hmi_serial, + ) + if hmi_hardware is not None and hmi_serial is not None + else None + ) + + def request_reauth(current_advertisement: BikeAdvertisement) -> None: + """Update encryption metadata and start reauthentication.""" + data = dict(entry.data) + if current_advertisement.hmi_hardware is not None: + data[CONF_HMI_HARDWARE] = current_advertisement.hmi_hardware + if current_advertisement.hmi_serial is not None: + data[CONF_HMI_SERIAL] = current_advertisement.hmi_serial + hass.config_entries.async_update_entry(entry, data=data) + entry.async_start_reauth(hass) + + coordinator = SpecializedTurboCoordinator( + hass, + _LOGGER, + address=address, + wrapped_key=wrapped_key, + advertisement=advertisement, + reauth_callback=request_reauth, + ) + + entry.runtime_data = coordinator + + await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) + + # The coordinator connects and subscribes when the first advertisement arrives. + entry.async_on_unload(coordinator.async_start()) + + return True + + +async def async_unload_entry( + hass: HomeAssistant, entry: SpecializedTurboConfigEntry +) -> bool: + """Unload a Specialized Turbo config entry.""" + unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) + + if unload_ok: + await entry.runtime_data.async_shutdown() + + return unload_ok diff --git a/homeassistant/components/specialized_turbo/config_flow.py b/homeassistant/components/specialized_turbo/config_flow.py new file mode 100644 index 000000000000..bbc16afb2181 --- /dev/null +++ b/homeassistant/components/specialized_turbo/config_flow.py @@ -0,0 +1,424 @@ +"""Config flow for Specialized Turbo bikes.""" + +from collections.abc import Callable, Mapping +from typing import Any, override + +from bleak import BleakClient +from bleak.backends.device import BLEDevice +from bleak.exc import BleakError +from bleak_retry_connector import establish_connection +from specialized_turbo import ( + BikeAdvertisement, + BikeInfo, + BLEProfile, + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + IdentificationError, + ProtocolEncryptionMethod, + SpecializedConnection, + WrappedKeyError, + is_specialized_advertisement, + parse_bike_advertisement, + parse_bike_info, + unwrap_keystore_key, +) +from specialized_turbo.cloud import CloudAuthenticationError, SpecializedCloudClient +import voluptuous as vol + +from homeassistant.components.bluetooth import ( + BluetoothServiceInfoBleak, + async_ble_device_from_address, + async_discovered_service_info, +) +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult +from homeassistant.const import CONF_ADDRESS, CONF_EMAIL, CONF_PASSWORD +from homeassistant.helpers.device_registry import format_mac +from homeassistant.helpers.httpx_client import get_async_client + +from .const import ( + CONF_HMI_HARDWARE, + CONF_HMI_SERIAL, + CONF_KEY_SOURCE, + CONF_WRAPPED_KEY, + DOMAIN, + KEY_SOURCE_ACCOUNT, + KEY_SOURCE_MANUAL, +) + + +class SpecializedTurboConfigFlow(ConfigFlow, domain=DOMAIN): + """Handle a config flow for Specialized Turbo bikes.""" + + VERSION = 3 + + def __init__(self) -> None: + """Initialize the config flow.""" + self._discovery_info: BluetoothServiceInfoBleak | None = None + self._discovered_devices: dict[str, BluetoothServiceInfoBleak] = {} + self._address: str | None = None + self._title = "Specialized Turbo" + self._advertisement: BikeAdvertisement | None = None + self._bike_info: BikeInfo | None = None + self._target_entry_id: str | None = None + + async def _async_test_connection(self) -> bool: + """Validate a legacy or advertisement-incomplete bike connection.""" + return await self._async_validate_connection() + + async def _async_validate_encrypted_connection(self, wrapped_key: str) -> bool: + """Run the encrypted identification handshake before saving an entry.""" + return await self._async_validate_connection(wrapped_key) + + async def _async_validate_connection( + self, + wrapped_key: str | None = None, + ) -> bool: + """Run upstream connection setup with Home Assistant's BLE client.""" + assert self._address is not None + address = self._address + ble_device = async_ble_device_from_address( + self.hass, + address, + connectable=True, + ) + if ble_device is None: + return False + + async def client_factory( + address_or_device: str | BLEDevice, + disconnected_callback: Callable[[BleakClient], None] | None, + ) -> BleakClient: + assert isinstance(address_or_device, BLEDevice) + return await establish_connection( + BleakClient, + address_or_device, + address, + disconnected_callback=disconnected_callback, + ) + + connection = SpecializedConnection( + ble_device, + advertisement=self._advertisement, + bike_info=self._bike_info, + wrapped_key=wrapped_key, + discovery_timeout=0, + client_factory=client_factory, + ) + try: + await connection.connect() + except ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + ): + raise + except ( + BleakError, + IdentificationError, + TimeoutError, + RuntimeError, + ValueError, + ): + return False + finally: + await connection.disconnect() + return True + + @override + async def async_step_bluetooth( + self, + discovery_info: BluetoothServiceInfoBleak, + ) -> ConfigFlowResult: + """Handle Bluetooth discovery.""" + await self.async_set_unique_id(format_mac(discovery_info.address)) + self._abort_if_unique_id_configured() + + self._set_device(discovery_info) + self.context["title_placeholders"] = { + "name": self._title, + "address": discovery_info.address, + } + return await self.async_step_bluetooth_confirm() + + async def async_step_bluetooth_confirm( + self, + user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Confirm Bluetooth discovery and collect encryption key choices.""" + assert self._discovery_info is not None + return await self._async_device_form( + "bluetooth_confirm", + user_input, + include_address=False, + ) + + @override + async def async_step_user( + self, + user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Handle a user-initiated flow.""" + if user_input is not None: + address = user_input[CONF_ADDRESS] + await self.async_set_unique_id(format_mac(address), raise_on_progress=False) + self._abort_if_unique_id_configured() + self._set_device(self._discovered_devices[address]) + return await self._async_device_form( + "user", + user_input, + include_address=True, + ) + + current_addresses = self._async_current_ids() + for info in async_discovered_service_info(self.hass): + if format_mac(info.address) in current_addresses: + continue + if _is_specialized_service_info(info): + self._discovered_devices[info.address] = info + + if not self._discovered_devices: + return self.async_abort(reason="no_devices_found") + + return self.async_show_form( + step_id="user", + data_schema=self._device_schema(include_address=True), + ) + + async def _async_device_form( + self, + step_id: str, + user_input: dict[str, Any] | None, + *, + include_address: bool, + ) -> ConfigFlowResult: + errors: dict[str, str] = {} + if user_input is not None: + if self._requires_encryption: + return await self.async_step_key_source() + try: + valid = await self._async_test_connection() + except EncryptionKeyRequiredError: + errors["base"] = "key_unavailable" + else: + if valid: + return self._create_or_update_entry({}) + errors["base"] = "cannot_connect" + + return self.async_show_form( + step_id=step_id, + data_schema=self._device_schema(include_address=include_address), + description_placeholders={ + "name": self._title, + "address": self._address or "", + }, + errors=errors, + ) + + async def async_step_key_source( + self, + _user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Choose automatic account lookup or manual wrapped key.""" + return self.async_show_menu( + step_id="key_source", + menu_options=["account", "manual_key"], + ) + + async def async_step_account( + self, + user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Fetch the bike key using Specialized account credentials.""" + errors: dict[str, str] = {} + if user_input is not None: + try: + wrapped_key = await self._async_fetch_account_key( + user_input[CONF_EMAIL], + user_input[CONF_PASSWORD], + ) + if not await self._async_validate_encrypted_connection(wrapped_key): + errors["base"] = "cannot_connect" + else: + return self._create_or_update_entry( + { + CONF_KEY_SOURCE: KEY_SOURCE_ACCOUNT, + CONF_WRAPPED_KEY: wrapped_key, + } + ) + except CloudAuthenticationError: + errors["base"] = "invalid_auth" + except ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + ): + errors["base"] = "key_unavailable" + + return self.async_show_form( + step_id="account", + data_schema=vol.Schema( + { + vol.Required(CONF_EMAIL): str, + vol.Required(CONF_PASSWORD): str, + } + ), + errors=errors, + ) + + async def async_step_manual_key( + self, + user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Accept a wrapped key obtained outside Home Assistant.""" + errors: dict[str, str] = {} + if user_input is not None: + wrapped_key = user_input[CONF_WRAPPED_KEY].strip() + try: + unwrap_keystore_key(wrapped_key) + except WrappedKeyError: + errors["base"] = "invalid_wrapped_key" + else: + try: + valid = await self._async_validate_encrypted_connection(wrapped_key) + except ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + ): + errors["base"] = "invalid_wrapped_key" + else: + if valid: + return self._create_or_update_entry( + { + CONF_KEY_SOURCE: KEY_SOURCE_MANUAL, + CONF_WRAPPED_KEY: wrapped_key, + } + ) + errors["base"] = "cannot_connect" + + return self.async_show_form( + step_id="manual_key", + data_schema=vol.Schema({vol.Required(CONF_WRAPPED_KEY): str}), + errors=errors, + ) + + async def async_step_reauth( + self, + entry_data: Mapping[str, Any], + ) -> ConfigFlowResult: + """Start reauthentication for an encrypted existing entry.""" + del entry_data + entry = self.hass.config_entries.async_get_entry(self.context["entry_id"]) + assert entry is not None + self._target_entry_id = entry.entry_id + self._address = entry.data[CONF_ADDRESS] + self._title = entry.title + hmi_hardware = entry.data.get(CONF_HMI_HARDWARE) + hmi_serial = entry.data.get(CONF_HMI_SERIAL) + if hmi_hardware is not None and hmi_serial is not None: + self._advertisement = BikeAdvertisement( + generation=BLEProfile.TCX, + encryption=ProtocolEncryptionMethod.AES_CTR, + hmi_hardware=hmi_hardware, + hmi_serial=hmi_serial, + ) + return await self.async_step_key_source() + + async def async_step_reconfigure( + self, + user_input: dict[str, Any] | None = None, + ) -> ConfigFlowResult: + """Replace the wrapped key for an encrypted bike.""" + del user_input + entry = self._get_reconfigure_entry() + hmi_hardware = entry.data.get(CONF_HMI_HARDWARE) + hmi_serial = entry.data.get(CONF_HMI_SERIAL) + if hmi_hardware is None or hmi_serial is None: + return self.async_abort(reason="not_encrypted") + self._target_entry_id = entry.entry_id + self._address = entry.data[CONF_ADDRESS] + self._title = entry.title + self._advertisement = BikeAdvertisement( + generation=BLEProfile.TCX, + encryption=ProtocolEncryptionMethod.AES_CTR, + hmi_hardware=hmi_hardware, + hmi_serial=hmi_serial, + ) + return await self.async_step_key_source() + + async def _async_fetch_account_key(self, email: str, password: str) -> str: + """Fetch a wrapped key with Home Assistant's managed HTTP client.""" + assert self._advertisement is not None + assert self._advertisement.hmi_hardware is not None + assert self._advertisement.hmi_serial is not None + cloud = SpecializedCloudClient(client=get_async_client(self.hass)) + await cloud.login(email, password) + return await cloud.get_wrapped_key( + hmi_hardware=self._advertisement.hmi_hardware, + hmi_serial=self._advertisement.hmi_serial, + ) + + def _set_device(self, info: BluetoothServiceInfoBleak) -> None: + """Store discovery data for the selected bike.""" + self._discovery_info = info + self._address = info.address + self._title = info.name or "Specialized Turbo" + self._advertisement = parse_bike_advertisement( + info.manufacturer_data, + local_name=info.name, + service_uuids=info.service_uuids, + ) + self._bike_info = parse_bike_info( + info.name or "", + info.manufacturer_data, + ) + + @property + def _requires_encryption(self) -> bool: + return ( + self._advertisement is not None + and self._advertisement.encryption == ProtocolEncryptionMethod.AES_CTR + ) + + def _device_schema(self, *, include_address: bool) -> vol.Schema: + fields: dict[vol.Marker, Any] = {} + if include_address: + fields[vol.Required(CONF_ADDRESS)] = vol.In( + { + address: f"{info.name or 'Specialized Turbo'} ({address})" + for address, info in self._discovered_devices.items() + } + ) + return vol.Schema(fields) + + def _create_or_update_entry( + self, + key_data: dict[str, Any], + ) -> ConfigFlowResult: + assert self._address is not None + data: dict[str, Any] = { + CONF_ADDRESS: self._address, + **key_data, + } + if self._advertisement is not None: + if self._advertisement.hmi_hardware is not None: + data[CONF_HMI_HARDWARE] = self._advertisement.hmi_hardware + if self._advertisement.hmi_serial is not None: + data[CONF_HMI_SERIAL] = self._advertisement.hmi_serial + + if self._target_entry_id is not None: + entry = self.hass.config_entries.async_get_entry(self._target_entry_id) + assert entry is not None + return self.async_update_reload_and_abort(entry, data_updates=data) + + return self.async_create_entry(title=self._title, data=data) + + +def _is_specialized_service_info(info: BluetoothServiceInfoBleak) -> bool: + """Check whether service information belongs to a Specialized bike.""" + return is_specialized_advertisement( + info.manufacturer_data, + local_name=info.name, + service_uuids=info.service_uuids, + ) diff --git a/homeassistant/components/specialized_turbo/const.py b/homeassistant/components/specialized_turbo/const.py new file mode 100644 index 000000000000..aaeb73672981 --- /dev/null +++ b/homeassistant/components/specialized_turbo/const.py @@ -0,0 +1,11 @@ +"""Constants for the Specialized Turbo integration.""" + +DOMAIN = "specialized_turbo" + +CONF_WRAPPED_KEY = "wrapped_key" +CONF_KEY_SOURCE = "key_source" +CONF_HMI_HARDWARE = "hmi_hardware" +CONF_HMI_SERIAL = "hmi_serial" + +KEY_SOURCE_ACCOUNT = "account" +KEY_SOURCE_MANUAL = "manual" diff --git a/homeassistant/components/specialized_turbo/coordinator.py b/homeassistant/components/specialized_turbo/coordinator.py new file mode 100644 index 000000000000..7de2450fca23 --- /dev/null +++ b/homeassistant/components/specialized_turbo/coordinator.py @@ -0,0 +1,282 @@ +"""BLE coordinator for Specialized Turbo bikes.""" + +import asyncio +from collections.abc import Callable +import logging +from typing import override + +from bleak import BleakClient +from bleak.backends.device import BLEDevice +from bleak_retry_connector import establish_connection +from specialized_turbo import ( + BikeAdvertisement, + BikeInfo, + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + SpecializedConnection, + TelemetryMonitor, + TelemetrySnapshot, + parse_bike_advertisement, + parse_bike_info, +) + +from homeassistant.components import bluetooth +from homeassistant.components.bluetooth.active_update_coordinator import ( + ActiveBluetoothDataUpdateCoordinator, +) +from homeassistant.core import HomeAssistant, callback + +_POLL_INTERVAL = 60 + + +class SpecializedTurboCoordinator( + ActiveBluetoothDataUpdateCoordinator[TelemetrySnapshot] +): + """Manage one Specialized Turbo bike through the upstream library.""" + + def __init__( + self, + hass: HomeAssistant, + logger: logging.Logger, + *, + address: str, + wrapped_key: str | None = None, + advertisement: BikeAdvertisement | None = None, + reauth_callback: Callable[[BikeAdvertisement], None] | None = None, + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass=hass, + logger=logger, + address=address, + needs_poll_method=self._needs_poll, + poll_method=self._do_poll, + mode=bluetooth.BluetoothScanningMode.ACTIVE, + connectable=True, + ) + self._address = address + self._wrapped_key = wrapped_key + self._advertisement = advertisement + self._bike_info: BikeInfo | None = None + self._connection: SpecializedConnection | None = None + self._monitor: TelemetryMonitor | None = None + self._snapshot = TelemetrySnapshot() + self._reauth_callback = reauth_callback + self._reauth_requested = False + self._poll_lock = asyncio.Lock() + self._shutdown_requested = False + self._was_unavailable = False + self.data = self._snapshot + + @property + def snapshot(self) -> TelemetrySnapshot: + """Return the current telemetry snapshot.""" + return self._snapshot + + @callback + def _needs_poll( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + seconds_since_last_poll: float | None, + ) -> bool: + """Return whether the bike needs a connection or periodic poll.""" + self._update_protocol_metadata(service_info) + return not self._shutdown_requested and ( + not self.connected + or seconds_since_last_poll is None + or seconds_since_last_poll >= _POLL_INTERVAL + ) + + async def _do_poll( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + ) -> TelemetrySnapshot: + """Connect if needed and poll the active protocol.""" + async with self._poll_lock: + if self._shutdown_requested: # pragma: no cover - scheduling race guard + return self._snapshot + await self._ensure_connected(service_info) + if self._monitor is not None: + await self._monitor.poll() + return self._snapshot + + async def _ensure_connected( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + ) -> None: + """Create the upstream connection and telemetry monitor.""" + if self.connected: + return + + self._update_protocol_metadata(service_info) + ble_device = service_info.device + + async def client_factory( + address_or_device: str | BLEDevice, + disconnected_callback: Callable[[BleakClient], None] | None, + ) -> BleakClient: + assert isinstance(address_or_device, BLEDevice) + return await establish_connection( + BleakClient, + address_or_device, + self._address, + disconnected_callback=disconnected_callback, + ) + + connection = SpecializedConnection( + ble_device, + advertisement=self._advertisement, + bike_info=self._bike_info, + wrapped_key=self._wrapped_key, + discovery_timeout=0, + disconnect_callback=self._on_disconnect, + client_factory=client_factory, + ) + try: + await connection.connect() + except ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + ): + self._request_reauth() + raise + + monitor = TelemetryMonitor( + connection, + notification_loop=self.hass.loop, + ) + monitor.on_update = self._handle_monitor_update + try: + await monitor.start(prime=False) + except Exception: + await connection.disconnect() + raise + + self._connection = connection + self._monitor = monitor + self._snapshot = monitor.snapshot + self.data = self._snapshot + + if self._was_unavailable: + self.logger.info( + "Specialized Turbo at %s is available again", self._address + ) + self._was_unavailable = False + + def _update_protocol_metadata( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + ) -> None: + """Retain the most complete advertisement and bike metadata.""" + advertisement = parse_bike_advertisement( + service_info.manufacturer_data, + local_name=service_info.name, + service_uuids=service_info.service_uuids, + ) + current_has_hmi = ( + self._advertisement is not None + and self._advertisement.hmi_hardware is not None + and self._advertisement.hmi_serial is not None + ) + new_has_hmi = ( + advertisement is not None + and advertisement.hmi_hardware is not None + and advertisement.hmi_serial is not None + ) + if advertisement is not None and (new_has_hmi or not current_has_hmi): + self._advertisement = advertisement + + bike_info = parse_bike_info( + service_info.name or "", + service_info.manufacturer_data, + ) + if bike_info.complete or (self._bike_info is None and not current_has_hmi): + self._bike_info = bike_info + + def _request_reauth(self) -> None: + """Start reauthentication once when complete HMI metadata is available.""" + if ( + self._reauth_requested + or self._reauth_callback is None + or self._advertisement is None + or self._advertisement.hmi_hardware is None + or self._advertisement.hmi_serial is None + ): + return + self._reauth_requested = True + self._reauth_callback(self._advertisement) + + def _handle_monitor_update( + self, + _message: object, + snapshot: TelemetrySnapshot, + ) -> None: + """Publish an upstream notification update to Home Assistant.""" + self._snapshot = snapshot + self.data = snapshot + self.async_update_listeners() + + @property + def connected(self) -> bool: + """Return whether the upstream BLE connection is active.""" + return self._connection is not None and self._connection.is_connected + + def _on_disconnect(self, _client: BleakClient) -> None: + """Schedule disconnect handling on the Home Assistant event loop.""" + self.hass.loop.call_soon_threadsafe(self._handle_disconnect) + + @callback + @override + def _async_handle_unavailable( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + ) -> None: + """Log and publish a passive Bluetooth availability change.""" + if self.connected: + return + if not self._was_unavailable: + self.logger.info("Specialized Turbo at %s is unavailable", self._address) + self._was_unavailable = True + super()._async_handle_unavailable(service_info) + + @callback + @override + def _async_handle_bluetooth_event( + self, + service_info: bluetooth.BluetoothServiceInfoBleak, + change: bluetooth.BluetoothChange, + ) -> None: + """Handle a Bluetooth advertisement.""" + super()._async_handle_bluetooth_event(service_info, change) + + @callback + def _handle_disconnect(self) -> None: + """Clear connection state and publish unavailability.""" + if not self._was_unavailable: + self.logger.info("Disconnected from Specialized Turbo at %s", self._address) + self._was_unavailable = True + self._connection = None + self._monitor = None + self.async_update_listeners() + + async def async_shutdown(self) -> None: + """Stop monitoring and close the upstream connection.""" + self._shutdown_requested = True + self._async_stop() + async with self._poll_lock: + monitor = self._monitor + connection = self._connection + self._monitor = None + self._connection = None + if monitor is not None: + try: + await monitor.stop() + except Exception: + self.logger.debug("Error stopping telemetry monitor", exc_info=True) + if connection is not None: + try: + await connection.disconnect() + except Exception: + self.logger.debug("Error disconnecting", exc_info=True) diff --git a/homeassistant/components/specialized_turbo/icons.json b/homeassistant/components/specialized_turbo/icons.json new file mode 100644 index 000000000000..200feef5d6dd --- /dev/null +++ b/homeassistant/components/specialized_turbo/icons.json @@ -0,0 +1,57 @@ +{ + "entity": { + "sensor": { + "altitude": { + "default": "mdi:altimeter" + }, + "altitude_gain": { + "default": "mdi:slope-uphill" + }, + "assist_eco_pct": { + "default": "mdi:leaf" + }, + "assist_level": { + "default": "mdi:lightning-bolt" + }, + "assist_trail_pct": { + "default": "mdi:pine-tree" + }, + "assist_turbo_pct": { + "default": "mdi:rocket-launch" + }, + "battery_charge_cycles": { + "default": "mdi:battery-sync" + }, + "battery_health": { + "default": "mdi:battery-heart-variant" + }, + "cadence": { + "default": "mdi:rotate-right" + }, + "consumption": { + "default": "mdi:lightning-bolt-circle" + }, + "gradient": { + "default": "mdi:slope-uphill" + }, + "kcal": { + "default": "mdi:fire" + }, + "motor_power": { + "default": "mdi:engine" + }, + "odometer": { + "default": "mdi:counter" + }, + "range_long": { + "default": "mdi:map-marker-distance" + }, + "range_short": { + "default": "mdi:map-marker-distance" + }, + "rider_power": { + "default": "mdi:bike" + } + } + } +} diff --git a/homeassistant/components/specialized_turbo/manifest.json b/homeassistant/components/specialized_turbo/manifest.json new file mode 100644 index 000000000000..380f14d7f991 --- /dev/null +++ b/homeassistant/components/specialized_turbo/manifest.json @@ -0,0 +1,53 @@ +{ + "domain": "specialized_turbo", + "name": "Specialized Turbo", + "bluetooth": [ + { + "connectable": true, + "manufacturer_data_start": [84, 85, 82, 66, 79, 72, 77, 73], + "manufacturer_id": 89 + }, + { + "connectable": true, + "manufacturer_data_start": [2, 21, 84, 85, 82, 66, 79, 72, 77, 73], + "manufacturer_id": 76 + }, + { + "connectable": true, + "local_name": "SPECIALIZED", + "manufacturer_id": 525 + }, + { + "connectable": true, + "manufacturer_id": 89, + "service_uuid": "00000001-3731-3032-494d-484f42525554" + }, + { + "connectable": true, + "manufacturer_id": 89, + "service_uuid": "00000002-3731-3032-494d-484f42525554" + }, + { + "connectable": true, + "manufacturer_id": 89, + "service_uuid": "00000003-3731-3032-494d-484f42525554" + }, + { + "connectable": true, + "local_name": "WSBC*" + }, + { + "connectable": true, + "local_name": "SPECIALIZED*" + } + ], + "codeowners": ["@JamieMagee"], + "config_flow": true, + "dependencies": ["bluetooth_adapters"], + "documentation": "https://www.home-assistant.io/integrations/specialized_turbo", + "integration_type": "device", + "iot_class": "local_push", + "loggers": ["specialized_turbo"], + "quality_scale": "silver", + "requirements": ["specialized-turbo[cloud]==0.8.2"] +} diff --git a/homeassistant/components/specialized_turbo/quality_scale.yaml b/homeassistant/components/specialized_turbo/quality_scale.yaml new file mode 100644 index 000000000000..a99fff846bf6 --- /dev/null +++ b/homeassistant/components/specialized_turbo/quality_scale.yaml @@ -0,0 +1,89 @@ +rules: + # Bronze + action-setup: + status: exempt + comment: Integration does not register service actions. + appropriate-polling: done + brands: done + common-modules: done + config-flow-test-coverage: done + config-flow: done + dependency-transparency: done + docs-actions: + status: exempt + comment: Integration does not have service actions. + docs-conditions: + status: exempt + comment: Integration does not provide conditions. + docs-high-level-description: done + docs-installation-instructions: done + docs-removal-instructions: done + docs-triggers: + status: exempt + comment: Integration does not provide triggers. + entity-event-setup: + status: exempt + comment: Integration does not subscribe to events. + entity-unique-id: done + has-entity-name: done + runtime-data: done + test-before-configure: done + test-before-setup: + status: exempt + comment: >- + BLE device may not be in range during setup. + Device is expected to be disconnected most of the time + but will connect quickly when reachable. + unique-config-entry: done + # Silver + action-exceptions: + status: exempt + comment: Integration does not register service actions. + config-entry-unloading: done + docs-configuration-parameters: + status: exempt + comment: Integration has no options flow. + docs-installation-parameters: done + entity-unavailable: done + integration-owner: done + log-when-unavailable: done + parallel-updates: done + reauthentication-flow: done + test-coverage: done + # Gold + devices: done + diagnostics: todo + discovery-update-info: + status: exempt + comment: Device is not connected to an IP network. + discovery: done + docs-data-update: done + docs-examples: done + docs-known-limitations: done + docs-supported-devices: done + docs-supported-functions: done + docs-troubleshooting: done + docs-use-cases: done + dynamic-devices: + status: exempt + comment: Only one device per config entry. New devices are set up as new entries. + entity-category: done + entity-device-class: done + entity-disabled-by-default: done + entity-translations: done + exception-translations: + status: exempt + comment: Integration does not raise exceptions with translatable messages. + icon-translations: done + reconfiguration-flow: done + repair-issues: + status: exempt + comment: No repair issues needed. + stale-devices: + status: exempt + comment: Only one device per config entry. Stale devices are removed with the config entry. + + # Platinum + async-dependency: done + inject-websession: done + strict-typing: done diff --git a/homeassistant/components/specialized_turbo/sensor.py b/homeassistant/components/specialized_turbo/sensor.py new file mode 100644 index 000000000000..d0000df23db5 --- /dev/null +++ b/homeassistant/components/specialized_turbo/sensor.py @@ -0,0 +1,351 @@ +"""Sensor platform for Specialized Turbo integration.""" + +from collections.abc import Callable +from dataclasses import dataclass +from typing import override + +from specialized_turbo import AssistLevel, TelemetrySnapshot + +from homeassistant.components.bluetooth.passive_update_coordinator import ( + PassiveBluetoothCoordinatorEntity, +) +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, + SensorStateClass, +) +from homeassistant.const import ( + CONF_ADDRESS, + PERCENTAGE, + REVOLUTIONS_PER_MINUTE, + EntityCategory, + UnitOfElectricCurrent, + UnitOfElectricPotential, + UnitOfEnergy, + UnitOfEnergyDistance, + UnitOfLength, + UnitOfPower, + UnitOfSpeed, + UnitOfTemperature, +) +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import ( + CONNECTION_BLUETOOTH, + DeviceInfo, + format_mac, +) +from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback +from homeassistant.helpers.typing import StateType + +from . import SpecializedTurboConfigEntry +from .coordinator import SpecializedTurboCoordinator + +PARALLEL_UPDATES = 0 + + +@dataclass(frozen=True, kw_only=True) +class SpecializedSensorEntityDescription(SensorEntityDescription): + """Describes a Specialized Turbo sensor entity.""" + + value_fn: Callable[[TelemetrySnapshot], StateType] + + +def _assist_level_name(snap: TelemetrySnapshot) -> str | None: + """Return assist level as a lowercase string, or None if unknown.""" + level = snap.motor.assist_level + if level is None: + return None + if isinstance(level, AssistLevel): + return level.name.lower() + return None + + +SENSOR_DESCRIPTIONS: tuple[SpecializedSensorEntityDescription, ...] = ( + # --- Battery --- + SpecializedSensorEntityDescription( + key="battery_charge_percent", + native_unit_of_measurement=PERCENTAGE, + device_class=SensorDeviceClass.BATTERY, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.battery.charge_pct, + ), + SpecializedSensorEntityDescription( + key="battery_capacity_wh", + translation_key="battery_capacity_wh", + native_unit_of_measurement=UnitOfEnergy.WATT_HOUR, + device_class=SensorDeviceClass.ENERGY_STORAGE, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.battery.capacity_wh, + ), + SpecializedSensorEntityDescription( + key="battery_remaining_wh", + translation_key="battery_remaining_wh", + native_unit_of_measurement=UnitOfEnergy.WATT_HOUR, + device_class=SensorDeviceClass.ENERGY_STORAGE, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.battery.remaining_wh, + ), + SpecializedSensorEntityDescription( + key="battery_health", + translation_key="battery_health", + native_unit_of_measurement=PERCENTAGE, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.battery.health_pct, + ), + SpecializedSensorEntityDescription( + key="battery_temp", + translation_key="battery_temp", + native_unit_of_measurement=UnitOfTemperature.CELSIUS, + device_class=SensorDeviceClass.TEMPERATURE, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.battery.temp_c, + ), + SpecializedSensorEntityDescription( + key="battery_charge_cycles", + translation_key="battery_charge_cycles", + state_class=SensorStateClass.TOTAL_INCREASING, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.battery.charge_cycles, + ), + SpecializedSensorEntityDescription( + key="battery_voltage", + translation_key="battery_voltage", + native_unit_of_measurement=UnitOfElectricPotential.VOLT, + device_class=SensorDeviceClass.VOLTAGE, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.battery.voltage_v, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="battery_current", + translation_key="battery_current", + native_unit_of_measurement=UnitOfElectricCurrent.AMPERE, + device_class=SensorDeviceClass.CURRENT, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.battery.current_a, + suggested_display_precision=1, + ), + # --- Motor / Rider --- + SpecializedSensorEntityDescription( + key="speed", + native_unit_of_measurement=UnitOfSpeed.KILOMETERS_PER_HOUR, + device_class=SensorDeviceClass.SPEED, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.motor.speed_kmh, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="rider_power", + translation_key="rider_power", + native_unit_of_measurement=UnitOfPower.WATT, + device_class=SensorDeviceClass.POWER, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.motor.rider_power_w, + ), + SpecializedSensorEntityDescription( + key="motor_power", + translation_key="motor_power", + native_unit_of_measurement=UnitOfPower.WATT, + device_class=SensorDeviceClass.POWER, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.motor.motor_power_w, + ), + SpecializedSensorEntityDescription( + key="cadence", + translation_key="cadence", + native_unit_of_measurement=REVOLUTIONS_PER_MINUTE, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda s: s.motor.cadence_rpm, + suggested_display_precision=0, + ), + SpecializedSensorEntityDescription( + key="odometer", + translation_key="odometer", + native_unit_of_measurement=UnitOfLength.KILOMETERS, + device_class=SensorDeviceClass.DISTANCE, + state_class=SensorStateClass.TOTAL_INCREASING, + value_fn=lambda s: s.motor.odometer_km, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="motor_temp", + translation_key="motor_temp", + native_unit_of_measurement=UnitOfTemperature.CELSIUS, + device_class=SensorDeviceClass.TEMPERATURE, + state_class=SensorStateClass.MEASUREMENT, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.motor.motor_temp_c, + ), + SpecializedSensorEntityDescription( + key="assist_level", + translation_key="assist_level", + device_class=SensorDeviceClass.ENUM, + options=["off", "eco", "trail", "turbo"], + value_fn=_assist_level_name, + ), + # --- Settings (informational, disabled by default) --- + SpecializedSensorEntityDescription( + key="assist_eco_pct", + translation_key="assist_eco_pct", + native_unit_of_measurement=PERCENTAGE, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.settings.assist_lev1_pct, + ), + SpecializedSensorEntityDescription( + key="assist_trail_pct", + translation_key="assist_trail_pct", + native_unit_of_measurement=PERCENTAGE, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.settings.assist_lev2_pct, + ), + SpecializedSensorEntityDescription( + key="assist_turbo_pct", + translation_key="assist_turbo_pct", + native_unit_of_measurement=PERCENTAGE, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.settings.assist_lev3_pct, + ), + # --- System (TCX2+ only, disabled by default) --- + SpecializedSensorEntityDescription( + key="range_long", + translation_key="range_long", + native_unit_of_measurement=UnitOfLength.KILOMETERS, + device_class=SensorDeviceClass.DISTANCE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.range_long_km, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="range_short", + translation_key="range_short", + native_unit_of_measurement=UnitOfLength.KILOMETERS, + device_class=SensorDeviceClass.DISTANCE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.range_short_km, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="altitude", + translation_key="altitude", + native_unit_of_measurement=UnitOfLength.METERS, + device_class=SensorDeviceClass.DISTANCE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.altitude_m, + ), + SpecializedSensorEntityDescription( + key="altitude_gain", + translation_key="altitude_gain", + native_unit_of_measurement=UnitOfLength.METERS, + device_class=SensorDeviceClass.DISTANCE, + state_class=SensorStateClass.TOTAL_INCREASING, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.altitude_gain_m, + ), + SpecializedSensorEntityDescription( + key="gradient", + translation_key="gradient", + native_unit_of_measurement=PERCENTAGE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.gradient_pct, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="system_temp", + translation_key="system_temp", + native_unit_of_measurement=UnitOfTemperature.CELSIUS, + device_class=SensorDeviceClass.TEMPERATURE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + entity_category=EntityCategory.DIAGNOSTIC, + value_fn=lambda s: s.system.system_temp_c, + ), + SpecializedSensorEntityDescription( + key="consumption", + translation_key="consumption", + native_unit_of_measurement=UnitOfEnergyDistance.WATT_HOUR_PER_KM, + device_class=SensorDeviceClass.ENERGY_DISTANCE, + state_class=SensorStateClass.MEASUREMENT, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.consumption_wh_km, + suggested_display_precision=1, + ), + SpecializedSensorEntityDescription( + key="kcal", + translation_key="kcal", + native_unit_of_measurement=UnitOfEnergy.KILO_CALORIE, + device_class=SensorDeviceClass.ENERGY, + state_class=SensorStateClass.TOTAL_INCREASING, + entity_registry_enabled_default=False, + value_fn=lambda s: s.system.kcal, + ), +) + + +async def async_setup_entry( + hass: HomeAssistant, + entry: SpecializedTurboConfigEntry, + async_add_entities: AddConfigEntryEntitiesCallback, +) -> None: + """Set up Specialized Turbo sensors from a config entry.""" + coordinator = entry.runtime_data + + async_add_entities( + SpecializedTurboSensor(coordinator, description, entry) + for description in SENSOR_DESCRIPTIONS + ) + + +class SpecializedTurboSensor( + PassiveBluetoothCoordinatorEntity[SpecializedTurboCoordinator], + SensorEntity, +): + """One telemetry field from a Specialized Turbo bike.""" + + entity_description: SpecializedSensorEntityDescription + _attr_has_entity_name = True + + def __init__( + self, + coordinator: SpecializedTurboCoordinator, + description: SpecializedSensorEntityDescription, + entry: SpecializedTurboConfigEntry, + ) -> None: + """Initialize the sensor.""" + super().__init__(coordinator) + self.entity_description = description + self._attr_unique_id = ( + f"{format_mac(entry.data[CONF_ADDRESS])}_{description.key}" + ) + self._attr_device_info = DeviceInfo( + connections={(CONNECTION_BLUETOOTH, format_mac(entry.data[CONF_ADDRESS]))}, + manufacturer="Specialized", + model="Turbo", + ) + + @property + @override + def available(self) -> bool: + """Return True when the bike is connected and has sent data.""" + return ( + super().available + and self.coordinator.connected + and self.coordinator.snapshot.message_count > 0 + ) + + @property + @override + def native_value(self) -> StateType: + """Return the sensor value from the coordinator's snapshot.""" + return self.entity_description.value_fn(self.coordinator.snapshot) diff --git a/homeassistant/components/specialized_turbo/strings.json b/homeassistant/components/specialized_turbo/strings.json new file mode 100644 index 000000000000..1fe62e59f183 --- /dev/null +++ b/homeassistant/components/specialized_turbo/strings.json @@ -0,0 +1,149 @@ +{ + "config": { + "abort": { + "already_configured": "[%key:common::config_flow::abort::already_configured_device%]", + "no_devices_found": "[%key:common::config_flow::abort::no_devices_found%]", + "not_encrypted": "This bike has no stored encryption key to reconfigure.", + "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]", + "reconfigure_successful": "[%key:common::config_flow::abort::reconfigure_successful%]" + }, + "error": { + "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", + "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", + "invalid_wrapped_key": "The wrapped key is invalid.", + "key_unavailable": "The Specialized encryption key could not be retrieved." + }, + "step": { + "account": { + "data": { + "email": "[%key:common::config_flow::data::email%]", + "password": "[%key:common::config_flow::data::password%]" + }, + "data_description": { + "email": "The email address for your Specialized account.", + "password": "The password is sent to Specialized only to retrieve the bike key." + }, + "description": "Your password is used only to retrieve this bike's encryption key. Home Assistant stores the per-bike wrapped key, not your password or account tokens.", + "title": "Sign in to Specialized" + }, + "bluetooth_confirm": { + "description": "A Specialized Turbo bike was found at {address} ({name}). Pairing prompts are handled by the active Bluetooth backend.", + "title": "Discovered Specialized Turbo bike" + }, + "key_source": { + "description": "Use your Specialized account to retrieve the bike key automatically, or enter a wrapped key manually.", + "menu_options": { + "account": "[%key:component::specialized_turbo::config::step::account::title%]", + "manual_key": "[%key:component::specialized_turbo::config::step::manual_key::title%]" + }, + "title": "Choose encryption key source" + }, + "manual_key": { + "data": { + "wrapped_key": "Wrapped key" + }, + "data_description": { + "wrapped_key": "The 64-character wrapped key for this bike." + }, + "description": "Enter the 64-character wrapped key returned by the Specialized keystore service.", + "title": "Enter wrapped bike key" + }, + "reconfigure": { + "description": "Replace the encryption key for your Specialized Turbo bike.", + "title": "Reconfigure Specialized Turbo" + }, + "user": { + "data": { + "address": "Bike" + }, + "data_description": { + "address": "Select the Specialized Turbo bike you want to add." + }, + "description": "Select your Specialized Turbo bike from the discovered devices.", + "title": "Add Specialized Turbo bike" + } + } + }, + "entity": { + "sensor": { + "altitude": { + "name": "Altitude" + }, + "altitude_gain": { + "name": "Altitude gain" + }, + "assist_eco_pct": { + "name": "ECO assist" + }, + "assist_level": { + "name": "Assist level", + "state": { + "eco": "Eco", + "off": "[%key:common::state::off%]", + "trail": "Trail", + "turbo": "Turbo" + } + }, + "assist_trail_pct": { + "name": "Trail assist" + }, + "assist_turbo_pct": { + "name": "Turbo assist" + }, + "battery_capacity_wh": { + "name": "Battery capacity" + }, + "battery_charge_cycles": { + "name": "Charge cycles" + }, + "battery_current": { + "name": "Battery current" + }, + "battery_health": { + "name": "Battery health" + }, + "battery_remaining_wh": { + "name": "Battery remaining" + }, + "battery_temp": { + "name": "Battery temperature" + }, + "battery_voltage": { + "name": "Battery voltage" + }, + "cadence": { + "name": "Cadence" + }, + "consumption": { + "name": "Consumption" + }, + "gradient": { + "name": "Gradient" + }, + "kcal": { + "name": "Calories" + }, + "motor_power": { + "name": "Motor power" + }, + "motor_temp": { + "name": "Motor temperature" + }, + "odometer": { + "name": "Odometer" + }, + "range_long": { + "name": "Range (long)" + }, + "range_short": { + "name": "Range (short)" + }, + "rider_power": { + "name": "Rider power" + }, + "system_temp": { + "name": "System temperature" + } + } + } +} diff --git a/homeassistant/generated/bluetooth.py b/homeassistant/generated/bluetooth.py index 9a160f044858..bd9c2437276a 100644 --- a/homeassistant/generated/bluetooth.py +++ b/homeassistant/generated/bluetooth.py @@ -761,6 +761,72 @@ BLUETOOTH: Final[list[dict[str, bool | str | int | list[int]]]] = [ "domain": "snooz", "service_uuid": "729f0608-496a-47fe-a124-3a62aaa3fbc0", }, + { + "connectable": True, + "domain": "specialized_turbo", + "manufacturer_data_start": [ + 84, + 85, + 82, + 66, + 79, + 72, + 77, + 73, + ], + "manufacturer_id": 89, + }, + { + "connectable": True, + "domain": "specialized_turbo", + "manufacturer_data_start": [ + 2, + 21, + 84, + 85, + 82, + 66, + 79, + 72, + 77, + 73, + ], + "manufacturer_id": 76, + }, + { + "connectable": True, + "domain": "specialized_turbo", + "local_name": "SPECIALIZED", + "manufacturer_id": 525, + }, + { + "connectable": True, + "domain": "specialized_turbo", + "manufacturer_id": 89, + "service_uuid": "00000001-3731-3032-494d-484f42525554", + }, + { + "connectable": True, + "domain": "specialized_turbo", + "manufacturer_id": 89, + "service_uuid": "00000002-3731-3032-494d-484f42525554", + }, + { + "connectable": True, + "domain": "specialized_turbo", + "manufacturer_id": 89, + "service_uuid": "00000003-3731-3032-494d-484f42525554", + }, + { + "connectable": True, + "domain": "specialized_turbo", + "local_name": "WSBC*", + }, + { + "connectable": True, + "domain": "specialized_turbo", + "local_name": "SPECIALIZED*", + }, { "connectable": False, "domain": "switchbot", diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py index 914ba1aebe41..fbe4c4feb58f 100644 --- a/homeassistant/generated/config_flows.py +++ b/homeassistant/generated/config_flows.py @@ -741,6 +741,7 @@ FLOWS = { "songpal", "sonos", "soundtouch", + "specialized_turbo", "speedtestdotnet", "splunk", "spotify", diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json index 6911bdb31f82..5336609afcd9 100644 --- a/homeassistant/generated/integrations.json +++ b/homeassistant/generated/integrations.json @@ -6912,6 +6912,12 @@ "config_flow": false, "iot_class": "local_push" }, + "specialized_turbo": { + "name": "Specialized Turbo", + "integration_type": "device", + "config_flow": true, + "iot_class": "local_push" + }, "speedtestdotnet": { "name": "Speedtest.net", "integration_type": "service", diff --git a/mypy.ini b/mypy.ini index b4f4fc9eedbb..451345b729d8 100644 --- a/mypy.ini +++ b/mypy.ini @@ -5238,6 +5238,16 @@ disallow_untyped_defs = true warn_return_any = true warn_unreachable = true +[mypy-homeassistant.components.specialized_turbo.*] +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + [mypy-homeassistant.components.speedtestdotnet.*] check_untyped_defs = true disallow_incomplete_defs = true diff --git a/requirements_all.txt b/requirements_all.txt index b22735c2b1a1..60c2c65992c2 100644 --- a/requirements_all.txt +++ b/requirements_all.txt @@ -3113,6 +3113,9 @@ sonos-websocket==0.2.0 # homeassistant.components.marytts speak2mary==1.4.0 +# homeassistant.components.specialized_turbo +specialized-turbo[cloud]==0.8.2 + # homeassistant.components.speedtestdotnet speedtest-cli==2.1.3 diff --git a/tests/components/specialized_turbo/__init__.py b/tests/components/specialized_turbo/__init__.py new file mode 100644 index 000000000000..69fb4b776f47 --- /dev/null +++ b/tests/components/specialized_turbo/__init__.py @@ -0,0 +1,25 @@ +"""Tests for the Specialized Turbo integration.""" + +from unittest.mock import patch + +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry +from tests.components.bluetooth import inject_bluetooth_service_info + + +async def setup_integration( + hass: HomeAssistant, + entry: MockConfigEntry, + service_info: BluetoothServiceInfoBleak, +) -> None: + """Set up the integration and inject one bike advertisement.""" + entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + with patch( + "homeassistant.components.bluetooth.manager.discovery_flow.async_create_flow" + ): + inject_bluetooth_service_info(hass, service_info) + await hass.async_block_till_done() diff --git a/tests/components/specialized_turbo/conftest.py b/tests/components/specialized_turbo/conftest.py new file mode 100644 index 000000000000..fc2007b38207 --- /dev/null +++ b/tests/components/specialized_turbo/conftest.py @@ -0,0 +1,248 @@ +"""Fixtures for Specialized Turbo integration tests.""" + +import base64 +from collections.abc import Generator +from dataclasses import dataclass +from unittest.mock import AsyncMock, MagicMock, patch + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +import pytest +from specialized_turbo import PRODUCTION_WRAPPING_KEY, AssistLevel, TelemetrySnapshot + +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.components.specialized_turbo.const import ( + CONF_HMI_HARDWARE, + CONF_HMI_SERIAL, + CONF_KEY_SOURCE, + CONF_WRAPPED_KEY, + DOMAIN, + KEY_SOURCE_MANUAL, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.const import CONF_ADDRESS +from homeassistant.core import HomeAssistant + +from tests.common import MockConfigEntry +from tests.components.bluetooth import generate_advertisement_data, generate_ble_device + +MOCK_ADDRESS = "DC:DD:BB:4A:D6:55" +MOCK_ADDRESS_FORMATTED = "dc:dd:bb:4a:d6:55" +MOCK_NAME = "SPECIALIZED" +MOCK_MANUFACTURER_DATA: dict[int, bytes] = {0x0059: b"TURBOHMItest1234"} +MOCK_ENCRYPTED_MANUFACTURER_DATA: dict[int, bytes] = { + 0x0059: bytes.fromhex("dac8c404423333330601") +} + +MOCK_TCU1_ADDRESS = "C6:1A:10:12:5E:48" +MOCK_TCU1_ADDRESS_FORMATTED = "c6:1a:10:12:5e:48" +MOCK_TCU1_MANUFACTURER_DATA: dict[int, bytes] = { + 0x020D: bytes.fromhex("028657" + "ff" * 24), +} + + +def make_service_info( + *, + name: str = MOCK_NAME, + address: str = MOCK_ADDRESS, + manufacturer_data: dict[int, bytes] | None = None, + service_uuids: list[str] | None = None, + time: float = 0, +) -> BluetoothServiceInfoBleak: + """Build Bluetooth service information for a bike.""" + manufacturer_data = ( + MOCK_MANUFACTURER_DATA if manufacturer_data is None else manufacturer_data + ) + service_uuids = service_uuids or [] + return BluetoothServiceInfoBleak( + name=name, + address=address, + device=generate_ble_device(address=address, name=name), + rssi=-61, + manufacturer_data=manufacturer_data, + service_data={}, + service_uuids=service_uuids, + source="local", + advertisement=generate_advertisement_data( + manufacturer_data=manufacturer_data, + service_uuids=service_uuids, + ), + connectable=True, + time=time, + tx_power=None, + ) + + +TCX_SERVICE_INFO = make_service_info() +ENCRYPTED_SERVICE_INFO = make_service_info( + manufacturer_data=MOCK_ENCRYPTED_MANUFACTURER_DATA +) +TCU1_SERVICE_INFO = make_service_info( + address=MOCK_TCU1_ADDRESS, + manufacturer_data=MOCK_TCU1_MANUFACTURER_DATA, +) +NAME_ONLY_SERVICE_INFO = make_service_info( + name="WSBC025079419R", + manufacturer_data={}, +) + + +def make_wrapped_key( + key: bytes = bytes.fromhex("00112233445566778899aabbccddeeff"), +) -> str: + """Build a valid wrapped key for config flow tests.""" + wrapping_iv = bytes(range(16)) + cipher = Cipher( + algorithms.AES(PRODUCTION_WRAPPING_KEY), + modes.CTR(wrapping_iv), + ) + encryptor = cipher.encryptor() + encrypted = encryptor.update(key.hex().encode()) + encryptor.finalize() + return base64.b64encode(wrapping_iv + encrypted).decode() + + +def make_populated_snapshot() -> TelemetrySnapshot: + """Build a snapshot containing values for every sensor.""" + snapshot = TelemetrySnapshot() + snapshot.message_count = 1 + snapshot.battery.charge_pct = 85 + snapshot.battery.capacity_wh = 700 + snapshot.battery.remaining_wh = 500 + snapshot.battery.health_pct = 95 + snapshot.battery.temp_c = 24 + snapshot.battery.charge_cycles = 12 + snapshot.battery.voltage_v = 48.2 + snapshot.battery.current_a = -3.4 + snapshot.motor.speed_kmh = 25.5 + snapshot.motor.rider_power_w = 120 + snapshot.motor.motor_power_w = 250 + snapshot.motor.cadence_rpm = 82 + snapshot.motor.odometer_km = 1234.5 + snapshot.motor.motor_temp_c = 42 + snapshot.motor.assist_level = AssistLevel.TRAIL + snapshot.settings.assist_lev1_pct = 35 + snapshot.settings.assist_lev2_pct = 70 + snapshot.settings.assist_lev3_pct = 100 + snapshot.system.range_long_km = 80 + snapshot.system.range_short_km = 35 + snapshot.system.altitude_m = 123 + snapshot.system.altitude_gain_m = 456 + snapshot.system.gradient_pct = 3.5 + snapshot.system.system_temp_c = 31 + snapshot.system.consumption_wh_km = 8.2 + snapshot.system.kcal = 640 + return snapshot + + +@dataclass +class MockLibrary: + """Mocks for the specialized-turbo runtime boundary.""" + + connection: MagicMock + monitor: MagicMock + connection_constructor: MagicMock + monitor_constructor: MagicMock + + +@pytest.fixture(autouse=True) +def mock_bluetooth(enable_bluetooth: None) -> None: + """Enable the mocked Bluetooth integration.""" + + +@pytest.fixture +def mock_config_entry() -> MockConfigEntry: + """Create a mock config entry.""" + return MockConfigEntry( + domain=DOMAIN, + version=3, + title="Mock title", + data={CONF_ADDRESS: MOCK_ADDRESS}, + unique_id=MOCK_ADDRESS_FORMATTED, + ) + + +@pytest.fixture +def encrypted_config_entry() -> MockConfigEntry: + """Create an encrypted-bike config entry.""" + return MockConfigEntry( + domain=DOMAIN, + version=3, + title="Mock title", + data={ + CONF_ADDRESS: MOCK_ADDRESS, + CONF_HMI_HARDWARE: "B.3.3", + CONF_HMI_SERIAL: "80005338", + CONF_KEY_SOURCE: KEY_SOURCE_MANUAL, + CONF_WRAPPED_KEY: make_wrapped_key(), + }, + unique_id=MOCK_ADDRESS_FORMATTED, + ) + + +@pytest.fixture +def mock_setup_entry() -> Generator[AsyncMock]: + """Prevent entry setup during config flow tests.""" + + async def setup_entry( + _hass: HomeAssistant, + entry: ConfigEntry, + ) -> bool: + runtime_data = MagicMock() + runtime_data.async_shutdown = AsyncMock() + entry.runtime_data = runtime_data + return True + + with patch( + "homeassistant.components.specialized_turbo.async_setup_entry", + side_effect=setup_entry, + ) as mock_setup: + yield mock_setup + + +@pytest.fixture +def mock_library() -> Generator[MockLibrary]: + """Mock the library boundary while retaining integration behavior.""" + connection = MagicMock() + connection.is_connected = False + + async def connect() -> None: + connection.is_connected = True + + async def disconnect() -> None: + connection.is_connected = False + + connection.connect = AsyncMock(side_effect=connect) + connection.disconnect = AsyncMock(side_effect=disconnect) + + monitor = MagicMock() + monitor.snapshot = TelemetrySnapshot() + monitor.start = AsyncMock() + monitor.stop = AsyncMock() + monitor.poll = AsyncMock(return_value=True) + + connection_constructor = MagicMock(return_value=connection) + monitor_constructor = MagicMock(return_value=monitor) + + with ( + patch( + "homeassistant.components.specialized_turbo.config_flow.async_ble_device_from_address", + return_value=TCX_SERVICE_INFO.device, + ), + patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedConnection", + new=connection_constructor, + ), + patch( + "homeassistant.components.specialized_turbo.coordinator.SpecializedConnection", + new=connection_constructor, + ), + patch( + "homeassistant.components.specialized_turbo.coordinator.TelemetryMonitor", + new=monitor_constructor, + ), + ): + yield MockLibrary( + connection=connection, + monitor=monitor, + connection_constructor=connection_constructor, + monitor_constructor=monitor_constructor, + ) diff --git a/tests/components/specialized_turbo/snapshots/test_sensor.ambr b/tests/components/specialized_turbo/snapshots/test_sensor.ambr new file mode 100644 index 000000000000..b51816450ba0 --- /dev/null +++ b/tests/components/specialized_turbo/snapshots/test_sensor.ambr @@ -0,0 +1,1480 @@ +# serializer version: 1 +# name: test_sensors[sensor.mock_title_altitude-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_altitude', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Altitude', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Altitude', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'altitude', + 'unique_id': 'dc:dd:bb:4a:d6:55_altitude', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_altitude-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'distance', + : 'Mock title Altitude', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_altitude', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '123', + }) +# --- +# name: test_sensors[sensor.mock_title_altitude_gain-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_altitude_gain', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Altitude gain', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 2, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Altitude gain', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'altitude_gain', + 'unique_id': 'dc:dd:bb:4a:d6:55_altitude_gain', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_altitude_gain-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'distance', + : 'Mock title Altitude gain', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_altitude_gain', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '456', + }) +# --- +# name: test_sensors[sensor.mock_title_assist_level-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : list([ + 'off', + 'eco', + 'trail', + 'turbo', + ]), + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_assist_level', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Assist level', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Assist level', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'assist_level', + 'unique_id': 'dc:dd:bb:4a:d6:55_assist_level', + 'unit_of_measurement': None, + }) +# --- +# name: test_sensors[sensor.mock_title_assist_level-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'enum', + : 'Mock title Assist level', + : list([ + 'off', + 'eco', + 'trail', + 'turbo', + ]), + }), + 'context': , + 'entity_id': 'sensor.mock_title_assist_level', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': 'trail', + }) +# --- +# name: test_sensors[sensor.mock_title_battery-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_battery', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery', + 'options': dict({ + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_charge_percent', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_battery-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'battery', + : 'Mock title Battery', + : , + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '85', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_capacity-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_battery_capacity', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery capacity', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery capacity', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_capacity_wh', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_capacity_wh', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_battery_capacity-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy_storage', + : 'Mock title Battery capacity', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_capacity', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '700', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_current-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_battery_current', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery current', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery current', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_current', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_current', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_battery_current-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'current', + : 'Mock title Battery current', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_current', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '-3.4', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_health-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_battery_health', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery health', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Battery health', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_health', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_health', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_health-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Battery health', + : , + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_health', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '95', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_remaining-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_battery_remaining', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery remaining', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery remaining', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_remaining_wh', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_remaining_wh', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_battery_remaining-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy_storage', + : 'Mock title Battery remaining', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_remaining', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '500', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_temperature-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_battery_temperature', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery temperature', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery temperature', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_temp', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_temp', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_battery_temperature-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'temperature', + : 'Mock title Battery temperature', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_temperature', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '24', + }) +# --- +# name: test_sensors[sensor.mock_title_battery_voltage-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_battery_voltage', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Battery voltage', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Battery voltage', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_voltage', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_voltage', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_battery_voltage-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'voltage', + : 'Mock title Battery voltage', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_battery_voltage', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '48.2', + }) +# --- +# name: test_sensors[sensor.mock_title_cadence-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_cadence', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Cadence', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Cadence', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'cadence', + 'unique_id': 'dc:dd:bb:4a:d6:55_cadence', + 'unit_of_measurement': 'rpm', + }) +# --- +# name: test_sensors[sensor.mock_title_cadence-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Cadence', + : , + : 'rpm', + }), + 'context': , + 'entity_id': 'sensor.mock_title_cadence', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '82', + }) +# --- +# name: test_sensors[sensor.mock_title_calories-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_calories', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Calories', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Calories', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'kcal', + 'unique_id': 'dc:dd:bb:4a:d6:55_kcal', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_calories-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy', + : 'Mock title Calories', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_calories', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '640', + }) +# --- +# name: test_sensors[sensor.mock_title_charge_cycles-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_charge_cycles', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Charge cycles', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Charge cycles', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'battery_charge_cycles', + 'unique_id': 'dc:dd:bb:4a:d6:55_battery_charge_cycles', + 'unit_of_measurement': None, + }) +# --- +# name: test_sensors[sensor.mock_title_charge_cycles-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Charge cycles', + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_charge_cycles', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '12', + }) +# --- +# name: test_sensors[sensor.mock_title_consumption-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_consumption', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Consumption', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Consumption', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'consumption', + 'unique_id': 'dc:dd:bb:4a:d6:55_consumption', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_consumption-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'energy_distance', + : 'Mock title Consumption', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_consumption', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '8.2', + }) +# --- +# name: test_sensors[sensor.mock_title_eco_assist-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_eco_assist', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'ECO assist', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'ECO assist', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'assist_eco_pct', + 'unique_id': 'dc:dd:bb:4a:d6:55_assist_eco_pct', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_eco_assist-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title ECO assist', + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_eco_assist', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '35', + }) +# --- +# name: test_sensors[sensor.mock_title_gradient-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_gradient', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Gradient', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Gradient', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'gradient', + 'unique_id': 'dc:dd:bb:4a:d6:55_gradient', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_gradient-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Gradient', + : , + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_gradient', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '3.5', + }) +# --- +# name: test_sensors[sensor.mock_title_motor_power-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_motor_power', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Motor power', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Motor power', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'motor_power', + 'unique_id': 'dc:dd:bb:4a:d6:55_motor_power', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_motor_power-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Mock title Motor power', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_motor_power', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '250', + }) +# --- +# name: test_sensors[sensor.mock_title_motor_temperature-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_motor_temperature', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Motor temperature', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Motor temperature', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'motor_temp', + 'unique_id': 'dc:dd:bb:4a:d6:55_motor_temp', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_motor_temperature-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'temperature', + : 'Mock title Motor temperature', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_motor_temperature', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '42', + }) +# --- +# name: test_sensors[sensor.mock_title_odometer-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_odometer', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Odometer', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Odometer', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'odometer', + 'unique_id': 'dc:dd:bb:4a:d6:55_odometer', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_odometer-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'distance', + : 'Mock title Odometer', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_odometer', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '1234.5', + }) +# --- +# name: test_sensors[sensor.mock_title_range_long-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_range_long', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Range (long)', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Range (long)', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'range_long', + 'unique_id': 'dc:dd:bb:4a:d6:55_range_long', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_range_long-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'distance', + : 'Mock title Range (long)', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_range_long', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '80', + }) +# --- +# name: test_sensors[sensor.mock_title_range_short-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_range_short', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Range (short)', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Range (short)', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'range_short', + 'unique_id': 'dc:dd:bb:4a:d6:55_range_short', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_range_short-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'distance', + : 'Mock title Range (short)', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_range_short', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '35', + }) +# --- +# name: test_sensors[sensor.mock_title_rider_power-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_rider_power', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Rider power', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 0, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Rider power', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'rider_power', + 'unique_id': 'dc:dd:bb:4a:d6:55_rider_power', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_rider_power-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'power', + : 'Mock title Rider power', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_rider_power', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '120', + }) +# --- +# name: test_sensors[sensor.mock_title_speed-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': None, + 'entity_id': 'sensor.mock_title_speed', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Speed', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'Speed', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': None, + 'unique_id': 'dc:dd:bb:4a:d6:55_speed', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_speed-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'speed', + : 'Mock title Speed', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_speed', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '25.5', + }) +# --- +# name: test_sensors[sensor.mock_title_system_temperature-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': dict({ + : , + }), + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_system_temperature', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'System temperature', + 'options': dict({ + 'sensor': dict({ + 'suggested_display_precision': 1, + }), + }), + 'original_device_class': , + 'original_icon': None, + 'original_name': 'System temperature', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'system_temp', + 'unique_id': 'dc:dd:bb:4a:d6:55_system_temp', + 'unit_of_measurement': , + }) +# --- +# name: test_sensors[sensor.mock_title_system_temperature-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'temperature', + : 'Mock title System temperature', + : , + : , + }), + 'context': , + 'entity_id': 'sensor.mock_title_system_temperature', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '31', + }) +# --- +# name: test_sensors[sensor.mock_title_trail_assist-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_trail_assist', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Trail assist', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Trail assist', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'assist_trail_pct', + 'unique_id': 'dc:dd:bb:4a:d6:55_assist_trail_pct', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_trail_assist-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Trail assist', + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_trail_assist', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '70', + }) +# --- +# name: test_sensors[sensor.mock_title_turbo_assist-entry] + EntityRegistryEntrySnapshot({ + 'aliases': list([ + None, + ]), + 'area_id': None, + 'capabilities': None, + 'config_entry_id': , + 'config_subentry_id': , + 'device_class': None, + 'device_id': , + 'disabled_by': None, + 'domain': 'sensor', + 'entity_category': , + 'entity_id': 'sensor.mock_title_turbo_assist', + 'has_entity_name': True, + 'hidden_by': None, + 'icon': None, + 'id': , + 'labels': set({ + }), + 'name': None, + 'object_id_base': 'Turbo assist', + 'options': dict({ + }), + 'original_device_class': None, + 'original_icon': None, + 'original_name': 'Turbo assist', + 'platform': 'specialized_turbo', + 'previous_unique_id': None, + 'suggested_object_id': None, + 'supported_features': 0, + 'translation_key': 'assist_turbo_pct', + 'unique_id': 'dc:dd:bb:4a:d6:55_assist_turbo_pct', + 'unit_of_measurement': '%', + }) +# --- +# name: test_sensors[sensor.mock_title_turbo_assist-state] + StateSnapshot({ + 'attributes': ReadOnlyDict({ + : 'Mock title Turbo assist', + : '%', + }), + 'context': , + 'entity_id': 'sensor.mock_title_turbo_assist', + 'last_changed': , + 'last_reported': , + 'last_updated': , + 'state': '100', + }) +# --- diff --git a/tests/components/specialized_turbo/test_config_flow.py b/tests/components/specialized_turbo/test_config_flow.py new file mode 100644 index 000000000000..7f7ab0f6f945 --- /dev/null +++ b/tests/components/specialized_turbo/test_config_flow.py @@ -0,0 +1,807 @@ +"""Tests for the Specialized Turbo config flow.""" + +from unittest.mock import ANY, AsyncMock, MagicMock, patch + +from bleak import BleakError +import pytest +from specialized_turbo import ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, + IdentificationError, +) +from specialized_turbo.cloud import CloudAuthenticationError, CloudRequestError + +from homeassistant import config_entries +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.components.specialized_turbo.const import ( + CONF_HMI_HARDWARE, + CONF_HMI_SERIAL, + CONF_KEY_SOURCE, + CONF_WRAPPED_KEY, + DOMAIN, + KEY_SOURCE_ACCOUNT, + KEY_SOURCE_MANUAL, +) +from homeassistant.config_entries import ConfigFlowResult +from homeassistant.const import CONF_ADDRESS, CONF_EMAIL, CONF_PASSWORD +from homeassistant.core import HomeAssistant +from homeassistant.data_entry_flow import FlowResultType + +from .conftest import ( + ENCRYPTED_SERVICE_INFO, + MOCK_ADDRESS, + MOCK_ADDRESS_FORMATTED, + MOCK_TCU1_ADDRESS, + NAME_ONLY_SERVICE_INFO, + TCU1_SERVICE_INFO, + TCX_SERVICE_INFO, + MockLibrary, + make_service_info, + make_wrapped_key, +) + +from tests.common import MockConfigEntry + +pytestmark = pytest.mark.usefixtures("mock_setup_entry") + + +async def _choose_key_source( + hass: HomeAssistant, + result: ConfigFlowResult, + next_step_id: str, +) -> ConfigFlowResult: + """Choose an encryption key source from the menu.""" + assert result["type"] is FlowResultType.MENU + assert result["step_id"] == "key_source" + return await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={"next_step_id": next_step_id}, + ) + + +async def test_bluetooth_discovery( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test Bluetooth discovery validates the connection and creates an entry.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCX_SERVICE_INFO, + ) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "bluetooth_confirm" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == "SPECIALIZED" + assert result["data"] == {CONF_ADDRESS: MOCK_ADDRESS} + assert result["result"].unique_id == MOCK_ADDRESS_FORMATTED + mock_library.connection.connect.assert_awaited_once() + mock_library.connection.disconnect.assert_awaited_once() + + +async def test_config_flow_uses_managed_ble_client( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test the library client factory uses bleak_retry_connector.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCX_SERVICE_INFO, + ) + await hass.config_entries.flow.async_configure(result["flow_id"], user_input={}) + client_factory = mock_library.connection_constructor.call_args.kwargs[ + "client_factory" + ] + client = MagicMock() + disconnected_callback = MagicMock() + + with patch( + "homeassistant.components.specialized_turbo.config_flow.establish_connection", + new_callable=AsyncMock, + return_value=client, + ) as establish_connection: + result_client = await client_factory( + TCX_SERVICE_INFO.device, + disconnected_callback, + ) + + assert result_client is client + establish_connection.assert_awaited_once_with( + ANY, + TCX_SERVICE_INFO.device, + MOCK_ADDRESS, + disconnected_callback=disconnected_callback, + ) + + +async def test_bluetooth_discovery_already_configured( + hass: HomeAssistant, +) -> None: + """Test Bluetooth discovery aborts for an existing bike.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={CONF_ADDRESS: MOCK_ADDRESS}, + unique_id=MOCK_ADDRESS_FORMATTED, + ) + entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCX_SERVICE_INFO, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "already_configured" + + +async def test_bluetooth_discovery_tcu1( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test Bluetooth discovery for a TCU1 bike.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCU1_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_ADDRESS: MOCK_TCU1_ADDRESS} + bike_info = mock_library.connection_constructor.call_args.kwargs["bike_info"] + assert bike_info.ble_profile is not None + + +@pytest.mark.parametrize( + "error", + [ + BleakError("failed"), + IdentificationError("failed"), + TimeoutError(), + RuntimeError("failed"), + ValueError("failed"), + ], +) +async def test_bluetooth_connection_errors( + hass: HomeAssistant, + mock_library: MockLibrary, + error: Exception, +) -> None: + """Test connection errors remain on the confirmation form.""" + mock_library.connection.connect.side_effect = error + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCX_SERVICE_INFO, + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + mock_library.connection.disconnect.assert_awaited_once() + + mock_library.connection.connect.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_bluetooth_device_unavailable( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test confirmation fails when the bike is no longer discoverable.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=TCX_SERVICE_INFO, + ) + + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_ble_device_from_address", + return_value=None, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + mock_library.connection.connect.assert_not_awaited() + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + mock_library.connection.connect.assert_awaited_once() + + +async def test_bluetooth_key_required_without_metadata( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test a late encryption requirement is reported without crashing the flow.""" + mock_library.connection.connect.side_effect = EncryptionKeyRequiredError("missing") + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=NAME_ONLY_SERVICE_INFO, + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "key_unavailable"} + + mock_library.connection.connect.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_encrypted_account_setup_uses_managed_http_client( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test account setup stores only the wrapped key and HMI identifiers.""" + wrapped_key = make_wrapped_key() + cloud = MagicMock() + cloud.login = AsyncMock() + cloud.get_wrapped_key = AsyncMock(return_value=wrapped_key) + http_client = MagicMock() + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "account") + assert result["step_id"] == "account" + + with ( + patch( + "homeassistant.components.specialized_turbo.config_flow.get_async_client", + return_value=http_client, + ), + patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedCloudClient", + return_value=cloud, + ) as cloud_constructor, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={ + CONF_EMAIL: "rider@example.com", + CONF_PASSWORD: "secret", + }, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == { + CONF_ADDRESS: MOCK_ADDRESS, + CONF_HMI_HARDWARE: "B.3.3", + CONF_HMI_SERIAL: "80005338", + CONF_KEY_SOURCE: KEY_SOURCE_ACCOUNT, + CONF_WRAPPED_KEY: wrapped_key, + } + cloud_constructor.assert_called_once_with(client=http_client) + cloud.login.assert_awaited_once_with("rider@example.com", "secret") + cloud.get_wrapped_key.assert_awaited_once_with( + hmi_hardware="B.3.3", + hmi_serial="80005338", + ) + mock_library.connection.connect.assert_awaited_once() + + +@pytest.mark.parametrize( + ("method_name", "error", "expected_error"), + [ + ("login", CloudAuthenticationError("failed"), "invalid_auth"), + ("get_wrapped_key", CloudRequestError("failed"), "key_unavailable"), + ], +) +async def test_encrypted_account_errors( + hass: HomeAssistant, + mock_library: MockLibrary, + method_name: str, + error: Exception, + expected_error: str, +) -> None: + """Test account authentication and key retrieval errors.""" + cloud = MagicMock() + cloud.login = AsyncMock() + cloud.get_wrapped_key = AsyncMock(return_value=make_wrapped_key()) + getattr(cloud, method_name).side_effect = error + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "account") + + with patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedCloudClient", + return_value=cloud, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": expected_error} + mock_library.connection.connect.assert_not_awaited() + + getattr(cloud, method_name).side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + + +@pytest.mark.parametrize( + "error", + [ + DecryptionError("stale"), + EncryptionKeyProviderError("invalid"), + EncryptionKeyRequiredError("missing"), + ], +) +async def test_encrypted_account_key_errors( + hass: HomeAssistant, + mock_library: MockLibrary, + error: Exception, +) -> None: + """Test key-specific account failures recover on the same form.""" + cloud = MagicMock() + cloud.login = AsyncMock() + cloud.get_wrapped_key = AsyncMock(return_value=make_wrapped_key()) + mock_library.connection.connect.side_effect = error + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "account") + + with patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedCloudClient", + return_value=cloud, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "key_unavailable"} + + mock_library.connection.connect.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_encrypted_account_key_fails_connection( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test an account key must complete the bike identification handshake.""" + cloud = MagicMock() + cloud.login = AsyncMock() + cloud.get_wrapped_key = AsyncMock(return_value=make_wrapped_key()) + mock_library.connection.connect.side_effect = IdentificationError("failed") + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "account") + + with patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedCloudClient", + return_value=cloud, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + + mock_library.connection.connect.side_effect = None + with patch( + "homeassistant.components.specialized_turbo.config_flow.SpecializedCloudClient", + return_value=cloud, + ): + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_EMAIL: "rider@example.com", CONF_PASSWORD: "secret"}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_encrypted_manual_key_setup( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test manual wrapped-key setup and validation.""" + wrapped_key = make_wrapped_key() + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "manual_key") + assert result["step_id"] == "manual_key" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: "invalid"}, + ) + assert result["errors"] == {"base": "invalid_wrapped_key"} + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: f" {wrapped_key} "}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"][CONF_KEY_SOURCE] == KEY_SOURCE_MANUAL + assert result["data"][CONF_WRAPPED_KEY] == wrapped_key + mock_library.connection.connect.assert_awaited_once() + + +@pytest.mark.parametrize( + "error", + [ + DecryptionError("stale"), + EncryptionKeyProviderError("invalid"), + EncryptionKeyRequiredError("missing"), + ], +) +async def test_manual_key_rejected_by_bike( + hass: HomeAssistant, + mock_library: MockLibrary, + error: Exception, +) -> None: + """Test a wrapped key rejected by the bike remains on the form.""" + mock_library.connection.connect.side_effect = error + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "manual_key") + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: make_wrapped_key()}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "invalid_wrapped_key"} + + mock_library.connection.connect.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: make_wrapped_key()}, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_manual_key_connection_failure_recovers( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test manual key setup recovers after an identification failure.""" + mock_library.connection.connect.side_effect = IdentificationError("failed") + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_BLUETOOTH}, + data=ENCRYPTED_SERVICE_INFO, + ) + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={}, + ) + result = await _choose_key_source(hass, result, "manual_key") + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: make_wrapped_key()}, + ) + + assert result["type"] is FlowResultType.FORM + assert result["errors"] == {"base": "cannot_connect"} + + mock_library.connection.connect.side_effect = None + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: make_wrapped_key()}, + ) + assert result["type"] is FlowResultType.CREATE_ENTRY + + +async def test_user_flow( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test user setup with a discovered bike.""" + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_discovered_service_info", + return_value=[TCX_SERVICE_INFO], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_ADDRESS: MOCK_ADDRESS}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["data"] == {CONF_ADDRESS: MOCK_ADDRESS} + mock_library.connection.connect.assert_awaited_once() + + +@pytest.mark.parametrize( + "service_info", + [TCU1_SERVICE_INFO, NAME_ONLY_SERVICE_INFO], + ids=["tcu1", "name_only"], +) +async def test_user_flow_discovers_supported_variants( + hass: HomeAssistant, + mock_library: MockLibrary, + service_info: BluetoothServiceInfoBleak, +) -> None: + """Test manual setup discovers TCU1 and name-only WSBC bikes.""" + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_discovered_service_info", + return_value=[service_info], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_ADDRESS: service_info.address}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + assert result["title"] == service_info.name + mock_library.connection.connect.assert_awaited_once() + + +async def test_user_flow_selects_encryption_source_after_bike( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test an encrypted user flow continues from the key-source menu.""" + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_discovered_service_info", + return_value=[ENCRYPTED_SERVICE_INFO], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert CONF_KEY_SOURCE not in result["data_schema"].schema + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_ADDRESS: MOCK_ADDRESS}, + ) + + result = await _choose_key_source(hass, result, "manual_key") + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: make_wrapped_key()}, + ) + + assert result["type"] is FlowResultType.CREATE_ENTRY + mock_library.connection.connect.assert_awaited_once() + + +@pytest.mark.parametrize( + "service_infos", + [ + [], + [ + make_service_info( + name="Other device", + address="AA:BB:CC:DD:EE:FF", + manufacturer_data={}, + ) + ], + ], + ids=["none", "unsupported"], +) +async def test_user_flow_no_supported_devices( + hass: HomeAssistant, + service_infos: list[BluetoothServiceInfoBleak], +) -> None: + """Test user setup aborts when no supported bikes are available.""" + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_discovered_service_info", + return_value=service_infos, + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + + +async def test_user_flow_filters_configured_bike(hass: HomeAssistant) -> None: + """Test configured bikes are excluded from manual setup.""" + entry = MockConfigEntry( + domain=DOMAIN, + data={CONF_ADDRESS: MOCK_ADDRESS}, + unique_id=MOCK_ADDRESS_FORMATTED, + ) + entry.add_to_hass(hass) + + with patch( + "homeassistant.components.specialized_turbo.config_flow.async_discovered_service_info", + return_value=[TCX_SERVICE_INFO], + ): + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={"source": config_entries.SOURCE_USER}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "no_devices_found" + + +async def test_reauth_adds_manual_key( + hass: HomeAssistant, + mock_library: MockLibrary, +) -> None: + """Test reauthentication updates an encrypted entry.""" + wrapped_key = make_wrapped_key() + entry = MockConfigEntry( + domain=DOMAIN, + version=3, + data={ + CONF_ADDRESS: MOCK_ADDRESS, + CONF_HMI_HARDWARE: "B.3.3", + CONF_HMI_SERIAL: "80005338", + }, + unique_id=MOCK_ADDRESS_FORMATTED, + ) + entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={ + "source": config_entries.SOURCE_REAUTH, + "entry_id": entry.entry_id, + }, + data=entry.data, + ) + + result = await _choose_key_source(hass, result, "manual_key") + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: wrapped_key}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reauth_successful" + assert entry.data[CONF_WRAPPED_KEY] == wrapped_key + mock_library.connection.connect.assert_awaited_once() + + +async def test_reconfigure_replaces_key( + hass: HomeAssistant, + encrypted_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test reconfiguration replaces the existing key.""" + encrypted_config_entry.add_to_hass(hass) + new_wrapped_key = make_wrapped_key( + bytes.fromhex("ffeeddccbbaa99887766554433221100") + ) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={ + "source": config_entries.SOURCE_RECONFIGURE, + "entry_id": encrypted_config_entry.entry_id, + }, + ) + + result = await _choose_key_source(hass, result, "manual_key") + result = await hass.config_entries.flow.async_configure( + result["flow_id"], + user_input={CONF_WRAPPED_KEY: new_wrapped_key}, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "reconfigure_successful" + assert encrypted_config_entry.data[CONF_KEY_SOURCE] == KEY_SOURCE_MANUAL + assert encrypted_config_entry.data[CONF_WRAPPED_KEY] == new_wrapped_key + mock_library.connection.connect.assert_awaited_once() + + +async def test_reconfigure_unencrypted_entry_not_supported( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, +) -> None: + """Test reconfiguration is limited to encrypted entries.""" + mock_config_entry.add_to_hass(hass) + + result = await hass.config_entries.flow.async_init( + DOMAIN, + context={ + "source": config_entries.SOURCE_RECONFIGURE, + "entry_id": mock_config_entry.entry_id, + }, + ) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "not_encrypted" diff --git a/tests/components/specialized_turbo/test_init.py b/tests/components/specialized_turbo/test_init.py new file mode 100644 index 000000000000..92bad30bc989 --- /dev/null +++ b/tests/components/specialized_turbo/test_init.py @@ -0,0 +1,307 @@ +"""Tests for Specialized Turbo integration setup.""" + +import asyncio +from datetime import timedelta +import logging +import time +from unittest.mock import ANY, AsyncMock, MagicMock, patch + +import pytest +from specialized_turbo import ( + DecryptionError, + EncryptionKeyProviderError, + EncryptionKeyRequiredError, +) + +from homeassistant import config_entries +from homeassistant.components.bluetooth import BluetoothServiceInfoBleak +from homeassistant.components.specialized_turbo.const import ( + CONF_HMI_HARDWARE, + CONF_HMI_SERIAL, + CONF_WRAPPED_KEY, + DOMAIN, +) +from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import CONF_ADDRESS +from homeassistant.core import HomeAssistant +from homeassistant.helpers.device_registry import format_mac +from homeassistant.util import dt as dt_util + +from . import setup_integration +from .conftest import ( + ENCRYPTED_SERVICE_INFO, + MOCK_ADDRESS, + MOCK_MANUFACTURER_DATA, + NAME_ONLY_SERVICE_INFO, + TCU1_SERVICE_INFO, + TCX_SERVICE_INFO, + MockLibrary, + make_service_info, +) + +from tests.common import MockConfigEntry, async_fire_time_changed +from tests.components.bluetooth import inject_bluetooth_service_info + + +async def test_setup_and_unload_entry( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test setup, polling, and unload through the library boundary.""" + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + assert mock_config_entry.state is ConfigEntryState.LOADED + mock_library.connection.connect.assert_awaited_once() + mock_library.monitor.start.assert_awaited_once_with(prime=False) + mock_library.monitor.poll.assert_awaited_once() + + assert await hass.config_entries.async_unload(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.NOT_LOADED + mock_library.monitor.stop.assert_awaited_once() + mock_library.connection.disconnect.assert_awaited_once() + + +async def test_runtime_uses_managed_ble_client( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test the runtime library client factory uses bleak_retry_connector.""" + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + client_factory = mock_library.connection_constructor.call_args.kwargs[ + "client_factory" + ] + client = MagicMock() + disconnected_callback = MagicMock() + + with patch( + "homeassistant.components.specialized_turbo.coordinator.establish_connection", + new_callable=AsyncMock, + return_value=client, + ) as establish_connection: + result_client = await client_factory( + TCX_SERVICE_INFO.device, + disconnected_callback, + ) + + assert result_client is client + establish_connection.assert_awaited_once_with( + ANY, + TCX_SERVICE_INFO.device, + MOCK_ADDRESS, + disconnected_callback=disconnected_callback, + ) + + +async def test_periodic_poll_reuses_connection( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test a periodic poll does not create another library connection.""" + with patch( + "homeassistant.components.bluetooth.active_update_coordinator.monotonic_time_coarse", + return_value=0.0, + ): + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + service_info = make_service_info( + manufacturer_data={ + **MOCK_MANUFACTURER_DATA, + 0xFFFF: b"\x01", + }, + time=time.monotonic() + 61, + ) + + with patch( + "homeassistant.components.bluetooth.manager.discovery_flow.async_create_flow" + ): + inject_bluetooth_service_info(hass, service_info) + async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=11)) + await hass.async_block_till_done() + + assert mock_library.monitor.poll.await_count == 2 + mock_library.connection_constructor.assert_called_once() + + +async def test_setup_succeeds_without_bike_in_range( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test setup does not require the bike to be awake.""" + mock_config_entry.add_to_hass(hass) + + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.LOADED + mock_library.connection_constructor.assert_not_called() + + +@pytest.mark.parametrize( + "service_info", + [TCU1_SERVICE_INFO, NAME_ONLY_SERVICE_INFO], + ids=["tcu1", "name_only"], +) +async def test_setup_supported_bike_variants( + hass: HomeAssistant, + mock_library: MockLibrary, + service_info: BluetoothServiceInfoBleak, +) -> None: + """Test setup passes parsed advertisement metadata to the library.""" + entry = MockConfigEntry( + domain=DOMAIN, + version=3, + title="Mock title", + data={CONF_ADDRESS: service_info.address}, + unique_id=format_mac(service_info.address), + ) + await setup_integration(hass, entry, service_info) + + kwargs = mock_library.connection_constructor.call_args.kwargs + assert kwargs["bike_info"] is not None + mock_library.connection.connect.assert_awaited_once() + + +async def test_setup_encrypted_entry( + hass: HomeAssistant, + encrypted_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test stored encryption metadata reaches the library connection.""" + await setup_integration(hass, encrypted_config_entry, ENCRYPTED_SERVICE_INFO) + + kwargs = mock_library.connection_constructor.call_args.kwargs + assert kwargs["wrapped_key"] == encrypted_config_entry.data[CONF_WRAPPED_KEY] + assert kwargs["advertisement"].hmi_hardware == "B.3.3" + assert kwargs["advertisement"].hmi_serial == "80005338" + + +async def test_setup_encrypted_entry_with_partial_advertisement( + hass: HomeAssistant, + encrypted_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test stored HMI metadata reconstructs bike info after a partial advertisement.""" + await setup_integration(hass, encrypted_config_entry, NAME_ONLY_SERVICE_INFO) + + kwargs = mock_library.connection_constructor.call_args.kwargs + assert kwargs["bike_info"] is None + assert kwargs["advertisement"].hmi_hardware == "B.3.3" + assert kwargs["advertisement"].hmi_serial == "80005338" + + +@pytest.mark.parametrize( + "error", + [ + EncryptionKeyRequiredError("missing"), + EncryptionKeyProviderError("invalid"), + DecryptionError("stale"), + ], +) +async def test_encryption_error_starts_reauthentication( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + error: Exception, +) -> None: + """Test missing, invalid, and stale keys start reauthentication.""" + mock_library.connection.connect.side_effect = error + await setup_integration(hass, mock_config_entry, ENCRYPTED_SERVICE_INFO) + + flows = hass.config_entries.flow.async_progress_by_handler(DOMAIN) + assert len(flows) == 1 + assert flows[0]["context"]["source"] == config_entries.SOURCE_REAUTH + assert mock_config_entry.data[CONF_HMI_HARDWARE] == "B.3.3" + assert mock_config_entry.data[CONF_HMI_SERIAL] == "80005338" + mock_library.monitor_constructor.assert_not_called() + + +async def test_key_error_without_hmi_does_not_start_broken_reauth( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test incomplete advertisements do not start an unusable reauth flow.""" + mock_library.connection.connect.side_effect = EncryptionKeyRequiredError("missing") + await setup_integration(hass, mock_config_entry, NAME_ONLY_SERVICE_INFO) + + assert hass.config_entries.flow.async_progress_by_handler(DOMAIN) == [] + + +async def test_monitor_start_failure_disconnects( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test notification setup failure closes the partial connection.""" + mock_library.monitor.start.side_effect = RuntimeError("failed") + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + assert mock_config_entry.state is ConfigEntryState.LOADED + mock_library.connection.disconnect.assert_awaited_once() + + +async def test_unload_waits_for_in_flight_connection( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, +) -> None: + """Test unload waits for a connection poll and then closes it.""" + connect_started = asyncio.Event() + connect_continue = asyncio.Event() + + async def connect() -> None: + connect_started.set() + await connect_continue.wait() + mock_library.connection.is_connected = True + + mock_library.connection.connect.side_effect = connect + mock_config_entry.add_to_hass(hass) + assert await hass.config_entries.async_setup(mock_config_entry.entry_id) + await hass.async_block_till_done() + coordinator = mock_config_entry.runtime_data + + with patch( + "homeassistant.components.bluetooth.manager.discovery_flow.async_create_flow" + ): + inject_bluetooth_service_info(hass, TCX_SERVICE_INFO) + await connect_started.wait() + + unload_task = hass.async_create_task( + hass.config_entries.async_unload(mock_config_entry.entry_id), + "Unload Specialized Turbo during connection", + ) + await asyncio.sleep(0) + assert not unload_task.done() + + connect_continue.set() + assert await unload_task + await hass.async_block_till_done() + + assert mock_config_entry.state is ConfigEntryState.NOT_LOADED + mock_library.monitor.stop.assert_awaited_once() + mock_library.connection.disconnect.assert_awaited_once() + assert coordinator.connected is False + + +async def test_unload_tolerates_library_cleanup_errors( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test cleanup errors do not prevent config entry unloading.""" + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + mock_library.monitor.stop.side_effect = RuntimeError("stop failed") + mock_library.connection.disconnect.side_effect = RuntimeError("disconnect failed") + + with caplog.at_level(logging.DEBUG): + assert await hass.config_entries.async_unload(mock_config_entry.entry_id) + await hass.async_block_till_done() + + assert "Error stopping telemetry monitor" in caplog.text + assert "Error disconnecting" in caplog.text diff --git a/tests/components/specialized_turbo/test_sensor.py b/tests/components/specialized_turbo/test_sensor.py new file mode 100644 index 000000000000..016ae065c1d4 --- /dev/null +++ b/tests/components/specialized_turbo/test_sensor.py @@ -0,0 +1,242 @@ +"""Tests for Specialized Turbo sensor entities.""" + +from datetime import timedelta +import logging +import time +from unittest.mock import MagicMock, patch + +from bleak import BleakError +import pytest +from syrupy.assertion import SnapshotAssertion + +from homeassistant.components.bluetooth import ( + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS, +) +from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN +from homeassistant.components.specialized_turbo.const import DOMAIN +from homeassistant.const import STATE_UNAVAILABLE, STATE_UNKNOWN +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry as er +from homeassistant.util import dt as dt_util + +from . import setup_integration +from .conftest import ( + MOCK_ADDRESS_FORMATTED, + MOCK_MANUFACTURER_DATA, + TCX_SERVICE_INFO, + MockLibrary, + make_populated_snapshot, + make_service_info, +) + +from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform +from tests.components.bluetooth import ( + inject_bluetooth_service_info, + patch_all_discovered_devices, + patch_bluetooth_time, +) + + +def _entity_id(entity_registry: er.EntityRegistry, key: str) -> str: + """Return an entity ID from its stable integration unique ID.""" + entity_id = entity_registry.async_get_entity_id( + SENSOR_DOMAIN, + DOMAIN, + f"{MOCK_ADDRESS_FORMATTED}_{key}", + ) + assert entity_id is not None + return entity_id + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_sensors( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, + snapshot: SnapshotAssertion, +) -> None: + """Test all sensor metadata and values.""" + mock_library.monitor.snapshot = make_populated_snapshot() + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + await snapshot_platform( + hass, + entity_registry, + snapshot, + mock_config_entry.entry_id, + ) + + +async def test_sensors_unavailable_before_first_message( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, +) -> None: + """Test sensors remain unavailable before telemetry arrives.""" + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + state = hass.states.get(_entity_id(entity_registry, "battery_charge_percent")) + assert state is not None + assert state.state == STATE_UNAVAILABLE + + +async def test_notification_updates_entities( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, +) -> None: + """Test a library notification updates entity state.""" + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + updated = make_populated_snapshot() + + callback = mock_library.monitor.on_update + assert callable(callback) + callback(MagicMock(), updated) + await hass.async_block_till_done() + + battery = hass.states.get(_entity_id(entity_registry, "battery_charge_percent")) + speed = hass.states.get(_entity_id(entity_registry, "speed")) + assist = hass.states.get(_entity_id(entity_registry, "assist_level")) + assert battery is not None + assert speed is not None + assert assist is not None + assert battery.state == "85" + assert speed.state == "25.5" + assert assist.state == "trail" + + +async def test_disconnect_marks_entities_unavailable( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test a library disconnect and reconnect update availability.""" + mock_library.monitor.snapshot = make_populated_snapshot() + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + battery_entity_id = _entity_id(entity_registry, "battery_charge_percent") + battery = hass.states.get(battery_entity_id) + assert battery is not None + assert battery.state == "85" + + disconnect_callback = mock_library.connection_constructor.call_args.kwargs[ + "disconnect_callback" + ] + with caplog.at_level(logging.INFO): + mock_library.connection.is_connected = False + disconnect_callback(mock_library.connection) + await hass.async_block_till_done() + + battery = hass.states.get(battery_entity_id) + assert battery is not None + assert battery.state == STATE_UNAVAILABLE + + with patch( + "homeassistant.components.bluetooth.manager.discovery_flow.async_create_flow" + ): + inject_bluetooth_service_info( + hass, + make_service_info( + manufacturer_data={ + **MOCK_MANUFACTURER_DATA, + 0xFFFF: b"\x01", + } + ), + ) + async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=11)) + await hass.async_block_till_done() + + battery = hass.states.get(battery_entity_id) + assert battery is not None + assert battery.state == "85" + assert ( + caplog.text.count( + "Specialized Turbo at DC:DD:BB:4A:D6:55 is available again" + ) + == 1 + ) + + +async def test_stale_advertisement_keeps_connected_entities_available( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test Bluetooth expiry does not override an active GATT connection.""" + start_monotonic = time.monotonic() + mock_library.monitor.snapshot = make_populated_snapshot() + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + battery_entity_id = _entity_id(entity_registry, "battery_charge_percent") + + monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 + with caplog.at_level(logging.INFO): + with ( + patch_bluetooth_time(monotonic_now), + patch_all_discovered_devices([]), + ): + async_fire_time_changed( + hass, + dt_util.utcnow() + + timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1), + ) + await hass.async_block_till_done() + + battery = hass.states.get(battery_entity_id) + assert battery is not None + assert battery.state == "85" + assert mock_library.connection.is_connected is True + assert "is unavailable" not in caplog.text + + +async def test_stale_advertisement_logs_failed_connection_unavailable( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test advertisement expiry logs an unavailable disconnected bike once.""" + start_monotonic = time.monotonic() + mock_library.connection.connect.side_effect = BleakError("failed") + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + monotonic_now = start_monotonic + FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1 + with ( + caplog.at_level(logging.INFO), + patch_bluetooth_time(monotonic_now), + patch_all_discovered_devices([]), + ): + async_fire_time_changed( + hass, + dt_util.utcnow() + + timedelta(seconds=FALLBACK_MAXIMUM_STALE_ADVERTISEMENT_SECONDS + 1), + ) + await hass.async_block_till_done() + + assert ( + caplog.text.count("Specialized Turbo at DC:DD:BB:4A:D6:55 is unavailable") == 1 + ) + + +@pytest.mark.parametrize("assist_level", [None, 99]) +async def test_unknown_assist_level( + hass: HomeAssistant, + mock_config_entry: MockConfigEntry, + mock_library: MockLibrary, + entity_registry: er.EntityRegistry, + assist_level: int | None, +) -> None: + """Test an unknown assist value produces an unknown entity state.""" + snapshot = make_populated_snapshot() + snapshot.motor.assist_level = assist_level + mock_library.monitor.snapshot = snapshot + await setup_integration(hass, mock_config_entry, TCX_SERVICE_INFO) + + assist = hass.states.get(_entity_id(entity_registry, "assist_level")) + assert assist is not None + assert assist.state == STATE_UNKNOWN