Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-02-21 02:18:47 +00:00
Code Activity

Replace access token authentication with OAuth2 in Teslemetry (#158905)

Browse Source 
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Brett Adams
2025-12-30 01:40:45 +10:00
committed by GitHub
parent ec5657753f
commit 043465e42f
13 changed files with 660 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
homeassistant/components/teslemetry/__init__.py
View File

@@ -4,6 +4,7 @@ import asyncio
from collections.abc import Callable
from typing import Final
from aiohttp import ClientResponseError
from tesla_fleet_api.const import Scope
from tesla_fleet_api.exceptions import (
Forbidden,
@@ -14,16 +15,24 @@ from tesla_fleet_api.exceptions import (
from tesla_fleet_api.teslemetry import Teslemetry
from teslemetry_stream import TeslemetryStream
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.config_entry_oauth2_flow import (
OAuth2Session,
async_get_config_entry_implementation,
)
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, LOGGER
from .const import CLIENT_ID, DOMAIN, LOGGER
from .coordinator import (
TeslemetryEnergyHistoryCoordinator,
TeslemetryEnergySiteInfoCoordinator,
@@ -56,6 +65,11 @@ CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the Telemetry integration."""
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, "", name="Teslemetry"),
)
async_setup_services(hass)
return True
@@ -63,13 +77,25 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -> bool:
"""Set up Teslemetry config."""
access_token = entry.data[CONF_ACCESS_TOKEN]
session = async_get_clientsession(hass)
implementation = await async_get_config_entry_implementation(hass, entry)
oauth_session = OAuth2Session(hass, entry, implementation)
async def _get_access_token() -> str:
try:
await oauth_session.async_ensure_token_valid()
except ClientResponseError as e:
if e.status == 401:
raise ConfigEntryAuthFailed from e
raise ConfigEntryNotReady from e
token: str = oauth_session.token[CONF_ACCESS_TOKEN]
return token
# Create API connection
teslemetry = Teslemetry(
session=session,
access_token=access_token,
access_token=_get_access_token,
)
try:
calls = await asyncio.gather(
@@ -125,7 +151,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
if not stream:
stream = TeslemetryStream(
session,
access_token,
_get_access_token,
server=f"{region.lower()}.teslemetry.com",
parse_timestamp=True,
manual=True,
@@ -276,23 +302,29 @@ async def async_migrate_entry(
hass: HomeAssistant, config_entry: TeslemetryConfigEntry
) -> bool:
"""Migrate config entry."""
if config_entry.version > 1:
if config_entry.version > 2:
# This means the user has downgraded from a future version
return False
if config_entry.version == 1 and config_entry.minor_version < 2:
# Add unique_id to existing entry
teslemetry = Teslemetry(
session=async_get_clientsession(hass),
access_token=config_entry.data[CONF_ACCESS_TOKEN],
)
try:
metadata = await teslemetry.metadata()
except TeslaFleetError as e:
LOGGER.error(e.message)
return False
if config_entry.version == 1:
access_token = config_entry.data[CONF_ACCESS_TOKEN]
session = async_get_clientsession(hass)
hass.config_entries.async_update_entry(
config_entry, unique_id=metadata["uid"], version=1, minor_version=2
# Convert legacy access token to OAuth tokens using migrate endpoint
try:
data = await Teslemetry(session, access_token).migrate_to_oauth(
CLIENT_ID, access_token, hass.config.location_name
)
except ClientResponseError as e:
raise ConfigEntryAuthFailed from e
# Add auth_implementation for OAuth2 flow compatibility
data["auth_implementation"] = DOMAIN
return hass.config_entries.async_update_entry(
config_entry,
data=data,
version=2,
)
return True

30
homeassistant/components/teslemetry/application_credentials.py Normal file
View File

@@ -0,0 +1,30 @@
"""Application Credentials platform the Teslemetry integration."""
from homeassistant.components.application_credentials import (
AuthorizationServer,
ClientCredential,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from .const import AUTHORIZE_URL, TOKEN_URL
from .oauth import TeslemetryImplementation
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
"""Return authorization server."""
return AuthorizationServer(
authorize_url=AUTHORIZE_URL,
token_url=TOKEN_URL,
)
async def async_get_auth_implementation(
hass: HomeAssistant, auth_domain: str, credential: ClientCredential
) -> config_entry_oauth2_flow.AbstractOAuth2Implementation:
"""Return auth implementation."""
return TeslemetryImplementation(
hass,
auth_domain,
credential.client_id,
)

132
homeassistant/components/teslemetry/config_flow.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Mapping
import logging
from typing import Any
from aiohttp import ClientConnectionError
@@ -12,68 +13,97 @@ from tesla_fleet_api.exceptions import (
TeslaFleetError,
)
from tesla_fleet_api.teslemetry import Teslemetry
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN, LOGGER
TESLEMETRY_SCHEMA = vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str})
DESCRIPTION_PLACEHOLDERS = {
"name": "Teslemetry",
"short_url": "teslemetry.com/console",
"url": "[teslemetry.com/console](https://teslemetry.com/console)",
}
from .const import CLIENT_ID, DOMAIN, LOGGER
class TeslemetryConfigFlow(ConfigFlow, domain=DOMAIN):
"""Config Teslemetry API connection."""
class OAuth2FlowHandler(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Config flow to handle Teslemetry OAuth2 authentication."""
VERSION = 1
MINOR_VERSION = 2
DOMAIN = DOMAIN
VERSION = 2
def __init__(self) -> None:
"""Initialize config flow."""
super().__init__()
self.data: dict[str, Any] = {}
self.uid: str | None = None
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return LOGGER
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle a flow start."""
await async_import_client_credential(
self.hass,
DOMAIN,
ClientCredential(CLIENT_ID, "", name="Teslemetry"),
)
return await super().async_step_user()
async def async_oauth_create_entry(
self,
data: dict[str, Any],
) -> ConfigFlowResult:
"""Handle OAuth completion and create config entry."""
self.data = data
# Test the connection with the OAuth token
errors = await self.async_test_connection(data)
if errors:
return self.async_abort(reason="oauth_error")
await self.async_set_unique_id(self.uid)
if self.source == SOURCE_REAUTH:
self._abort_if_unique_id_mismatch(reason="reauth_account_mismatch")
return self.async_update_reload_and_abort(
self._get_reauth_entry(), data=data
)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="Teslemetry",
data=data,
)
async def async_test_connection(self, token_data: dict[str, Any]) -> dict[str, str]:
"""Test the connection with OAuth token."""
access_token = token_data["token"]["access_token"]
async def async_auth(self, user_input: Mapping[str, Any]) -> dict[str, str]:
"""Reusable Auth Helper."""
teslemetry = Teslemetry(
session=async_get_clientsession(self.hass),
access_token=user_input[CONF_ACCESS_TOKEN],
access_token=access_token,
)
try:
metadata = await teslemetry.metadata()
except InvalidToken:
return {CONF_ACCESS_TOKEN: "invalid_access_token"}
return {"base": "invalid_access_token"}
except SubscriptionRequired:
return {"base": "subscription_required"}
except ClientConnectionError:
return {"base": "cannot_connect"}
except TeslaFleetError as e:
LOGGER.error(e)
LOGGER.error("Teslemetry API error: %s", e)
return {"base": "unknown"}
await self.async_set_unique_id(metadata["uid"])
self.uid = metadata["uid"]
return {}
async def async_step_user(
self, user_input: Mapping[str, Any] | None = None
) -> ConfigFlowResult:
"""Get configuration from the user."""
errors: dict[str, str] = {}
if user_input and not (errors := await self.async_auth(user_input)):
self._abort_if_unique_id_configured()
return self.async_create_entry(
title="Teslemetry",
data=user_input,
)
return self.async_show_form(
step_id="user",
data_schema=TESLEMETRY_SCHEMA,
description_placeholders=DESCRIPTION_PLACEHOLDERS,
errors=errors,
)
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
@@ -81,21 +111,13 @@ class TeslemetryConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: Mapping[str, Any] | None = None
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle users reauth credentials."""
errors: dict[str, str] = {}
if user_input and not (errors := await self.async_auth(user_input)):
return self.async_update_reload_and_abort(
self._get_reauth_entry(),
data=user_input,
"""Confirm reauth dialog."""
if user_input is None:
return self.async_show_form(
step_id="reauth_confirm",
description_placeholders={"name": "Teslemetry"},
)
return self.async_show_form(
step_id="reauth_confirm",
description_placeholders=DESCRIPTION_PLACEHOLDERS,
data_schema=TESLEMETRY_SCHEMA,
errors=errors,
)
return await super().async_step_user()

5
homeassistant/components/teslemetry/const.py
View File

@@ -9,6 +9,11 @@ DOMAIN = "teslemetry"
LOGGER = logging.getLogger(__package__)
# OAuth
AUTHORIZE_URL = "https://teslemetry.com/connect"
TOKEN_URL = "https://api.teslemetry.com/oauth/token"
CLIENT_ID = "homeassistant"
ENERGY_HISTORY_FIELDS = [
"solar_energy_exported",
"generator_energy_exported",

1
homeassistant/components/teslemetry/manifest.json
View File

@@ -3,6 +3,7 @@
"name": "Teslemetry",
"codeowners": ["@Bre77"],
"config_flow": true,
"dependencies": ["application_credentials"],
"documentation": "https://www.home-assistant.io/integrations/teslemetry",
"integration_type": "hub",
"iot_class": "cloud_polling",

50
homeassistant/components/teslemetry/oauth.py Normal file
View File

@@ -0,0 +1,50 @@
"""Provide oauth implementations for the Teslemetry integration."""
from __future__ import annotations
from typing import Any
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_entry_oauth2_flow
from .const import AUTHORIZE_URL, TOKEN_URL
class TeslemetryImplementation(
config_entry_oauth2_flow.LocalOAuth2ImplementationWithPkce
):
"""Teslemetry OAuth2 implementation."""
def __init__(self, hass: HomeAssistant, domain: str, client_id: str) -> None:
"""Initialize OAuth2 implementation."""
super().__init__(
hass,
domain,
client_id,
AUTHORIZE_URL,
TOKEN_URL,
)
@property
def name(self) -> str:
"""Name of the implementation."""
return "Teslemetry OAuth2"
@property
def extra_authorize_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the authorize url."""
data: dict = {
"name": self.hass.config.location_name,
}
data.update(super().extra_authorize_data)
return data
@property
def extra_token_resolve_data(self) -> dict[str, Any]:
"""Extra data that needs to be appended to the token resolve request."""
data: dict = {
"name": self.hass.config.location_name,
}
data.update(super().extra_token_resolve_data)
return data

15
homeassistant/components/teslemetry/strings.json
View File

@@ -28,6 +28,10 @@
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"reauth_account_mismatch": "The reauthentication account does not match the original account",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
},
@@ -39,17 +43,8 @@
},
"step": {
"reauth_confirm": {
"data": {
"access_token": "[%key:common::config_flow::data::access_token%]"
},
"description": "The {name} integration needs to re-authenticate your account, please enter an access token from {url}",
"description": "The {name} integration needs to re-authenticate your account",
"title": "[%key:common::config_flow::title::reauth%]"
},
"user": {
"data": {
"access_token": "[%key:common::config_flow::data::access_token%]"
},
"description": "Enter an access token from {url}."
}
}
},

1
homeassistant/generated/application_credentials.py generated
View File

@@ -38,6 +38,7 @@ APPLICATION_CREDENTIALS = [
"smartthings",
"spotify",
"tesla_fleet",
"teslemetry",
"twitch",
"volvo",
"watts",

28
tests/components/teslemetry/__init__.py
View File

@@ -1,5 +1,6 @@
"""Tests for the Teslemetry integration."""
import time
from unittest.mock import patch
from syrupy.assertion import SnapshotAssertion
@@ -9,19 +10,34 @@ from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from .const import CONFIG
from tests.common import MockConfigEntry
def mock_config_entry() -> MockConfigEntry:
"""Create a mock config entry."""
return MockConfigEntry(
domain=DOMAIN,
version=2,
unique_id="abc-123",
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "test_access_token",
"refresh_token": "test_refresh_token",
"expires_at": int(time.time()) + 3600,
},
},
)
async def setup_platform(
hass: HomeAssistant, platforms: list[Platform] | None = None
hass: HomeAssistant,
platforms: list[Platform] | None = None,
) -> MockConfigEntry:
"""Set up the Teslemetry platform."""
mock_entry = MockConfigEntry(
domain=DOMAIN, data=CONFIG, minor_version=2, unique_id="abc-123"
)
mock_entry = mock_config_entry()
mock_entry.add_to_hass(hass)
if platforms is None:

11
tests/components/teslemetry/conftest.py
View File

@@ -1,4 +1,4 @@
"""Fixtures for Teslemetry."""
"""Test fixtures for Teslemetry component."""
from __future__ import annotations
@@ -22,6 +22,15 @@ from .const import (
)
@pytest.fixture
def mock_setup_entry():
"""Mock Teslemetry async_setup_entry method."""
with patch(
"homeassistant.components.teslemetry.async_setup_entry", return_value=True
) as mock_async_setup_entry:
yield mock_async_setup_entry
@pytest.fixture(autouse=True)
def mock_metadata():
"""Mock Tesla Fleet Api metadata method."""

9
tests/components/teslemetry/const.py
View File

@@ -5,7 +5,8 @@ from homeassistant.const import CONF_ACCESS_TOKEN
from tests.common import load_json_object_fixture
CONFIG = {CONF_ACCESS_TOKEN: "1234567890"}
UNIQUE_ID = "abc-123"
CONFIG_V1 = {CONF_ACCESS_TOKEN: "abc-123"}
WAKE_UP_ONLINE = {"response": {"state": TeslemetryState.ONLINE}, "error": None}
WAKE_UP_ASLEEP = {"response": {"state": TeslemetryState.ASLEEP}, "error": None}
@@ -37,7 +38,7 @@ COMMAND_ERRORS = (COMMAND_REASON, COMMAND_NOREASON, COMMAND_ERROR, COMMAND_NOERR
RESPONSE_OK = {"response": {}, "error": None}
METADATA = {
"uid": "abc-123",
"uid": UNIQUE_ID,
"region": "NA",
"scopes": [
"openid",
@@ -63,7 +64,7 @@ METADATA = {
},
}
METADATA_LEGACY = {
"uid": "abc-123",
"uid": UNIQUE_ID,
"region": "NA",
"scopes": [
"openid",
@@ -89,7 +90,7 @@ METADATA_LEGACY = {
},
}
METADATA_NOSCOPE = {
"uid": "abc-123",
"uid": UNIQUE_ID,
"region": "NA",
"scopes": ["openid", "offline_access", "vehicle_device_data"],
"vehicles": {

426
tests/components/teslemetry/test_config_flow.py
View File

@@ -1,6 +1,8 @@
"""Test the Teslemetry config flow."""
import time
from unittest.mock import AsyncMock, patch
from urllib.parse import parse_qs, urlparse
from aiohttp import ClientConnectionError
import pytest
@@ -10,222 +12,294 @@ from tesla_fleet_api.exceptions import (
TeslaFleetError,
)
from homeassistant import config_entries
from homeassistant.components.teslemetry.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.components.teslemetry.const import (
AUTHORIZE_URL,
CLIENT_ID,
DOMAIN,
TOKEN_URL,
)
from homeassistant.config_entries import SOURCE_USER, ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from .const import CONFIG, METADATA
from . import setup_platform
from .const import CONFIG_V1, UNIQUE_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
BAD_CONFIG = {CONF_ACCESS_TOKEN: "bad_access_token"}
REDIRECT = "https://example.com/auth/external/callback"
async def test_form(
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_oauth_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test we get the form."""
result1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result1["type"] is FlowResultType.FORM
assert not result1["errors"]
with patch(
"homeassistant.components.teslemetry.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
CONFIG,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert result["type"] is FlowResultType.EXTERNAL_STEP
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["data"] == CONFIG
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
assert result["url"].startswith(AUTHORIZE_URL)
parsed_url = urlparse(result["url"])
parsed_query = parse_qs(parsed_url.query)
assert parsed_query["response_type"][0] == "code"
assert parsed_query["client_id"][0] == CLIENT_ID
assert parsed_query["redirect_uri"][0] == REDIRECT
assert parsed_query["state"][0] == state
assert parsed_query["code_challenge"][0]
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
response = {
"refresh_token": "test_refresh_token",
"access_token": "test_access_token",
"type": "Bearer",
"expires_in": 60,
}
aioclient_mock.clear_requests()
aioclient_mock.post(
TOKEN_URL,
json=response,
)
# Complete OAuth
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["result"].unique_id == UNIQUE_ID
assert result["data"]["auth_implementation"] == "teslemetry"
assert result["data"]["token"]["refresh_token"] == response["refresh_token"]
assert result["data"]["token"]["access_token"] == response["access_token"]
assert result["data"]["token"]["type"] == response["type"]
assert result["data"]["token"]["expires_in"] == response["expires_in"]
assert "expires_at" in result["result"].data["token"]
@pytest.mark.parametrize(
("side_effect", "error"),
[
(InvalidToken, {CONF_ACCESS_TOKEN: "invalid_access_token"}),
(SubscriptionRequired, {"base": "subscription_required"}),
(ClientConnectionError, {"base": "cannot_connect"}),
(TeslaFleetError, {"base": "unknown"}),
],
)
async def test_form_errors(
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reauth(
hass: HomeAssistant,
side_effect: TeslaFleetError,
error: dict[str, str],
mock_metadata: AsyncMock,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test errors are handled."""
result1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
mock_metadata.side_effect = side_effect
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
CONFIG,
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == error
# Complete the flow
mock_metadata.side_effect = None
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
CONFIG,
)
assert result3["type"] is FlowResultType.CREATE_ENTRY
async def test_reauth(hass: HomeAssistant, mock_metadata: AsyncMock) -> None:
"""Test reauth flow."""
mock_entry = MockConfigEntry(
domain=DOMAIN, data=BAD_CONFIG, minor_version=2, unique_id="abc-123"
)
mock_entry.add_to_hass(hass)
result1 = await mock_entry.start_reauth_flow(hass)
assert result1["type"] is FlowResultType.FORM
assert result1["step_id"] == "reauth_confirm"
assert not result1["errors"]
with patch(
"homeassistant.components.teslemetry.async_setup_entry",
return_value=True,
) as mock_setup_entry:
result2 = await hass.config_entries.flow.async_configure(
result1["flow_id"],
CONFIG,
)
await hass.async_block_till_done()
assert len(mock_setup_entry.mock_calls) == 1
assert len(mock_metadata.mock_calls) == 1
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert mock_entry.data == CONFIG
@pytest.mark.parametrize(
("side_effect", "error"),
[
(InvalidToken, {CONF_ACCESS_TOKEN: "invalid_access_token"}),
(SubscriptionRequired, {"base": "subscription_required"}),
(ClientConnectionError, {"base": "cannot_connect"}),
(TeslaFleetError, {"base": "unknown"}),
],
)
async def test_reauth_errors(
hass: HomeAssistant,
mock_metadata: AsyncMock,
side_effect: TeslaFleetError,
error: dict[str, str],
) -> None:
"""Test reauth flows that fail."""
# Start the reauth
mock_entry = MockConfigEntry(
domain=DOMAIN, data=BAD_CONFIG, minor_version=2, unique_id="abc-123"
)
mock_entry.add_to_hass(hass)
mock_entry = await setup_platform(hass, [])
result = await mock_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
mock_metadata.side_effect = side_effect
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
BAD_CONFIG,
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
# Progress from reauth_confirm to external OAuth step
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
await hass.async_block_till_done()
client = await hass_client_no_auth()
await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == error
# Complete the flow
mock_metadata.side_effect = None
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
CONFIG,
aioclient_mock.post(
TOKEN_URL,
json={
"refresh_token": "test_refresh_token",
"access_token": "test_access_token",
"type": "Bearer",
"expires_in": 60,
},
)
assert "errors" not in result3
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reauth_successful"
assert mock_entry.data == CONFIG
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
async def test_unique_id_abort(
@pytest.mark.usefixtures("current_request_with_host")
async def test_reauth_account_mismatch(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test duplicate unique ID in config."""
result1 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=CONFIG
)
assert result1["type"] is FlowResultType.CREATE_ENTRY
# Setup a duplicate
result2 = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}, data=CONFIG
)
assert result2["type"] is FlowResultType.ABORT
async def test_migrate_from_1_1(hass: HomeAssistant, mock_metadata: AsyncMock) -> None:
"""Test config migration."""
mock_entry = MockConfigEntry(
"""Test Tesla Fleet reauthentication with different account."""
# Create an entry with a different unique_id to test account mismatch
old_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
minor_version=1,
unique_id=None,
data=CONFIG,
version=2,
unique_id="baduid",
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "old_access_token",
"refresh_token": "old_refresh_token",
"expires_at": int(time.time()) + 3600,
},
},
)
old_entry.add_to_hass(hass)
# Setup the integration properly to import client credentials
with patch(
"homeassistant.components.teslemetry.async_setup_entry", return_value=True
):
await hass.config_entries.async_setup(old_entry.entry_id)
await hass.async_block_till_done()
result = await old_entry.start_reauth_flow(hass)
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
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": "test_access_token",
"type": "Bearer",
"expires_in": 60,
},
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
with patch(
"homeassistant.components.teslemetry.async_setup_entry", return_value=True
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry.version == 1
assert entry.minor_version == 2
assert entry.unique_id == METADATA["uid"]
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_account_mismatch"
async def test_migrate_error_from_1_1(
hass: HomeAssistant, mock_metadata: AsyncMock
@pytest.mark.usefixtures("current_request_with_host")
async def test_duplicate_unique_id_abort(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
) -> None:
"""Test config migration handles errors."""
"""Test duplicate unique ID aborts flow."""
# Create existing entry
await setup_platform(hass, [])
mock_metadata.side_effect = TeslaFleetError
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
minor_version=1,
unique_id=None,
data=CONFIG,
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": REDIRECT,
},
)
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry.state is ConfigEntryState.MIGRATION_ERROR
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": "mock-access-token",
"type": "Bearer",
"expires_in": 60,
},
)
# Complete OAuth - should abort due to duplicate unique_id
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
"exception",
[
InvalidToken,
SubscriptionRequired,
ClientConnectionError,
TeslaFleetError("API error"),
],
)
async def test_oauth_error_handling(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
exception: Exception,
) -> None:
"""Test OAuth flow with various API errors."""
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": "test_refresh_token",
"access_token": "test_access_token",
"type": "Bearer",
"expires_in": 60,
},
)
with patch(
"tesla_fleet_api.teslemetry.Teslemetry.metadata",
side_effect=exception,
):
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "oauth_error"
async def test_migrate_error_from_future(
@@ -237,10 +311,10 @@ async def test_migrate_error_from_future(
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=2,
version=3,
minor_version=1,
unique_id="abc-123",
data=CONFIG,
data=CONFIG_V1,
)
mock_entry.add_to_hass(hass)

157
tests/components/teslemetry/test_init.py
View File

@@ -1,7 +1,9 @@
"""Test the Teslemetry init."""
from unittest.mock import AsyncMock, patch
import time
from unittest.mock import AsyncMock, MagicMock, patch
from aiohttp import ClientResponseError
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
@@ -11,11 +13,12 @@ from tesla_fleet_api.exceptions import (
TeslaFleetError,
)
from homeassistant.components.teslemetry.const import DOMAIN
from homeassistant.components.teslemetry.const import CLIENT_ID, DOMAIN
from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
from homeassistant.components.teslemetry.models import TeslemetryData
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
CONF_ACCESS_TOKEN,
STATE_OFF,
STATE_ON,
STATE_UNAVAILABLE,
@@ -26,7 +29,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from . import setup_platform
from .const import PRODUCTS_MODERN, VEHICLE_DATA_ALT
from .const import CONFIG_V1, PRODUCTS_MODERN, UNIQUE_ID, VEHICLE_DATA_ALT
from tests.common import MockConfigEntry
ERRORS = [
(InvalidToken, ConfigEntryState.SETUP_ERROR),
@@ -127,9 +132,11 @@ async def test_vehicle_stream(
mock_add_listener.assert_called()
state = hass.states.get("binary_sensor.test_status")
assert state is not None
assert state.state == STATE_UNKNOWN
state = hass.states.get("binary_sensor.test_user_present")
assert state is not None
assert state.state == STATE_UNAVAILABLE
mock_add_listener.send(
@@ -143,9 +150,11 @@ async def test_vehicle_stream(
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_status")
assert state is not None
assert state.state == STATE_ON
state = hass.states.get("binary_sensor.test_user_present")
assert state is not None
assert state.state == STATE_ON
mock_add_listener.send(
@@ -158,6 +167,7 @@ async def test_vehicle_stream(
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_status")
assert state is not None
assert state.state == STATE_OFF
@@ -279,3 +289,144 @@ async def test_device_retention_during_reload(
# Since the products data didn't change, we should have the same devices
assert post_count == pre_count
assert post_identifiers == original_identifiers
async def test_migrate_from_version_1_success(hass: HomeAssistant) -> None:
"""Test successful config migration from version 1."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
unique_id=UNIQUE_ID,
data=CONFIG_V1,
)
# Mock the migrate token endpoint response
with patch(
"homeassistant.components.teslemetry.Teslemetry.migrate_to_oauth",
new_callable=AsyncMock,
) as mock_migrate:
mock_migrate.return_value = {
"token": {
"access_token": "migrated_token",
"token_type": "Bearer",
"refresh_token": "migrated_refresh_token",
"expires_in": 3600,
"expires_at": time.time() + 3600,
}
}
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
mock_migrate.assert_called_once_with(
CLIENT_ID, CONFIG_V1[CONF_ACCESS_TOKEN], hass.config.location_name
)
assert mock_entry is not None
assert mock_entry.version == 2
# Verify data was converted to OAuth format
assert "token" in mock_entry.data
assert mock_entry.data["token"]["access_token"] == "migrated_token"
assert mock_entry.data["token"]["refresh_token"] == "migrated_refresh_token"
# Verify auth_implementation was added for OAuth2 flow compatibility
assert mock_entry.data["auth_implementation"] == DOMAIN
assert mock_entry.state is ConfigEntryState.LOADED
async def test_migrate_from_version_1_token_endpoint_error(hass: HomeAssistant) -> None:
"""Test config migration handles token endpoint errors."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=1,
unique_id=UNIQUE_ID,
data=CONFIG_V1,
)
# Mock the migrate token endpoint to raise an HTTP error
with patch(
"homeassistant.components.teslemetry.Teslemetry.migrate_to_oauth",
new_callable=AsyncMock,
) as mock_migrate:
mock_migrate.side_effect = ClientResponseError(
request_info=MagicMock(), history=(), status=400
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
mock_migrate.assert_called_once_with(
CLIENT_ID, CONFIG_V1[CONF_ACCESS_TOKEN], hass.config.location_name
)
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry is not None
assert entry.state is ConfigEntryState.MIGRATION_ERROR
assert entry.version == 1 # Version should remain unchanged on migration failure
async def test_migrate_version_2_no_migration_needed(hass: HomeAssistant) -> None:
"""Test that version 2 entries don't need migration."""
oauth_config = {
"auth_implementation": DOMAIN,
"token": {
"access_token": "existing_oauth_token",
"token_type": "Bearer",
"refresh_token": "existing_refresh_token",
"expires_in": 3600,
"expires_at": 1234567890,
},
}
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=2, # Already current version
unique_id=UNIQUE_ID,
data=oauth_config,
)
# Should not call the migrate endpoint since already version 2
with patch(
"homeassistant.components.teslemetry.Teslemetry.migrate_to_oauth",
new_callable=AsyncMock,
) as mock_migrate:
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
# Migration should not be called
mock_migrate.assert_not_called()
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry is not None
assert entry.version == 2
# Verify data was not modified
assert entry.data == oauth_config
assert entry.state is ConfigEntryState.LOADED
async def test_migrate_from_future_version_fails(hass: HomeAssistant) -> None:
"""Test migration fails for future versions."""
mock_entry = MockConfigEntry(
domain=DOMAIN,
version=3, # Future version
unique_id=UNIQUE_ID,
data={
"token": {
"access_token": "future_token",
"token_type": "Bearer",
"refresh_token": "future_refresh_token",
"expires_in": 3600,
}
},
)
mock_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_entry.entry_id)
await hass.async_block_till_done()
entry = hass.config_entries.async_get_entry(mock_entry.entry_id)
assert entry is not None
assert entry.state is ConfigEntryState.MIGRATION_ERROR
assert entry.version == 3 # Version should remain unchanged