From a3add179a0aff2acda5332eb3fc2e69a91a3c4ed Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 25 Mar 2026 16:46:47 +1000 Subject: [PATCH] Fix Tesla Fleet partner_login to not require vehicle scope. (#166435) --- .../components/tesla_fleet/config_flow.py | 43 ++++++- .../tesla_fleet/test_config_flow.py | 116 ++++++++++++++++++ 2 files changed, 154 insertions(+), 5 deletions(-) diff --git a/homeassistant/components/tesla_fleet/config_flow.py b/homeassistant/components/tesla_fleet/config_flow.py index 0f93a7f3328..14c197bc7c5 100644 --- a/homeassistant/components/tesla_fleet/config_flow.py +++ b/homeassistant/components/tesla_fleet/config_flow.py @@ -9,8 +9,14 @@ from typing import Any, cast import jwt from tesla_fleet_api import TeslaFleetApi -from tesla_fleet_api.const import SERVERS -from tesla_fleet_api.exceptions import PreconditionFailed, TeslaFleetError +from tesla_fleet_api.const import SERVERS, Scope +from tesla_fleet_api.exceptions import ( + InvalidToken, + LoginRequired, + OAuthExpired, + PreconditionFailed, + TeslaFleetError, +) import voluptuous as vol from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult @@ -69,6 +75,7 @@ class OAuth2FlowHandler( # OAuth done, setup Partner API connections for all regions implementation = cast(TeslaUserImplementation, self.flow_impl) session = async_get_clientsession(self.hass) + failed_regions: list[str] = [] for region, server_url in SERVERS.items(): if region == "cn": @@ -84,11 +91,37 @@ class OAuth2FlowHandler( vehicle_scope=False, ) await api.get_private_key(self.hass.config.path("tesla_fleet.key")) - await api.partner_login( - implementation.client_id, implementation.client_secret - ) + try: + await api.partner_login( + implementation.client_id, + implementation.client_secret, + [Scope.OPENID], + ) + except (InvalidToken, OAuthExpired, LoginRequired) as err: + LOGGER.warning( + "Partner login failed for %s due to an authentication error: %s", + server_url, + err, + ) + return self.async_abort(reason="oauth_error") + except TeslaFleetError as err: + LOGGER.warning("Partner login failed for %s: %s", server_url, err) + failed_regions.append(server_url) + continue self.apis.append(api) + if not self.apis: + LOGGER.warning( + "Partner login failed for all regions: %s", ", ".join(failed_regions) + ) + return self.async_abort(reason="oauth_error") + + if failed_regions: + LOGGER.warning( + "Partner login succeeded on some regions but failed on: %s", + ", ".join(failed_regions), + ) + return await self.async_step_domain_input() async def async_step_domain_input( diff --git a/tests/components/tesla_fleet/test_config_flow.py b/tests/components/tesla_fleet/test_config_flow.py index 1da4911f733..c54c3f6c655 100644 --- a/tests/components/tesla_fleet/test_config_flow.py +++ b/tests/components/tesla_fleet/test_config_flow.py @@ -6,6 +6,7 @@ from urllib.parse import parse_qs, urlparse import pytest from tesla_fleet_api.exceptions import ( InvalidResponse, + LoginRequired, PreconditionFailed, TeslaFleetError, ) @@ -87,6 +88,121 @@ def mock_private_key(): return private_key +@pytest.mark.usefixtures("current_request_with_host") +async def test_partner_login_auth_error( + hass: HomeAssistant, + hass_client_no_auth: ClientSessionGenerator, + aioclient_mock: AiohttpClientMocker, + access_token: str, + mock_private_key, +) -> None: + """Test partner login auth errors abort the flow cleanly.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + state = config_entry_oauth2_flow._encode_jwt( + hass, + { + "flow_id": result["flow_id"], + "redirect_uri": REDIRECT, + }, + ) + + client = await hass_client_no_auth() + await client.get(f"/auth/external/callback?code=abcd&state={state}") + + aioclient_mock.post( + TOKEN_URL, + json={ + "refresh_token": "mock-refresh-token", + "access_token": access_token, + "type": "Bearer", + "expires_in": 60, + }, + ) + + with patch( + "homeassistant.components.tesla_fleet.config_flow.TeslaFleetApi" + ) as mock_api_class: + mock_api = AsyncMock() + mock_api.private_key = mock_private_key + mock_api.get_private_key = AsyncMock() + mock_api.partner_login = AsyncMock(side_effect=LoginRequired) + mock_api_class.return_value = mock_api + + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + + assert result["type"] is FlowResultType.ABORT + assert result["reason"] == "oauth_error" + + +@pytest.mark.usefixtures("current_request_with_host") +async def test_partner_login_partial_failure( + hass: HomeAssistant, + hass_client_no_auth: ClientSessionGenerator, + aioclient_mock: AiohttpClientMocker, + access_token: str, + mock_private_key, +) -> None: + """Test partner login succeeds when one region fails.""" + result = await hass.config_entries.flow.async_init( + DOMAIN, context={"source": SOURCE_USER} + ) + + state = config_entry_oauth2_flow._encode_jwt( + hass, + { + "flow_id": result["flow_id"], + "redirect_uri": REDIRECT, + }, + ) + + client = await hass_client_no_auth() + await client.get(f"/auth/external/callback?code=abcd&state={state}") + + aioclient_mock.post( + TOKEN_URL, + json={ + "refresh_token": "mock-refresh-token", + "access_token": access_token, + "type": "Bearer", + "expires_in": 60, + }, + ) + + public_key = "0404112233445566778899aabbccddeeff112233445566778899aabbccddeeff112233445566778899aabbccddeeff112233445566778899aabbccddeeff1122" + + mock_api_na = AsyncMock() + mock_api_na.private_key = mock_private_key + mock_api_na.get_private_key = AsyncMock() + mock_api_na.partner_login = AsyncMock() + mock_api_na.public_uncompressed_point = public_key + mock_api_na.partner.register.return_value = {"response": {"public_key": public_key}} + + mock_api_eu = AsyncMock() + mock_api_eu.private_key = mock_private_key + mock_api_eu.get_private_key = AsyncMock() + mock_api_eu.partner_login = AsyncMock( + side_effect=TeslaFleetError("EU partner login failed") + ) + + with patch( + "homeassistant.components.tesla_fleet.config_flow.TeslaFleetApi", + side_effect=[mock_api_na, mock_api_eu], + ): + result = await hass.config_entries.flow.async_configure(result["flow_id"]) + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "domain_input" + + result = await hass.config_entries.flow.async_configure( + result["flow_id"], {CONF_DOMAIN: "example.com"} + ) + + assert result["type"] is FlowResultType.FORM + assert result["step_id"] == "registration_complete" + + @pytest.mark.usefixtures("current_request_with_host") async def test_full_flow_with_domain_registration( hass: HomeAssistant,