From db9bcd62be212ab3377581487902186c055f753b Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 26 Aug 2026 21:42:16 +1000 Subject: [PATCH] Stream Teslemetry energy live status, site info and tariff over SSE (#178735) --- .../components/teslemetry/__init__.py | 214 +++++++++++++----- .../components/teslemetry/coordinator.py | 88 +++++-- .../components/teslemetry/strings.json | 3 + tests/components/teslemetry/conftest.py | 98 +++++++- tests/components/teslemetry/const.py | 24 ++ tests/components/teslemetry/test_calendar.py | 114 +++++++++- tests/components/teslemetry/test_init.py | 159 ++++++++++--- tests/components/teslemetry/test_select.py | 23 +- tests/components/teslemetry/test_sensor.py | 63 +++++- 9 files changed, 663 insertions(+), 123 deletions(-) diff --git a/homeassistant/components/teslemetry/__init__.py b/homeassistant/components/teslemetry/__init__.py index e32579c3cac7..f51bcb69d810 100644 --- a/homeassistant/components/teslemetry/__init__.py +++ b/homeassistant/components/teslemetry/__init__.py @@ -14,8 +14,9 @@ from tesla_fleet_api.exceptions import ( SubscriptionRequired, TeslaFleetError, ) -from tesla_fleet_api.teslemetry import Teslemetry +from tesla_fleet_api.teslemetry import EnergySite, Teslemetry from teslemetry_stream import TeslemetryStream +from teslemetry_stream.const import SseTopic from homeassistant.components.application_credentials import ( ClientCredential, @@ -43,6 +44,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import ( ) from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.typing import ConfigType +from homeassistant.helpers.update_coordinator import UpdateFailed from .const import CLIENT_ID, DOMAIN, LOGGER, VEHICLE_ISSUE_LEARN_MORE from .coordinator import ( @@ -76,6 +78,19 @@ type TeslemetryConfigEntry = ConfigEntry[TeslemetryData] CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN) +# Exact SSE topics the integration consumes. An explicit allowlist keeps a +# new server topic from silently adding traffic or data exposure to HA. +STREAM_TOPICS: Final = ( + SseTopic.STATE, + SseTopic.VEHICLE_DATA, + SseTopic.DATA, + SseTopic.CONNECTIVITY, + SseTopic.CREDITS, + SseTopic.LIVE_STATUS, + SseTopic.SITE_INFO, + SseTopic.TARIFF_CONTENT_V2, +) + async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the Telemetry integration.""" @@ -314,9 +329,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) - vehicles: list[TeslemetryVehicleData] = [] energysites: list[TeslemetryEnergyData] = [] - # Create the stream (created lazily when first vehicle is found) + # Create the stream (created lazily for the first eligible vehicle or + # energy site, so energy-only accounts still open the account stream) stream: TeslemetryStream | None = None + def create_stream() -> TeslemetryStream: + return TeslemetryStream( + session, + access_token, + server=f"{region.lower()}.teslemetry.com", + parse_timestamp=True, + manual=True, + topics=STREAM_TOPICS, + ) + # Remember each device identifier we create current_devices: set[tuple[str, str]] = set() @@ -334,13 +360,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) - # Create stream if required (for first vehicle) if not stream: - stream = TeslemetryStream( - session, - access_token, - server=f"{region.lower()}.teslemetry.com", - parse_timestamp=True, - manual=True, - ) + stream = create_stream() # Remove the protobuff 'cached_data' that we do not use to save memory product.pop("cached_data", None) @@ -404,6 +424,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) - ) continue + # Create stream if required (for first energy site) + if not stream: + stream = create_stream() + current_devices.add((DOMAIN, str(site_id))) if wall_connector: current_devices |= { @@ -419,53 +443,25 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) - serial_number=str(site_id), ) - # For initial setup, raise auth errors properly - try: - live_status = (await energy_site.live_status())["response"] - except InvalidToken as e: - raise ConfigEntryAuthFailed( - translation_domain=DOMAIN, - translation_key="auth_failed_invalid_token", - ) from e - except LoginRequired as e: - raise ConfigEntryAuthFailed( - translation_domain=DOMAIN, - translation_key="auth_failed_login_required", - ) from e - except SubscriptionRequired as e: - raise ConfigEntryAuthFailed( - translation_domain=DOMAIN, - translation_key="auth_failed_subscription_required", - ) from e - except Forbidden as e: - raise ConfigEntryAuthFailed( - translation_domain=DOMAIN, - translation_key="auth_failed_invalid_token", - ) from e - except TeslaFleetError as e: - raise ConfigEntryNotReady( - translation_domain=DOMAIN, - translation_key="not_ready_api_error", - ) from e - + ( + live_coordinator, + info_coordinator, + history_coordinator, + ) = await _async_setup_energy_site( + hass, + entry, + stream, + energy_site, + product, + site_id, + powerwall, + ) energysites.append( TeslemetryEnergyData( api=energy_site, - live_coordinator=( - TeslemetryEnergySiteLiveCoordinator( - hass, entry, energy_site, live_status - ) - if isinstance(live_status, dict) - else None - ), - info_coordinator=TeslemetryEnergySiteInfoCoordinator( - hass, entry, energy_site, product - ), - history_coordinator=( - TeslemetryEnergyHistoryCoordinator(hass, entry, energy_site) - if powerwall - else None - ), + live_coordinator=live_coordinator, + info_coordinator=info_coordinator, + history_coordinator=history_coordinator, id=site_id, device=device, ) @@ -526,11 +522,123 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) - if stream: entry.async_on_unload(stream.close) + # The stream is the only freshness signal for the energy coordinators, so + # a dropped connection must mark their entities unavailable rather than + # leaving stale live/info/tariff data available indefinitely. + if energysites: + entry.async_on_unload( + stream.async_add_connection_listener( + create_handle_energy_stream_connection(energysites) + ) + ) entry.async_create_background_task(hass, stream.listen(), "Teslemetry Stream") return True +def create_handle_energy_stream_connection( + energysites: list[TeslemetryEnergyData], +) -> Callable[[bool], None]: + """Create a stream connection listener for the energy coordinators.""" + + @callback + def handle_connection(connected: bool) -> None: + """Fail stream-driven energy coordinators while the stream is down. + + Each subsequent streamed document restores its coordinator via + async_set_updated_data, so no reload is required on reconnect. + """ + if connected: + return + error = UpdateFailed( + translation_domain=DOMAIN, + translation_key="stream_disconnected", + ) + for energysite in energysites: + if energysite.live_coordinator is not None: + energysite.live_coordinator.async_set_update_error(error) + energysite.info_coordinator.async_set_update_error(error) + + return handle_connection + + +async def _async_setup_energy_site( + hass: HomeAssistant, + entry: TeslemetryConfigEntry, + stream: TeslemetryStream, + energy_site: EnergySite, + product: dict[str, Any], + site_id: int, + powerwall: Any, +) -> tuple[ + TeslemetryEnergySiteLiveCoordinator | None, + TeslemetryEnergySiteInfoCoordinator, + TeslemetryEnergyHistoryCoordinator | None, +]: + """Cold-read live status, build the energy coordinators, and register listeners.""" + # The stream has no ready boundary, so keep a deterministic REST cold read + # for setup auth/error handling before switching to listener-driven updates. + try: + live_status = (await energy_site.live_status())["response"] + except InvalidToken as e: + raise ConfigEntryAuthFailed( + translation_domain=DOMAIN, + translation_key="auth_failed_invalid_token", + ) from e + except LoginRequired as e: + raise ConfigEntryAuthFailed( + translation_domain=DOMAIN, + translation_key="auth_failed_login_required", + ) from e + except SubscriptionRequired as e: + raise ConfigEntryAuthFailed( + translation_domain=DOMAIN, + translation_key="auth_failed_subscription_required", + ) from e + except Forbidden as e: + raise ConfigEntryAuthFailed( + translation_domain=DOMAIN, + translation_key="auth_failed_invalid_token", + ) from e + except TeslaFleetError as e: + raise ConfigEntryNotReady( + translation_domain=DOMAIN, + translation_key="not_ready_api_error", + ) from e + + live_coordinator = ( + TeslemetryEnergySiteLiveCoordinator(hass, entry, energy_site, live_status) + if isinstance(live_status, dict) + else None + ) + info_coordinator = TeslemetryEnergySiteInfoCoordinator( + hass, entry, energy_site, product + ) + + # Register before stream.listen() so the opening snapshot cannot be missed. + stream_energysite = stream.get_energysite(site_id) + if live_coordinator is not None: + entry.async_on_unload( + stream_energysite.listen_LiveStatus(live_coordinator.handle_stream_update) + ) + entry.async_on_unload( + stream_energysite.listen_SiteInfo(info_coordinator.handle_site_info) + ) + entry.async_on_unload( + stream_energysite.listen_TariffContentV2( + info_coordinator.handle_tariff_content_v2 + ) + ) + + history_coordinator = ( + TeslemetryEnergyHistoryCoordinator(hass, entry, energy_site) + if powerwall + else None + ) + + return live_coordinator, info_coordinator, history_coordinator + + async def async_unload_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool: """Unload Teslemetry Config.""" return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) diff --git a/homeassistant/components/teslemetry/coordinator.py b/homeassistant/components/teslemetry/coordinator.py index 43f512d53594..38af78da7aad 100644 --- a/homeassistant/components/teslemetry/coordinator.py +++ b/homeassistant/components/teslemetry/coordinator.py @@ -45,11 +45,13 @@ def _get_retry_after(e: TeslaFleetError) -> float: VEHICLE_INTERVAL = timedelta(seconds=60) VEHICLE_WAIT = timedelta(minutes=15) -ENERGY_LIVE_INTERVAL = timedelta(seconds=30) -ENERGY_INFO_INTERVAL = timedelta(seconds=30) ENERGY_HISTORY_INTERVAL = timedelta(seconds=60) METADATA_INTERVAL = timedelta(hours=1) +# Keys within tariff_content_v2 kept as nested dicts rather than flattened, +# since entities and calendars read them as whole structures. +TARIFF_SKIP_KEYS = ["daily_charges", "demand_charges", "energy_charges", "seasons"] + # Insufficient credits will not resolve themselves quickly, so back off polling # instead of hammering the API at the coordinator's normal interval. INSUFFICIENT_CREDITS_RETRY_AFTER = timedelta(hours=1).total_seconds() @@ -175,8 +177,21 @@ class TeslemetryVehicleDataCoordinator(DataUpdateCoordinator[dict[str, Any]]): return data +def _index_wall_connectors(data: dict[str, Any]) -> dict[str, Any]: + """Convert the live_status wall_connectors list into a DIN-keyed dict.""" + data["wall_connectors"] = { + wc["din"]: wc for wc in (data.get("wall_connectors") or []) + } + return data + + class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]): - """Class to manage fetching energy site live status from the Teslemetry API.""" + """Class to manage energy site live status from the Teslemetry stream. + + Updates are driven by ``live_status`` stream events; the REST update + method is retained for the deterministic setup cold read and manual + recovery only. + """ config_entry: TeslemetryConfigEntry updated_once: bool @@ -194,15 +209,13 @@ class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]) LOGGER, config_entry=config_entry, name="Teslemetry Energy Site Live", - update_interval=ENERGY_LIVE_INTERVAL, ) self.api = api + self.data = _index_wall_connectors(data) - # Convert Wall Connectors from array to dict - data["wall_connectors"] = { - wc["din"]: wc for wc in (data.get("wall_connectors") or []) - } - self.data = data + def handle_stream_update(self, data: dict[str, Any]) -> None: + """Handle a live_status document from the stream.""" + self.async_set_updated_data(_index_wall_connectors(data)) @override async def _async_update_data(self) -> dict[str, Any]: @@ -224,15 +237,18 @@ class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]) translation_key="update_failed", translation_placeholders={"message": e.message}, ) from e - # Convert Wall Connectors from array to dict - data["wall_connectors"] = { - wc["din"]: wc for wc in (data.get("wall_connectors") or []) - } - return data + return _index_wall_connectors(data) class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]): - """Class to manage fetching energy site info from the Teslemetry API.""" + """Class to manage energy site info from the Teslemetry stream. + + Site info and the V2 tariff are two independently replaceable partitions. + The flattened coordinator view is recomposed from both whenever either + changes, so a removed field or a cleared tariff never lingers. Stream + events drive updates; the REST update method is retained for the + deterministic setup cold read and manual recovery only. + """ config_entry: TeslemetryConfigEntry @@ -249,11 +265,46 @@ class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]) LOGGER, config_entry=config_entry, name="Teslemetry Energy Site Info", - update_interval=ENERGY_INFO_INTERVAL, ) self.api = api + self._site_info: dict[str, Any] = product + self._tariff_content_v2: dict[str, Any] | None = None self.data = product + def _compose(self) -> dict[str, Any]: + """Flatten the two partitions into the coordinator view.""" + result = flatten(self._site_info, skip_keys=TARIFF_SKIP_KEYS) + if self._tariff_content_v2 is not None: + result.update( + flatten( + {"tariff_content_v2": self._tariff_content_v2}, + skip_keys=TARIFF_SKIP_KEYS, + ) + ) + return result + + def _ingest_site_info(self, site_info: dict[str, Any]) -> dict[str, Any]: + """Split a full REST site_info response into both partitions. + + The REST document carries both tariff versions inline; the V2 tariff + moves to its own partition (matching the slim stream event shape) so + removal semantics stay identical across both delivery paths. + """ + site_info = dict(site_info) + self._tariff_content_v2 = site_info.pop("tariff_content_v2", None) + self._site_info = site_info + return self._compose() + + def handle_site_info(self, site_info: dict[str, Any]) -> None: + """Handle a slim site_info document from the stream.""" + self._site_info = site_info + self.async_set_updated_data(self._compose()) + + def handle_tariff_content_v2(self, tariff: dict[str, Any] | None) -> None: + """Handle a V2 tariff document (or removal) from the stream.""" + self._tariff_content_v2 = tariff + self.async_set_updated_data(self._compose()) + @override async def _async_update_data(self) -> dict[str, Any]: """Update energy site data using Teslemetry API.""" @@ -275,10 +326,7 @@ class TeslemetryEnergySiteInfoCoordinator(DataUpdateCoordinator[dict[str, Any]]) translation_placeholders={"message": e.message}, ) from e - return flatten( - data, - skip_keys=["daily_charges", "demand_charges", "energy_charges", "seasons"], - ) + return self._ingest_site_info(data) class TeslemetryEnergyHistoryCoordinator(DataUpdateCoordinator[dict[str, Any]]): diff --git a/homeassistant/components/teslemetry/strings.json b/homeassistant/components/teslemetry/strings.json index 37ca7a831ca9..c4059041ff42 100644 --- a/homeassistant/components/teslemetry/strings.json +++ b/homeassistant/components/teslemetry/strings.json @@ -1203,6 +1203,9 @@ "set_scheduled_departure_preconditioning": { "message": "Preconditioning departure time is required when enabling" }, + "stream_disconnected": { + "message": "Disconnected from the Teslemetry stream" + }, "token_data_malformed": { "message": "Token data malformed, try reauthenticating" }, diff --git a/tests/components/teslemetry/conftest.py b/tests/components/teslemetry/conftest.py index 17e8e45eef35..0b2c0f244ce5 100644 --- a/tests/components/teslemetry/conftest.py +++ b/tests/components/teslemetry/conftest.py @@ -3,7 +3,7 @@ from collections.abc import Generator from copy import deepcopy from typing import Any -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from teslemetry_stream.stream import recursive_match @@ -15,8 +15,10 @@ from .const import ( ENERGY_HISTORY, LIVE_STATUS, METADATA, + METADATA_ENERGY, METADATA_LEGACY, PRODUCTS, + PRODUCTS_ENERGY, SITE_INFO, VEHICLE_DATA, WAKE_UP_ONLINE, @@ -184,6 +186,100 @@ def mock_add_listener(): yield mock_add_listener +@pytest.fixture +def mock_add_connection_listener(): + """Mock Teslemetry Stream add connection listener method.""" + with patch( + "teslemetry_stream.TeslemetryStream.async_add_connection_listener", + ) as mock_add_connection_listener: + mock_add_connection_listener.listeners = [] + + def unsubscribe() -> None: + return + + def side_effect(callback): + mock_add_connection_listener.listeners.append(callback) + return unsubscribe + + def send(connected: bool) -> None: + for listener in mock_add_connection_listener.listeners: + listener(connected) + + mock_add_connection_listener.send = send + mock_add_connection_listener.side_effect = side_effect + yield mock_add_connection_listener + + +@pytest.fixture +def mock_energy_live_stream() -> Generator[MagicMock]: + """Capture the callback the integration registers for live_status events.""" + with patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_LiveStatus", + ) as mock_listen: + callbacks: list = [] + + def side_effect(callback): + callbacks.append(callback) + return MagicMock() + + def send(live_status: dict) -> None: + for callback in callbacks: + callback(live_status) + + mock_listen.side_effect = side_effect + mock_listen.send = send + yield mock_listen + + +@pytest.fixture +def mock_energy_info_stream() -> Generator[MagicMock]: + """Capture the callback the integration registers for site_info events.""" + with patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_SiteInfo", + ) as mock_listen: + callbacks: list = [] + + def side_effect(callback): + callbacks.append(callback) + return MagicMock() + + def send(site_info: dict) -> None: + for callback in callbacks: + callback(site_info) + + mock_listen.side_effect = side_effect + mock_listen.send = send + yield mock_listen + + +@pytest.fixture +def mock_energy_tariff_stream() -> Generator[MagicMock]: + """Capture the callback the integration registers for tariff events.""" + with patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_TariffContentV2", + ) as mock_listen: + callbacks: list = [] + + def side_effect(callback): + callbacks.append(callback) + return MagicMock() + + def send(tariff: dict | None) -> None: + for callback in callbacks: + callback(tariff) + + mock_listen.side_effect = side_effect + mock_listen.send = send + yield mock_listen + + +@pytest.fixture +def mock_energy_only(mock_products: AsyncMock, mock_metadata: MagicMock) -> None: + """Patch products and metadata to an energy-only account.""" + mock_products.return_value = PRODUCTS_ENERGY + mock_metadata.return_value = METADATA_ENERGY + + @pytest.fixture(autouse=True) def mock_stream_get_config(): """Mock Teslemetry Stream listen method.""" diff --git a/tests/components/teslemetry/const.py b/tests/components/teslemetry/const.py index 547e998ce393..ae574bc9470f 100644 --- a/tests/components/teslemetry/const.py +++ b/tests/components/teslemetry/const.py @@ -145,3 +145,27 @@ METADATA_NOSCOPE = { } }, } + +# Energy-only account: no accessible vehicle, one accessible energy site. +METADATA_ENERGY = { + "uid": UNIQUE_ID, + "region": "NA", + "scopes": [ + "openid", + "offline_access", + "user_data", + "energy_device_data", + "energy_cmds", + ], + "vehicles": {}, + "energy_sites": { + "123456": { + "access": True, + "name": "Energy Site", + } + }, +} +PRODUCTS_ENERGY = load_json_object_fixture("products.json", DOMAIN) +PRODUCTS_ENERGY["response"] = [ + product for product in PRODUCTS_ENERGY["response"] if "energy_site_id" in product +] diff --git a/tests/components/teslemetry/test_calendar.py b/tests/components/teslemetry/test_calendar.py index dbe79a473e17..ca5a2ed29722 100644 --- a/tests/components/teslemetry/test_calendar.py +++ b/tests/components/teslemetry/test_calendar.py @@ -3,7 +3,7 @@ from collections.abc import Generator from copy import deepcopy from datetime import datetime -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from freezegun.api import FrozenDateTimeFactory import pytest @@ -15,7 +15,12 @@ from homeassistant.components.calendar import ( EVENT_START_DATETIME, SERVICE_GET_EVENTS, ) -from homeassistant.const import ATTR_ENTITY_ID, Platform +from homeassistant.const import ( + ATTR_ENTITY_ID, + STATE_UNAVAILABLE, + STATE_UNKNOWN, + Platform, +) from homeassistant.core import HomeAssistant from homeassistant.helpers import entity_registry as er from homeassistant.util import dt as dt_util @@ -25,6 +30,14 @@ from .const import SITE_INFO, SITE_INFO_MULTI_SEASON, SITE_INFO_WEEK_CROSSING ENTITY_BUY = "calendar.energy_site_buy_tariff" ENTITY_SELL = "calendar.energy_site_sell_tariff" +ENTITY_OPERATION_MODE = "select.energy_site_operation_mode" + +TARIFF_V2 = SITE_INFO["response"]["tariff_content_v2"] +SLIM_SITE_INFO = { + key: value + for key, value in SITE_INFO["response"].items() + if key != "tariff_content_v2" +} @pytest.fixture @@ -469,3 +482,100 @@ async def test_calendar_invalid_price( assert state assert state.state == "on" assert "Unknown Price" in state.attributes["message"] + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +@pytest.mark.parametrize( + "order", + [ + pytest.param(("site_info", "tariff"), id="site_info_first"), + pytest.param(("tariff", "site_info"), id="tariff_first"), + ], +) +async def test_energy_stream_site_info_and_tariff_compose( + hass: HomeAssistant, + freezer: FrozenDateTimeFactory, + mock_legacy: AsyncMock, + mock_energy_info_stream: MagicMock, + mock_energy_tariff_stream: MagicMock, + order: tuple[str, str], +) -> None: + """Slim site_info and tariff events are independently replaceable partitions.""" + tz = dt_util.get_default_time_zone() + freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz)) + + await setup_platform(hass, [Platform.CALENDAR, Platform.SELECT]) + + # The REST cold read populated both partitions. + assert hass.states.get(ENTITY_OPERATION_MODE).state == "self_consumption" + buy = hass.states.get(ENTITY_BUY) + assert buy.state == "on" + assert "0.20/kWh" in buy.attributes["message"] + + slim = deepcopy(SLIM_SITE_INFO) + slim["default_real_mode"] = "autonomous" + # A distinct OFF_PEAK price proves the streamed tariff replaced the cold read. + streamed_tariff = deepcopy(TARIFF_V2) + streamed_tariff["energy_charges"]["Summer"]["rates"]["OFF_PEAK"] = 0.99 + events = { + "site_info": lambda: mock_energy_info_stream.send(slim), + "tariff": lambda: mock_energy_tariff_stream.send(streamed_tariff), + } + + for name in order: + events[name]() + await hass.async_block_till_done() + + # Both partitions survive regardless of arrival order. + assert hass.states.get(ENTITY_OPERATION_MODE).state == "autonomous" + buy = hass.states.get(ENTITY_BUY) + assert buy.state == "on" + assert "0.99/kWh" in buy.attributes["message"] + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_energy_stream_tariff_removal_clears_calendar( + hass: HomeAssistant, + freezer: FrozenDateTimeFactory, + mock_legacy: AsyncMock, + mock_energy_tariff_stream: MagicMock, +) -> None: + """A tariff_content_v2 removal marks the tariff calendar unavailable.""" + tz = dt_util.get_default_time_zone() + freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz)) + + await setup_platform(hass, [Platform.CALENDAR]) + + assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE + + mock_energy_tariff_stream.send(None) + await hass.async_block_till_done() + + assert hass.states.get(ENTITY_BUY).state == STATE_UNAVAILABLE + + +@pytest.mark.usefixtures("entity_registry_enabled_by_default") +async def test_energy_stream_slim_site_info_drops_removed_field( + hass: HomeAssistant, + freezer: FrozenDateTimeFactory, + mock_legacy: AsyncMock, + mock_energy_info_stream: MagicMock, +) -> None: + """A replacement site_info event drops a removed field without touching tariff.""" + tz = dt_util.get_default_time_zone() + freezer.move_to(datetime(2024, 1, 1, 10, 0, 0, tzinfo=tz)) + + await setup_platform(hass, [Platform.CALENDAR, Platform.SELECT]) + + assert hass.states.get(ENTITY_OPERATION_MODE).state == "self_consumption" + assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE + + slim = deepcopy(SLIM_SITE_INFO) + slim.pop("default_real_mode") + mock_energy_info_stream.send(slim) + await hass.async_block_till_done() + + # The removed field no longer lingers in the site_info-derived entity. + assert hass.states.get(ENTITY_OPERATION_MODE).state == STATE_UNKNOWN + # The tariff partition is untouched by a site_info replacement. + assert hass.states.get(ENTITY_BUY).state != STATE_UNAVAILABLE diff --git a/tests/components/teslemetry/test_init.py b/tests/components/teslemetry/test_init.py index 87fc7bd9342e..c0f84c4dd3ab 100644 --- a/tests/components/teslemetry/test_init.py +++ b/tests/components/teslemetry/test_init.py @@ -19,14 +19,12 @@ from tesla_fleet_api.exceptions import ( TeslaFleetError, ) -from homeassistant.components.teslemetry import _get_access_token +from homeassistant.components.teslemetry import STREAM_TOPICS, _get_access_token from homeassistant.components.teslemetry.const import CLIENT_ID, DOMAIN # Coordinator constants from homeassistant.components.teslemetry.coordinator import ( ENERGY_HISTORY_INTERVAL, - ENERGY_INFO_INTERVAL, - ENERGY_LIVE_INTERVAL, INSUFFICIENT_CREDITS_RETRY_AFTER, METADATA_INTERVAL, VEHICLE_INTERVAL, @@ -165,7 +163,7 @@ async def test_energy_site_refresh_error( @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_vehicle_stream( hass: HomeAssistant, - mock_add_listener: AsyncMock, + mock_add_listener: MagicMock, snapshot: SnapshotAssertion, ) -> None: """Test vehicle stream events.""" @@ -593,7 +591,6 @@ async def test_vehicle_data_retry_exceptions( @pytest.mark.parametrize(("exception", "expected_retry_after"), RETRY_EXCEPTIONS) async def test_live_status_coordinator_retry_exceptions( hass: HomeAssistant, - freezer: FrozenDateTimeFactory, mock_live_status: AsyncMock, exception: TeslaFleetError, expected_retry_after: float, @@ -616,9 +613,8 @@ async def test_live_status_coordinator_retry_exceptions( assert entry.state is ConfigEntryState.LOADED assert call_count == 1 - # Trigger coordinator refresh - this will raise the exception - freezer.tick(ENERGY_LIVE_INTERVAL) - async_fire_time_changed(hass) + # The recovery/manual REST path still raises the exception + await entry.runtime_data.energysites[0].live_coordinator.async_refresh() await hass.async_block_till_done() # API was called exactly once for this refresh (no manual retry loop) @@ -665,7 +661,6 @@ async def test_energy_history_coordinator_retry_exceptions( async def test_live_status_auth_error( hass: HomeAssistant, - freezer: FrozenDateTimeFactory, ) -> None: """Test live status coordinator handles auth errors.""" call_count = 0 @@ -684,9 +679,8 @@ async def test_live_status_auth_error( entry = await setup_platform(hass) assert entry.state is ConfigEntryState.LOADED - # Trigger a coordinator refresh by advancing time - freezer.tick(ENERGY_LIVE_INTERVAL) - async_fire_time_changed(hass) + # The recovery/manual REST path surfaces the auth error + await entry.runtime_data.energysites[0].live_coordinator.async_refresh() await hass.async_block_till_done() # Auth error triggers reauth flow @@ -695,7 +689,6 @@ async def test_live_status_auth_error( async def test_live_status_generic_error( hass: HomeAssistant, - freezer: FrozenDateTimeFactory, ) -> None: """Test live status coordinator handles generic TeslaFleetError.""" call_count = 0 @@ -714,9 +707,8 @@ async def test_live_status_generic_error( entry = await setup_platform(hass) assert entry.state is ConfigEntryState.LOADED - # Trigger a coordinator refresh by advancing time - freezer.tick(ENERGY_LIVE_INTERVAL) - async_fire_time_changed(hass) + # The recovery/manual REST path surfaces the error + await entry.runtime_data.energysites[0].live_coordinator.async_refresh() await hass.async_block_till_done() # Entry stays loaded but coordinator will have failed @@ -911,10 +903,9 @@ async def test_vehicle_polling_stops_when_all_entities_disabled( async def test_energy_site_version_update( hass: HomeAssistant, device_registry: dr.DeviceRegistry, - mock_site_info: AsyncMock, - freezer: FrozenDateTimeFactory, + mock_add_listener: MagicMock, ) -> None: - """Test energy site sw_version updates when info coordinator refreshes.""" + """Test energy site sw_version updates from a site_info stream event.""" entry = await setup_platform(hass) assert entry.state is ConfigEntryState.LOADED @@ -925,14 +916,11 @@ async def test_energy_site_version_update( assert device is not None assert device.sw_version == "23.44.0 eb113390" - # Update mock to return new version on next poll - updated_site_info = deepcopy(SITE_INFO) - updated_site_info["response"]["version"] = "24.1.0 abc123" - mock_site_info.side_effect = lambda: updated_site_info - - # Trigger coordinator refresh - freezer.tick(ENERGY_INFO_INTERVAL) - async_fire_time_changed(hass) + # A slim site_info stream event carries the new version + updated_site_info = deepcopy(SITE_INFO["response"]) + updated_site_info.pop("tariff_content_v2", None) + updated_site_info["version"] = "24.1.0 abc123" + mock_add_listener.send({"site_id": site_id, "site_info": updated_site_info}) await hass.async_block_till_done() # Check device sw_version was updated @@ -962,7 +950,6 @@ async def test_live_status_auth_failed_forbidden( ) async def test_live_status_coordinator_refresh_error( hass: HomeAssistant, - freezer: FrozenDateTimeFactory, mock_live_status: AsyncMock, side_effect: list, ) -> None: @@ -972,8 +959,7 @@ async def test_live_status_coordinator_refresh_error( entry = await setup_platform(hass) assert entry.state is ConfigEntryState.LOADED - freezer.tick(ENERGY_LIVE_INTERVAL) - async_fire_time_changed(hass) + await entry.runtime_data.energysites[0].live_coordinator.async_refresh() await hass.async_block_till_done() assert entry.state is ConfigEntryState.LOADED @@ -1229,4 +1215,115 @@ async def test_get_access_token_rate_limited_after_setup_is_not_fatal( await _get_access_token(session) await hass.async_block_till_done() - assert not hass.config_entries.flow.async_progress() + +def test_stream_topic_allowlist() -> None: + """The stream subscribes to exactly the topics the integration consumes.""" + assert [topic.value for topic in STREAM_TOPICS] == [ + "state", + "vehicle_data", + "data", + "connectivity", + "credits", + "live_status", + "site_info", + "tariff_content_v2", + ] + + +async def test_energy_stream_no_recurring_rest_polling( + hass: HomeAssistant, + freezer: FrozenDateTimeFactory, + mock_live_status: AsyncMock, + mock_site_info: AsyncMock, +) -> None: + """The live/info REST cold reads happen once and do not recur.""" + await setup_platform(hass, [Platform.SENSOR]) + assert mock_live_status.call_count == 1 + assert mock_site_info.call_count == 1 + + # Advancing well past the old 30-second poll intervals triggers no REST reads. + freezer.tick(ENERGY_HISTORY_INTERVAL * 2) + async_fire_time_changed(hass) + await hass.async_block_till_done() + + assert mock_live_status.call_count == 1 + assert mock_site_info.call_count == 1 + + +async def test_energy_stream_unload_unsubscribes_and_closes_stream( + hass: HomeAssistant, +) -> None: + """Unload runs each listener unsubscribe and closes the shared stream.""" + live_unsub = MagicMock() + info_unsub = MagicMock() + tariff_unsub = MagicMock() + + with ( + patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_LiveStatus", + return_value=live_unsub, + ), + patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_SiteInfo", + return_value=info_unsub, + ), + patch( + "teslemetry_stream.TeslemetryStreamEnergySite.listen_TariffContentV2", + return_value=tariff_unsub, + ), + patch("teslemetry_stream.TeslemetryStream.close") as mock_close, + ): + entry = await setup_platform(hass, [Platform.SENSOR]) + assert entry.state is ConfigEntryState.LOADED + + assert await hass.config_entries.async_unload(entry.entry_id) + await hass.async_block_till_done() + + live_unsub.assert_called_once() + info_unsub.assert_called_once() + tariff_unsub.assert_called_once() + mock_close.assert_called_once() + + +async def test_energy_stream_disconnect_marks_unavailable_and_recovers( + hass: HomeAssistant, + mock_add_connection_listener: MagicMock, + mock_energy_live_stream: MagicMock, + mock_energy_info_stream: MagicMock, +) -> None: + """A dropped stream marks energy entities unavailable until documents resume.""" + await setup_platform(hass, [Platform.SENSOR, Platform.CALENDAR]) + + # Both stream-driven coordinators start available from the setup cold read. + assert hass.states.get("sensor.energy_site_solar_power").state == "1.185" + assert hass.states.get("calendar.energy_site_buy_tariff").state != STATE_UNAVAILABLE + + # A stream disconnect fails the live and info/tariff coordinators. + mock_add_connection_listener.send(False) + await hass.async_block_till_done() + + assert hass.states.get("sensor.energy_site_solar_power").state == STATE_UNAVAILABLE + assert hass.states.get("calendar.energy_site_buy_tariff").state == STATE_UNAVAILABLE + + # A streamed live_status document restores the live coordinator on reconnect. + live_status = deepcopy(LIVE_STATUS["response"]) + live_status["solar_power"] = 456 + mock_energy_live_stream.send(live_status) + await hass.async_block_till_done() + assert hass.states.get("sensor.energy_site_solar_power").state == "0.456" + + # A streamed site_info document restores the info/tariff coordinator. + slim_site_info = { + key: value + for key, value in deepcopy(SITE_INFO["response"]).items() + if key != "tariff_content_v2" + } + mock_energy_info_stream.send(slim_site_info) + await hass.async_block_till_done() + assert hass.states.get("calendar.energy_site_buy_tariff").state != STATE_UNAVAILABLE + + assert not [ + flow + for flow in hass.config_entries.flow.async_progress() + if flow["handler"] == DOMAIN + ] diff --git a/tests/components/teslemetry/test_select.py b/tests/components/teslemetry/test_select.py index b9b893652206..d140f35ffe7a 100644 --- a/tests/components/teslemetry/test_select.py +++ b/tests/components/teslemetry/test_select.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Callable from copy import deepcopy -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from freezegun.api import FrozenDateTimeFactory import pytest @@ -16,10 +16,7 @@ from homeassistant.components.select import ( DOMAIN as SELECT_DOMAIN, SERVICE_SELECT_OPTION, ) -from homeassistant.components.teslemetry.coordinator import ( - ENERGY_INFO_INTERVAL, - VEHICLE_INTERVAL, -) +from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL from homeassistant.components.teslemetry.select import HIGH, LEVEL, LOW, MEDIUM, OFF from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN, Platform from homeassistant.core import HomeAssistant @@ -534,8 +531,8 @@ async def test_export_rule_restore( ) async def test_export_rule_update_attrs_logic( hass: HomeAssistant, - freezer: FrozenDateTimeFactory, mock_site_info: AsyncMock, + mock_energy_info_stream: MagicMock, previous_data: dict, new_data: str | None, expected_state: str, @@ -549,14 +546,12 @@ async def test_export_rule_update_attrs_logic( # Set up platform await setup_platform(hass, [Platform.SELECT]) - # Change the state - test_site_info = deepcopy(SITE_INFO) - test_site_info["response"]["components"].update(new_data) - mock_site_info.side_effect = lambda: test_site_info - - # Coordinator refresh - freezer.tick(ENERGY_INFO_INTERVAL) - async_fire_time_changed(hass) + # Change the state via a streamed site_info event, driven through the + # callback the integration registered with the library. + streamed_site_info = deepcopy(SITE_INFO["response"]) + streamed_site_info.pop("tariff_content_v2", None) + streamed_site_info["components"].update(new_data) + mock_energy_info_stream.send(streamed_site_info) await hass.async_block_till_done() # Check the final state matches expected diff --git a/tests/components/teslemetry/test_sensor.py b/tests/components/teslemetry/test_sensor.py index 35a1e23275db..eb0245c16145 100644 --- a/tests/components/teslemetry/test_sensor.py +++ b/tests/components/teslemetry/test_sensor.py @@ -1,7 +1,7 @@ """Test the Teslemetry sensor platform.""" from copy import deepcopy -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock from freezegun.api import FrozenDateTimeFactory import pytest @@ -23,7 +23,13 @@ from homeassistant.helpers import entity_registry as er from homeassistant.util.unit_conversion import PressureConverter from . import assert_entities, assert_entities_alt, setup_platform -from .const import ENERGY_HISTORY_EMPTY, METADATA, PRODUCTS, VEHICLE_DATA_ALT +from .const import ( + ENERGY_HISTORY_EMPTY, + LIVE_STATUS, + METADATA, + PRODUCTS, + VEHICLE_DATA_ALT, +) from tests.common import async_fire_time_changed @@ -38,6 +44,59 @@ def _products_with_driver_assist(driver_assist: str) -> dict: return products +def _live_status(**overrides: object) -> dict: + """Return a copy of the live_status document with overrides applied.""" + data = deepcopy(LIVE_STATUS["response"]) + data.update(overrides) + return data + + +async def test_energy_live_status_stream_updates( + hass: HomeAssistant, + mock_energy_live_stream: MagicMock, +) -> None: + """A streamed live_status document drives the energy sensor states.""" + await setup_platform(hass, [Platform.SENSOR]) + + # The REST cold read populated the fixture values. + assert hass.states.get("sensor.energy_site_solar_power").state == "1.185" + assert hass.states.get("sensor.wall_connector_power").state == "0.0" + + live_status = _live_status(solar_power=456) + live_status["wall_connectors"][0]["wall_connector_power"] = 789 + mock_energy_live_stream.send(live_status) + await hass.async_block_till_done() + + assert hass.states.get("sensor.energy_site_solar_power").state == "0.456" + assert hass.states.get("sensor.wall_connector_power").state == "0.789" + + +@pytest.mark.usefixtures("mock_energy_only") +async def test_energy_only_account_streams_live_status( + hass: HomeAssistant, + entity_registry: er.EntityRegistry, + mock_stream_listen: AsyncMock, + mock_energy_live_stream: MagicMock, +) -> None: + """An energy-only account starts one stream and streams live_status to sensors.""" + entry = await setup_platform(hass, [Platform.SENSOR]) + assert entry.state is ConfigEntryState.LOADED + + # The account-wide stream is started and the live_status listener registered. + mock_stream_listen.assert_called_once() + mock_energy_live_stream.assert_called_once() + + mock_energy_live_stream.send(_live_status(solar_power=999)) + await hass.async_block_till_done() + assert hass.states.get("sensor.energy_site_solar_power").state == "0.999" + + # Credit sensors are still created for an energy-only account. + assert entry.unique_id is not None + assert entity_registry.async_get_entity_id( + Platform.SENSOR, "teslemetry", f"{entry.unique_id}_credit_quota" + ) + + @pytest.mark.usefixtures("entity_registry_enabled_by_default") async def test_sensors( hass: HomeAssistant,