diff --git a/homeassistant/components/teslemetry/config_flow.py b/homeassistant/components/teslemetry/config_flow.py index 602b85a05d14..f7f9ad2829f7 100644 --- a/homeassistant/components/teslemetry/config_flow.py +++ b/homeassistant/components/teslemetry/config_flow.py @@ -25,6 +25,7 @@ from tesla_fleet_api.exceptions import ( TeslaFleetError, WhitelistOperationAttemptingToAddExistingKey, ) +from tesla_fleet_api.tesla import EnergySiteRouter from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth from tesla_fleet_api.teslemetry import Teslemetry from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, TeslemetryEnergySite @@ -75,6 +76,7 @@ from .helpers import ( async_verify_local_gateway, cloud_energy_site, ) +from .models import TeslemetryEnergyData class PowerwallLookupError(Exception): @@ -428,7 +430,7 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow): def __init__(self) -> None: """Initialize the energy site subentry flow.""" - self._energy_site: TeslemetryEnergySite | None = None + self._energy_site: TeslemetryEnergySite | EnergySiteRouter | None = None self._key_pem: bytes | None = None self._public_key_der: bytes = b"" self._public_key_b64: str = "" @@ -470,9 +472,7 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow): energy_data = available[user_input[CONF_SITE_ID]] self._site_id = energy_data.id self._site_name = energy_data.device.get("name") or "Energy Site" - if abort := await self._prepare_energy_site( - cloud_energy_site(energy_data.api) - ): + if abort := await self._prepare_energy_site(energy_data): return abort return await self._async_begin_pairing() @@ -509,14 +509,15 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow): ) if energy_data is None: return self.async_abort(reason="cannot_connect") - if abort := await self._prepare_energy_site(cloud_energy_site(energy_data.api)): + if abort := await self._prepare_energy_site(energy_data): return abort return await self._async_begin_pairing() async def _prepare_energy_site( - self, energy_site: TeslemetryEnergySite + self, energy_data: TeslemetryEnergyData ) -> SubentryFlowResult | None: """Discover the gateway address and load the integration's RSA key.""" + energy_site = cast(TeslemetryEnergySite | EnergySiteRouter, energy_data.api) self._energy_site = energy_site try: @@ -564,10 +565,12 @@ class EnergySiteSubentryFlowHandler(ConfigSubentryFlow): if TYPE_CHECKING: assert self._energy_site is not None + # Registration must reach Tesla, so it never routes to the local gateway. + cloud_site = cloud_energy_site(self._energy_site) try: # Not revoked on removal: other consumers may share this key. - LOGGER.info("Powerwall key setup: id=%s", self._energy_site.energy_site_id) - await self._energy_site.add_authorized_client( + LOGGER.info("Powerwall key setup: id=%s", cloud_site.energy_site_id) + await cloud_site.add_authorized_client( self._public_key_der, description="Home Assistant", key_type=AuthorizedClientKeyType.RSA, diff --git a/tests/components/teslemetry/test_config_flow.py b/tests/components/teslemetry/test_config_flow.py index 0db8020c68a1..8924a8f8ba4b 100644 --- a/tests/components/teslemetry/test_config_flow.py +++ b/tests/components/teslemetry/test_config_flow.py @@ -31,7 +31,7 @@ from tesla_fleet_api.exceptions import ( TeslaFleetError, WhitelistOperationAttemptingToAddExistingKey, ) -from tesla_fleet_api.tesla import EnergySiteRouter, VehicleRouter +from tesla_fleet_api.tesla import VehicleRouter from tesla_fleet_api.tesla.bluetooth import TeslaBluetooth from tesla_fleet_api.teslemetry.energysite import AuthorizedClient, AuthorizedClients @@ -1519,6 +1519,16 @@ def mock_gateway_discovery() -> Generator[AsyncMock]: yield mock_find +@pytest.fixture(autouse=True) +def mock_local_authorized_clients() -> Generator[AsyncMock]: + """Default a paired gateway's local authorized-clients read to unreachable.""" + with patch( + "aiopowerwall.client.PowerwallClient.list_authorized_clients", + new=AsyncMock(side_effect=PowerwallConnectionError), + ) as mock_list: + yield mock_list + + @pytest.fixture def mock_rsa_key() -> Generator[None]: """Mock RSA key generation/loading, avoiding real crypto and disk I/O.""" @@ -1590,6 +1600,22 @@ def _empty_clients() -> AuthorizedClients: return AuthorizedClients(clients=[], raw=None) +def _local_client(public_key: str, state: str) -> dict[str, Any]: + """Return one client entry as the gateway's local read reports it.""" + return { + "public_key": public_key, + "state": state, + "type": "CUSTOMER_MOBILE_APP", + "description": "Home Assistant", + "key_type": "RSA", + "roles": ["CUSTOMER"], + "verification": "PRESENCE_PROOF", + "added_time": None, + "identifier": None, + "authorized_by_public_key": None, + } + + async def _start_add_flow_select_site( hass: HomeAssistant, entry: MockConfigEntry ) -> SubentryFlowResult: @@ -2262,7 +2288,7 @@ async def test_pair_step_timeout_retry_reopens_window_and_succeeds( new=AsyncMock(), ) as mock_add, patch( - "homeassistant.components.teslemetry.config_flow.PowerwallClient", + "homeassistant.components.teslemetry.helpers.PowerwallClient", return_value=client, ), patch.object(hass.config_entries, "async_schedule_reload"), @@ -2349,6 +2375,26 @@ async def _setup_paired_account(hass: HomeAssistant) -> MockConfigEntry: return entry +async def _setup_cloud_only_account(hass: HomeAssistant) -> MockConfigEntry: + """Set up a paired account whose local control failed to initialize at setup. + + The RSA key the local gateway client needs cannot be loaded, so the integration + falls back to the bare cloud api for the site. + """ + entry = _entry_with_powerwall() + entry.add_to_hass(hass) + with ( + patch( + "homeassistant.components.teslemetry._async_get_rsa_key_pem", + side_effect=OSError, + ), + patch("homeassistant.components.teslemetry.PLATFORMS", []), + ): + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + return entry + + @pytest.mark.usefixtures("mock_rsa_key") async def test_reconfigure_updates_credentials_and_schedules_reload( hass: HomeAssistant, @@ -2510,42 +2556,122 @@ async def test_reconfigure_aborts_when_rsa_key_load_fails(hass: HomeAssistant) - @pytest.mark.usefixtures("mock_rsa_key") -async def test_reconfigure_pairs_via_cloud_secondary_not_local_primary( +async def test_reconfigure_aborts_when_local_and_cloud_lookups_fail( hass: HomeAssistant, + mock_local_authorized_clients: AsyncMock, ) -> None: - """Reconfiguring a paired site pairs through the cloud site, not the local gateway. - - A paired site's api is a local-first router, so the pairing lookup must be - unwrapped to the cloud secondary; routing it would hit the local Powerwall. - """ + """Reconfigure aborts when both the local and the cloud lookup fail.""" entry = await _setup_paired_account(hass) subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id - energy_data = entry.runtime_data.energysites[0] - assert isinstance(energy_data.api, EnergySiteRouter) - cloud_lookup = AsyncMock( - return_value=_own_key_clients(AuthorizedClientState.VERIFIED) - ) - # find_authorized_clients is a cloud-only method; adding it to the local - # backend makes the router route to it local-first, so an accidentally - # routed lookup would land on the primary and this test would catch it. - local_lookup = AsyncMock( - return_value=_own_key_clients(AuthorizedClientState.VERIFIED) - ) + cloud_lookup = AsyncMock(side_effect=TeslaFleetError) + add_client = AsyncMock(return_value={}) with ( - patch.object( - energy_data.api.secondary, "find_authorized_clients", cloud_lookup + patch( + "tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients", + new=cloud_lookup, ), - patch.object( - energy_data.api.primary, - "find_authorized_clients", - local_lookup, - create=True, + patch( + "tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.add_authorized_client", + new=add_client, ), ): result = await entry.start_subentry_reconfigure_flow(hass, subentry_id) - # Reaching credentials means the cloud lookup reported the key verified. + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "cannot_connect" + mock_local_authorized_clients.assert_awaited_once() + cloud_lookup.assert_awaited_once() + # A failed lookup must not be mistaken for an unregistered key. + add_client.assert_not_awaited() + + +@pytest.mark.usefixtures("mock_rsa_key") +async def test_reconfigure_reads_authorized_clients_locally( + hass: HomeAssistant, + mock_local_authorized_clients: AsyncMock, +) -> None: + """A paired site's verified key is confirmed on the gateway, not the cloud. + + The router reads local-first, so a reachable gateway answers the lookup and + the flow reaches the credentials step without asking the cloud. + """ + entry = await _setup_paired_account(hass) + subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id + mock_local_authorized_clients.side_effect = None + mock_local_authorized_clients.return_value = { + "clients": [ + _local_client("some-other-key", "VERIFIED"), + _local_client(PUBLIC_KEY_B64, "VERIFIED"), + ], + "enable_line_switch_off": False, + } + + cloud_lookup = AsyncMock(return_value=_empty_clients()) + with patch( + "tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients", + new=cloud_lookup, + ): + result = await entry.start_subentry_reconfigure_flow(hass, subentry_id) + + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "credentials" - cloud_lookup.assert_awaited() - local_lookup.assert_not_awaited() + mock_local_authorized_clients.assert_awaited_once() + cloud_lookup.assert_not_awaited() + + +@pytest.mark.usefixtures("mock_rsa_key") +async def test_reconfigure_cloud_only_site_skips_local_lookup( + hass: HomeAssistant, + mock_local_authorized_clients: AsyncMock, +) -> None: + """A site whose local control failed at setup is looked up only in the cloud.""" + entry = await _setup_cloud_only_account(hass) + subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id + + cloud_lookup = AsyncMock( + return_value=_own_key_clients(AuthorizedClientState.VERIFIED) + ) + with patch( + "tesla_fleet_api.teslemetry.energysite.TeslemetryEnergySite.find_authorized_clients", + new=cloud_lookup, + ): + result = await entry.start_subentry_reconfigure_flow(hass, subentry_id) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "credentials" + cloud_lookup.assert_awaited_once() + mock_local_authorized_clients.assert_not_awaited() + + +@pytest.mark.usefixtures("mock_rsa_key") +async def test_reconfigure_registers_key_via_cloud_not_local_gateway( + hass: HomeAssistant, + mock_local_authorized_clients: AsyncMock, +) -> None: + """Registering an unknown key goes to the cloud site, not the local gateway. + + A paired site's api is a local-first router, so registration must be unwrapped + to the cloud secondary. The local backend is given a working + add_authorized_client here so that a routed registration would land on the + Powerwall and fail this test. + """ + entry = await _setup_paired_account(hass) + subentry_id = entry.get_subentries_of_type(SUBENTRY_TYPE_ENERGY_SITE)[0].subentry_id + energy_data = entry.runtime_data.energysites[0] + # An empty local read means our key is not registered yet. + mock_local_authorized_clients.side_effect = None + mock_local_authorized_clients.return_value = {"clients": []} + + cloud_add = AsyncMock(return_value={}) + local_add = AsyncMock(return_value={}) + with ( + patch.object(energy_data.api.secondary, "add_authorized_client", cloud_add), + patch.object(energy_data.api.primary, "add_authorized_client", local_add), + ): + result = await entry.start_subentry_reconfigure_flow(hass, subentry_id) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "pair" + cloud_add.assert_awaited_once() + local_add.assert_not_awaited()