mirror of
https://github.com/home-assistant/core.git
synced 2026-08-27 18:14:46 -05:00
Add Specialized Turbo e-bike BLE integration (#168444)
This commit is contained in:
Jamie Magee
@@ -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.*
|
||||
|
||||
Generated
+2
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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"
|
||||
@@ -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)
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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"]
|
||||
}
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+66
@@ -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",
|
||||
|
||||
Generated
+1
@@ -741,6 +741,7 @@ FLOWS = {
|
||||
"songpal",
|
||||
"sonos",
|
||||
"soundtouch",
|
||||
"specialized_turbo",
|
||||
"speedtestdotnet",
|
||||
"splunk",
|
||||
"spotify",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
Generated
+3
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
@@ -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,
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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"
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user