starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-24 15:31:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Continue central error handling for OAuth (#180588)

Browse Source
This commit is contained in:
Josef Zweck
2026-09-01 22:35:32 +02:00
committed by GitHub
parent f945b3d2ff
commit 80359bfb46
6 changed files with 519 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/cloud/account_link.py
+6
View File
@@ -113,6 +113,12 @@ class CloudOAuth2Implementation(config_entry_oauth2_flow.AbstractOAuth2Implement
"""Domain that is providing the implementation."""
return DOMAIN
@property
@override
def service_domain(self) -> str:
"""Domain of the service the tokens are for."""
return self.service
@override
async def async_generate_authorize_url(self, flow_id: str) -> str:
"""Generate a url for the user to authorize."""
homeassistant/exceptions.py
+34 -5
View File
@@ -3,7 +3,7 @@
from collections.abc import Callable, Generator, Sequence
from typing import TYPE_CHECKING, Any, override
from aiohttp import ClientResponse, ClientResponseError, RequestInfo
from aiohttp import ClientError, ClientResponse, ClientResponseError, RequestInfo
from multidict import MultiMapping
from .util.event_type import EventType
@@ -253,8 +253,20 @@ class ConfigEntryAuthFailed(IntegrationError):
"""Error to indicate that config entry could not authenticate."""
class OAuth2TokenRequestError(ClientResponseError, HomeAssistantError):
"""Error to indicate that the OAuth 2.0 flow could not refresh token."""
class OAuth2TokenRequestBaseError(ConfigEntryNotReady):
"""Base class for the errors a failed OAuth 2.0 token request raises.
Catch this to handle every token request failure; the subclasses differ in
whether a status was received and what should happen to the config entry.
"""
class OAuth2TokenRequestError(ClientResponseError, OAuth2TokenRequestBaseError):
"""Error to indicate that the OAuth 2.0 flow could not refresh token.
Inherits ConfigEntryNotReady so setup retries without the integration having to
map it. Catch it explicitly to handle it differently.
"""
def __init__(
self,
@@ -275,7 +287,7 @@ class OAuth2TokenRequestError(ClientResponseError, HomeAssistantError):
message=message,
headers=headers,
)
HomeAssistantError.__init__(self)
OAuth2TokenRequestBaseError.__init__(self)
self.domain = domain
self.translation_domain = "homeassistant"
self.translation_key = "oauth2_helper_refresh_failed"
@@ -283,7 +295,24 @@ class OAuth2TokenRequestError(ClientResponseError, HomeAssistantError):
self.generate_message = True
class OAuth2TokenRequestTransientError(OAuth2TokenRequestError, ConfigEntryNotReady):
class OAuth2TokenRequestConnectionError(ClientError, OAuth2TokenRequestBaseError):
"""Recoverable error to indicate the token request yielded no usable token.
Covers a request that never got a response and one whose response could not
be used, neither of which has a status to tell the causes apart.
"""
def __init__(self, *, domain: str) -> None:
"""Initialize OAuth2TokenRequestConnectionError."""
OAuth2TokenRequestBaseError.__init__(self)
self.domain = domain
self.translation_domain = "homeassistant"
self.translation_key = "oauth2_helper_refresh_transient"
self.translation_placeholders = {"domain": domain}
self.generate_message = True
class OAuth2TokenRequestTransientError(OAuth2TokenRequestError):
"""Recoverable error to indicate flow could not refresh token.
Inherits ConfigEntryNotReady so setup retries without the integration having to
homeassistant/helpers/config_entry_oauth2_flow.py
+65 -34
View File
@@ -17,7 +17,7 @@ import json
import logging
import secrets
import time
from typing import Any, cast, override
from typing import Any, NoReturn, cast, override
from aiohttp import ClientError, ClientResponseError, client, hdrs, web
from habluetooth import BluetoothServiceInfoBleak
@@ -30,6 +30,7 @@ from homeassistant import config_entries
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant, callback
from homeassistant.exceptions import (
ImplementationUnavailableError,
OAuth2TokenRequestConnectionError,
OAuth2TokenRequestError,
OAuth2TokenRequestReauthError,
OAuth2TokenRequestTransientError,
@@ -105,6 +106,28 @@ _SHARED_ABORT_REASONS = frozenset(
)
def _raise_mapped_token_error(err: ClientError, domain: str) -> NoReturn:
"""Re-raise a failed token request as the matching OAuth2 token error."""
if not isinstance(err, ClientResponseError):
# Nothing was received, so there is no status to tell the causes apart.
_LOGGER.debug("Token request for %s got no response: %s", domain, err)
raise OAuth2TokenRequestConnectionError(domain=domain) from err
kwargs: dict[str, Any] = {
"request_info": err.request_info,
"history": err.history,
"status": err.status,
"message": err.message,
"headers": err.headers,
"domain": domain,
}
if err.status == HTTPStatus.TOO_MANY_REQUESTS or 500 <= err.status <= 599:
raise OAuth2TokenRequestTransientError(**kwargs) from err
if 400 <= err.status <= 499:
raise OAuth2TokenRequestReauthError(**kwargs) from err
raise OAuth2TokenRequestError(**kwargs) from err
@callback
def async_get_redirect_uri(hass: HomeAssistant) -> str:
"""Return the redirect uri."""
@@ -163,11 +186,30 @@ class AbstractOAuth2Implementation(ABC):
config entry data.
"""
@property
def service_domain(self) -> str:
"""Domain of the service the tokens are for.
Defaults to the implementation itself, but an implementation that obtains
tokens on behalf of other integrations has to name the one it serves.
"""
return self.domain
async def async_refresh_token(self, token: dict) -> dict:
"""Refresh a token and update expires info."""
new_token = await self._async_refresh_token(token)
try:
new_token = await self._async_refresh_token(token)
except OAuth2TokenRequestError, OAuth2TokenRequestConnectionError:
raise
except ClientError as err:
# Implementations that issue their own token request may not map their
# failures, so callers would see a raw aiohttp error instead.
_raise_mapped_token_error(err, self.service_domain)
# Force int for non-compliant oauth2 providers
new_token["expires_in"] = int(new_token["expires_in"])
try:
new_token["expires_in"] = int(new_token["expires_in"])
except (KeyError, TypeError, ValueError) as err:
raise OAuth2TokenRequestConnectionError(domain=self.service_domain) from err
new_token["expires_at"] = time.time() + new_token["expires_in"]
return new_token
@@ -268,6 +310,11 @@ class LocalOAuth2Implementation(AbstractOAuth2Implementation):
}
)
# Merging a response without one would keep the stale access token while
# extending its expiry, so the session would never recover.
if not new_token.get("access_token"):
raise OAuth2TokenRequestConnectionError(domain=self.service_domain)
return {**token, **new_token}
async def _token_request(self, data: dict) -> dict:
@@ -306,38 +353,13 @@ class LocalOAuth2Implementation(AbstractOAuth2Implementation):
detail,
)
resp.raise_for_status()
return cast(dict, await resp.json())
except ClientResponseError as err:
if err.status == HTTPStatus.TOO_MANY_REQUESTS or 500 <= err.status <= 599:
# Recoverable error
raise OAuth2TokenRequestTransientError(
request_info=err.request_info,
history=err.history,
status=err.status,
message=err.message,
headers=err.headers,
domain=self._domain,
) from err
if 400 <= err.status <= 499:
# Non-recoverable error
raise OAuth2TokenRequestReauthError(
request_info=err.request_info,
history=err.history,
status=err.status,
message=err.message,
headers=err.headers,
domain=self._domain,
) from err
raise OAuth2TokenRequestError(
request_info=err.request_info,
history=err.history,
status=err.status,
message=err.message,
headers=err.headers,
domain=self._domain,
) from err
return cast(dict, await resp.json())
_raise_mapped_token_error(err, self.service_domain)
except ClientError as err:
# Bare TimeoutError is left alone so an enclosing asyncio.timeout still
# aborts with oauth_timeout; aiohttp's own timeouts are ClientErrors.
_raise_mapped_token_error(err, self.service_domain)
class LocalOAuth2ImplementationWithPkce(LocalOAuth2Implementation):
@@ -844,6 +866,15 @@ class OAuth2Session:
self.config_entry.async_start_reauth_if_available(self.hass)
raise
# Checked before storing, so reads can trust what is on the entry.
if any(
new_token.get(field) in (None, "")
for field in ("access_token", "expires_at")
):
raise OAuth2TokenRequestConnectionError(
domain=self.implementation.service_domain
)
self.hass.config_entries.async_update_entry(
self.config_entry, data={**self.config_entry.data, "token": new_token}
)
script/hassfest/quality_scale_validation/test_before_setup.py
+23 -11
View File
@@ -14,6 +14,13 @@ _VALID_EXCEPTIONS = {
"ConfigEntryError",
}
# Helpers that raise one of the above on the caller's behalf, so an integration
# awaiting them satisfies the rule without repeating the mapping itself.
_VALID_AWAITED_CALLS = {
"async_config_entry_first_refresh",
"async_ensure_token_valid",
}
def _get_exception_name(expression: ast.expr) -> str:
"""Get the name of the exception being raised."""
@@ -58,17 +65,20 @@ def _raises_exception(integration: Integration) -> bool:
return False
def _calls_first_refresh(async_setup_entry_function: ast.AsyncFunctionDef) -> bool:
"""Check that a async_config_entry_first_refresh within `async_setup_entry`."""
for node in ast.walk(async_setup_entry_function):
if (
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "async_config_entry_first_refresh"
):
return True
def _awaits_raising_helper(async_setup_entry_function: ast.AsyncFunctionDef) -> bool:
"""Check that `async_setup_entry` awaits a helper that raises on its behalf.
return False
The call only has to sit somewhere inside an await, so gathering several of
them still counts, while an unawaited call does not.
"""
return any(
isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr in _VALID_AWAITED_CALLS
for await_node in ast.walk(async_setup_entry_function)
if isinstance(await_node, ast.Await)
for node in ast.walk(await_node)
)
def _get_setup_entry_function(module: ast.Module) -> ast.AsyncFunctionDef | None:
@@ -90,6 +100,8 @@ def validate(
if not (async_setup_entry := _get_setup_entry_function(init)):
return [f"Could not find `async_setup_entry` in {init_file}"]
if not (_calls_first_refresh(async_setup_entry) or _raises_exception(integration)):
if not (
_awaits_raising_helper(async_setup_entry) or _raises_exception(integration)
):
return [f"Integration does not raise one of {_VALID_EXCEPTIONS}"]
return None
tests/components/cloud/test_account_link.py
+23 -1
View File
@@ -6,7 +6,7 @@ import logging
from time import time
from unittest.mock import AsyncMock, Mock, patch
from aiohttp import ClientResponseError, RequestInfo
from aiohttp import ClientError, ClientResponseError, RequestInfo
import pytest
from yarl import URL
@@ -16,6 +16,7 @@ from homeassistant.components.cloud.const import DATA_CLOUD
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import (
OAuth2TokenRequestConnectionError,
OAuth2TokenRequestError,
OAuth2TokenRequestReauthError,
OAuth2TokenRequestTransientError,
@@ -304,3 +305,24 @@ async def test_refresh_token_error(
assert exc_info.value.status == status
assert exc_info.value.domain == "test"
async def test_refresh_token_connection_error(hass: HomeAssistant) -> None:
"""Test a failure without a response reports the service, not the cloud domain."""
hass.data[DATA_CLOUD] = None
impl = account_link.CloudOAuth2Implementation(hass, "test")
with (
patch(
"hass_nabucasa.account_link.async_fetch_access_token",
side_effect=ClientError("Cannot connect"),
),
pytest.raises(OAuth2TokenRequestConnectionError) as exc_info,
):
await impl.async_refresh_token(
{"refresh_token": "mock-refresh", "access_token": "mock-access"}
)
assert impl.domain == "cloud"
assert exc_info.value.domain == "test"
assert exc_info.value.translation_placeholders == {"domain": "test"}
tests/helpers/test_config_entry_oauth2_flow.py
+368 -7
View File
@@ -7,15 +7,24 @@ import time
from typing import Any
from unittest.mock import AsyncMock, patch
from aiohttp import ClientError
from multidict import CIMultiDict
from aiohttp import (
ClientError,
ClientPayloadError,
ClientResponseError,
ContentTypeError,
RequestInfo,
ServerTimeoutError,
)
from multidict import CIMultiDict, CIMultiDictProxy
import pytest
from yarl import URL
from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, HomeAssistant
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryNotReady,
OAuth2TokenRequestConnectionError,
OAuth2TokenRequestError,
OAuth2TokenRequestReauthError,
OAuth2TokenRequestTransientError,
@@ -24,7 +33,7 @@ from homeassistant.helpers import config_entry_oauth2_flow
from homeassistant.helpers.network import NoURLAvailableError
from tests.common import MockConfigEntry, MockModule, mock_integration, mock_platform
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.test_util.aiohttp import AiohttpClientMocker, AiohttpClientMockResponse
from tests.typing import ClientSessionGenerator
TEST_DOMAIN = "oauth2_test"
@@ -35,6 +44,8 @@ ACCESS_TOKEN_1 = "mock-access-token-1"
ACCESS_TOKEN_2 = "mock-access-token-2"
AUTHORIZE_URL = "https://example.como/auth/authorize"
TOKEN_URL = "https://example.como/auth/token"
# Far enough ahead that a token carrying it always counts as unexpired.
FUTURE_EXPIRES_AT = 2000000000
MOCK_SECRET_TOKEN_URLSAFE = (
"token-"
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
@@ -1106,6 +1117,356 @@ async def test_oauth_session_refresh_failure_exceptions(
assert f"Token request for {TEST_DOMAIN} failed" in caplog.text
@pytest.mark.parametrize(
"raised",
[
pytest.param(ClientError("Cannot connect"), id="client_error"),
pytest.param(ServerTimeoutError("Timeout"), id="timeout"),
],
)
async def test_oauth_session_refresh_connection_error_is_transient(
hass: HomeAssistant,
flow_handler: type[config_entry_oauth2_flow.AbstractOAuth2FlowHandler],
local_impl: config_entry_oauth2_flow.LocalOAuth2Implementation,
aioclient_mock: AiohttpClientMocker,
raised: Exception,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test a token request that never gets a response is mapped to a transient error."""
mock_integration(hass, MockModule(domain=TEST_DOMAIN))
flow_handler.async_register_implementation(hass, local_impl)
aioclient_mock.post(TOKEN_URL, exc=raised)
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"refresh_token": REFRESH_TOKEN,
"access_token": ACCESS_TOKEN_1,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
session = config_entry_oauth2_flow.OAuth2Session(hass, config_entry, local_impl)
with (
caplog.at_level(logging.DEBUG),
pytest.raises(OAuth2TokenRequestConnectionError) as err,
):
await session.async_ensure_token_valid()
# Integrations rely on this to retry setup without mapping the error themselves.
assert isinstance(err.value, ConfigEntryNotReady)
assert err.value.translation_domain == HOMEASSISTANT_DOMAIN
assert err.value.translation_key == "oauth2_helper_refresh_transient"
assert f"Token request for {TEST_DOMAIN} got no response" in caplog.text
assert str(raised) in caplog.text
@pytest.mark.parametrize(
"response",
[
pytest.param({"access_token": ACCESS_TOKEN_2}, id="missing_expires_in"),
pytest.param(
{"access_token": ACCESS_TOKEN_2, "expires_in": "soon"},
id="unparsable_expires_in",
),
pytest.param(
{"access_token": ACCESS_TOKEN_2, "expires_in": None},
id="null_expires_in",
),
],
)
async def test_oauth_session_malformed_refresh_response_is_not_reauth(
hass: HomeAssistant,
local_impl: config_entry_oauth2_flow.LocalOAuth2Implementation,
aioclient_mock: AiohttpClientMocker,
response: dict[str, Any],
) -> None:
"""Test an unusable token response retries instead of blaming stored credentials."""
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"access_token": ACCESS_TOKEN_1,
"refresh_token": REFRESH_TOKEN,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
aioclient_mock.post(TOKEN_URL, json=response)
session = config_entry_oauth2_flow.OAuth2Session(hass, config_entry, local_impl)
with (
patch.object(config_entry, "async_start_reauth_if_available") as start_reauth,
pytest.raises(OAuth2TokenRequestConnectionError) as err,
):
await session.async_ensure_token_valid()
assert isinstance(err.value, ConfigEntryNotReady)
assert err.value.translation_domain == HOMEASSISTANT_DOMAIN
assert err.value.translation_key == "oauth2_helper_refresh_transient"
# Relinking the account cannot fix a bad response, so it must not ask for it.
start_reauth.assert_not_called()
@pytest.mark.parametrize(
"response",
[
pytest.param({"expires_in": 100}, id="no_access_token"),
pytest.param({"access_token": None, "expires_in": 100}, id="null_access_token"),
pytest.param({"access_token": "", "expires_in": 100}, id="blank_access_token"),
],
)
async def test_oauth_session_refresh_without_access_token_is_rejected(
hass: HomeAssistant,
local_impl: config_entry_oauth2_flow.LocalOAuth2Implementation,
aioclient_mock: AiohttpClientMocker,
response: dict[str, Any],
) -> None:
"""Test a response with no usable access token is not merged over the old one."""
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"access_token": ACCESS_TOKEN_1,
"refresh_token": REFRESH_TOKEN,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
aioclient_mock.post(TOKEN_URL, json=response)
session = config_entry_oauth2_flow.OAuth2Session(hass, config_entry, local_impl)
with pytest.raises(OAuth2TokenRequestConnectionError):
await session.async_ensure_token_valid()
# The stale token must stay expired so the next attempt refreshes again.
assert config_entry.data["token"]["access_token"] == ACCESS_TOKEN_1
assert config_entry.data["token"]["expires_at"] == 0
@pytest.mark.parametrize(
"refreshed",
[
pytest.param({"expires_in": 100}, id="no_access_token"),
pytest.param({"access_token": None, "expires_in": 100}, id="null_access_token"),
],
)
async def test_oauth_session_custom_implementation_without_access_token(
hass: HomeAssistant,
refreshed: dict[str, Any],
) -> None:
"""Test an implementation returning no usable access token is rejected."""
class BadImplementation(MockOAuth2Implementation):
"""Implementation whose refresh skips the local token request."""
async def _async_refresh_token(self, token: dict) -> dict:
"""Refresh a token."""
return refreshed
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"access_token": ACCESS_TOKEN_1,
"refresh_token": REFRESH_TOKEN,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, BadImplementation()
)
with pytest.raises(OAuth2TokenRequestConnectionError):
await session.async_ensure_token_valid()
assert config_entry.data["token"]["access_token"] == ACCESS_TOKEN_1
@pytest.mark.parametrize(
"refreshed",
[
pytest.param({"expires_at": FUTURE_EXPIRES_AT}, id="no_access_token"),
pytest.param({"access_token": ACCESS_TOKEN_2}, id="no_expires_at"),
],
)
async def test_oauth_session_never_stores_an_unusable_token(
hass: HomeAssistant,
refreshed: dict[str, Any],
) -> None:
"""Test the session checks the new token even when the refresh skips its own."""
class UncheckedImplementation(MockOAuth2Implementation):
"""Implementation overriding the public refresh, so nothing maps for it."""
async def async_refresh_token(self, token: dict) -> dict:
"""Refresh a token without the base class validation."""
return refreshed
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"access_token": ACCESS_TOKEN_1,
"refresh_token": REFRESH_TOKEN,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
session = config_entry_oauth2_flow.OAuth2Session(
hass, config_entry, UncheckedImplementation()
)
with pytest.raises(OAuth2TokenRequestConnectionError):
await session.async_ensure_token_valid()
assert config_entry.data["token"]["access_token"] == ACCESS_TOKEN_1
assert config_entry.data["token"]["expires_at"] == 0
@pytest.mark.parametrize(
("raised", "expected_exception"),
[
pytest.param(
ClientPayloadError("Disconnected"),
OAuth2TokenRequestConnectionError,
id="payload_error",
),
pytest.param(
ContentTypeError(
RequestInfo(
url=URL(TOKEN_URL),
method="POST",
headers=CIMultiDictProxy(CIMultiDict()),
),
(),
),
OAuth2TokenRequestError,
id="content_type_error",
),
],
)
async def test_oauth_session_refresh_body_error_is_mapped(
hass: HomeAssistant,
flow_handler: type[config_entry_oauth2_flow.AbstractOAuth2FlowHandler],
local_impl: config_entry_oauth2_flow.LocalOAuth2Implementation,
aioclient_mock: AiohttpClientMocker,
raised: Exception,
expected_exception: type[Exception],
) -> None:
"""Test a failure reading the token response body does not leak an aiohttp error."""
mock_integration(hass, MockModule(domain=TEST_DOMAIN))
flow_handler.async_register_implementation(hass, local_impl)
aioclient_mock.post(TOKEN_URL, json={})
config_entry = MockConfigEntry(
domain=TEST_DOMAIN,
data={
"auth_implementation": TEST_DOMAIN,
"token": {
"refresh_token": REFRESH_TOKEN,
"access_token": ACCESS_TOKEN_1,
"expires_at": 0,
},
},
)
config_entry.add_to_hass(hass)
session = config_entry_oauth2_flow.OAuth2Session(hass, config_entry, local_impl)
with (
patch.object(AiohttpClientMockResponse, "json", side_effect=raised),
pytest.raises(expected_exception) as err,
):
await session.async_ensure_token_valid()
assert type(err.value) is expected_exception
assert isinstance(err.value, ConfigEntryNotReady)
@pytest.mark.parametrize(
("raised", "expected_exception", "expected_base"),
[
pytest.param(
ClientResponseError(
RequestInfo(
url=URL(TOKEN_URL),
method="POST",
headers=CIMultiDictProxy(CIMultiDict()),
),
(),
status=HTTPStatus.UNAUTHORIZED,
),
OAuth2TokenRequestReauthError,
ConfigEntryAuthFailed,
id="reauth",
),
pytest.param(
ClientResponseError(
RequestInfo(
url=URL(TOKEN_URL),
method="POST",
headers=CIMultiDictProxy(CIMultiDict()),
),
(),
status=HTTPStatus.INTERNAL_SERVER_ERROR,
),
OAuth2TokenRequestTransientError,
ConfigEntryNotReady,
id="transient",
),
pytest.param(
ClientError("Cannot connect"),
OAuth2TokenRequestConnectionError,
ConfigEntryNotReady,
id="connection_error",
),
],
)
async def test_refresh_maps_errors_from_custom_implementation(
hass: HomeAssistant,
raised: Exception,
expected_exception: type[Exception],
expected_base: type[Exception],
) -> None:
"""Test an implementation issuing its own token request still raises mapped errors."""
class UnmappedImplementation(config_entry_oauth2_flow.LocalOAuth2Implementation):
"""Implementation that lets raw aiohttp errors escape, like a custom one."""
async def _async_refresh_token(self, token: dict) -> dict:
raise raised
implementation = UnmappedImplementation(
hass, TEST_DOMAIN, CLIENT_ID, CLIENT_SECRET, AUTHORIZE_URL, TOKEN_URL
)
with pytest.raises(expected_exception) as err:
await implementation.async_refresh_token({"refresh_token": REFRESH_TOKEN})
assert type(err.value) is expected_exception
assert isinstance(err.value, expected_base)
assert err.value.__cause__ is raised
@pytest.mark.parametrize(
"entry_state",
[
@@ -1534,8 +1895,8 @@ async def test_oauth2_request_replaces_caller_authorization_header(
),
pytest.param(
600,
config_entries.ConfigEntryState.SETUP_ERROR,
None,
config_entries.ConfigEntryState.SETUP_RETRY,
"oauth2_helper_refresh_failed",
id="generic",
),
],
@@ -1551,8 +1912,8 @@ async def test_token_error_handled_without_integration_mapping(
) -> None:
"""Test setup maps token refresh errors when the integration does not.
Only the transient and reauth subclasses carry config entry semantics, the
base error is left to the integration.
Every subclass carries config entry semantics, so the reauth subclass fails
setup while the others retry it.
"""
aioclient_mock.post(TOKEN_URL, status=status_code, json={})