home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-08-30 18:01:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Switch to August OAuth with official API (#151080)

Browse Source
This commit is contained in:
J. Nick Koston
2025-08-23 22:29:53 -05:00
committed by GitHub
parent 3c11f8e50e
commit 03ca164fb3
21 changed files with 897 additions and 1031 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
homeassistant/components/august/__init__.py
View File

@@ -6,18 +6,21 @@ from pathlib import Path
from typing import cast
from aiohttp import ClientResponseError
from yalexs.const import Brand
from yalexs.exceptions import AugustApiAIOHTTPError
from yalexs.manager.exceptions import CannotConnect, InvalidAuth, RequireValidation
from yalexs.manager.gateway import Config as YaleXSConfig
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr, issue_registry as ir
from homeassistant.helpers import (
config_entry_oauth2_flow,
device_registry as dr,
issue_registry as ir,
)
from .const import DOMAIN, PLATFORMS
from .const import DEFAULT_AUGUST_BRAND, DOMAIN, PLATFORMS
from .data import AugustData
from .gateway import AugustGateway
from .util import async_create_august_clientsession
@@ -25,30 +28,21 @@ from .util import async_create_august_clientsession
type AugustConfigEntry = ConfigEntry[AugustData]
@callback
def _async_create_yale_brand_migration_issue(
hass: HomeAssistant, entry: AugustConfigEntry
) -> None:
"""Create an issue for a brand migration."""
ir.async_create_issue(
hass,
DOMAIN,
"yale_brand_migration",
breaks_in_ha_version="2024.9",
learn_more_url="https://www.home-assistant.io/integrations/yale",
translation_key="yale_brand_migration",
is_fixable=False,
severity=ir.IssueSeverity.CRITICAL,
translation_placeholders={
"migrate_url": "https://my.home-assistant.io/redirect/config_flow_start?domain=yale"
},
)
async def async_setup_entry(hass: HomeAssistant, entry: AugustConfigEntry) -> bool:
"""Set up August from a config entry."""
# Check if this is a legacy config entry that needs migration to OAuth
if "auth_implementation" not in entry.data:
# This is a legacy entry using username/password, trigger reauth
raise ConfigEntryAuthFailed("Migration to OAuth required")
session = async_create_august_clientsession(hass)
august_gateway = AugustGateway(Path(hass.config.config_dir), session)
implementation = (
await config_entry_oauth2_flow.async_get_config_entry_implementation(
hass, entry
)
)
oauth_session = config_entry_oauth2_flow.OAuth2Session(hass, entry, implementation)
august_gateway = AugustGateway(Path(hass.config.config_dir), session, oauth_session)
try:
await async_setup_august(hass, entry, august_gateway)
except (RequireValidation, InvalidAuth) as err:
@@ -76,9 +70,7 @@ async def async_setup_august(
) -> None:
"""Set up the August component."""
config = cast(YaleXSConfig, entry.data)
await august_gateway.async_setup(config)
if august_gateway.api.brand == Brand.YALE_HOME:
_async_create_yale_brand_migration_issue(hass, entry)
await august_gateway.async_setup({**config, "brand": DEFAULT_AUGUST_BRAND})
await august_gateway.async_authenticate()
await august_gateway.async_refresh_access_token_if_needed()
data = entry.runtime_data = AugustData(hass, august_gateway)

15
homeassistant/components/august/application_credentials.py Normal file
View File

@@ -0,0 +1,15 @@
"""application_credentials platform for the august integration."""
from homeassistant.components.application_credentials import AuthorizationServer
from homeassistant.core import HomeAssistant
OAUTH2_AUTHORIZE = "https://auth.august.com/authorization"
OAUTH2_TOKEN = "https://auth.august.com/access_token"
async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
"""Return authorization server."""
return AuthorizationServer(
authorize_url=OAUTH2_AUTHORIZE,
token_url=OAUTH2_TOKEN,
)

314
homeassistant/components/august/config_flow.py
View File

@@ -1,284 +1,86 @@
"""Config flow for August integration."""
from collections.abc import Mapping
from dataclasses import dataclass
import logging
from pathlib import Path
from typing import Any
import aiohttp
import voluptuous as vol
from yalexs.authenticator_common import ValidationResult
from yalexs.const import BRANDS_WITHOUT_OAUTH, DEFAULT_BRAND, Brand
from yalexs.manager.exceptions import CannotConnect, InvalidAuth, RequireValidation
import jwt
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback
from .const import (
CONF_ACCESS_TOKEN_CACHE_FILE,
CONF_BRAND,
CONF_LOGIN_METHOD,
DEFAULT_LOGIN_METHOD,
DOMAIN,
LOGIN_METHODS,
VERIFICATION_CODE_KEY,
)
from .gateway import AugustGateway
from .util import async_create_august_clientsession
# The Yale Home Brand is not supported by the August integration
# anymore and should migrate to the Yale integration
AVAILABLE_BRANDS = BRANDS_WITHOUT_OAUTH.copy()
del AVAILABLE_BRANDS[Brand.YALE_HOME]
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult
from homeassistant.helpers import config_entry_oauth2_flow
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
async def async_validate_input(
data: dict[str, Any], august_gateway: AugustGateway
) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
Request configuration steps from the user.
"""
assert august_gateway.authenticator is not None
authenticator = august_gateway.authenticator
if (code := data.get(VERIFICATION_CODE_KEY)) is not None:
result = await authenticator.async_validate_verification_code(code)
_LOGGER.debug("Verification code validation: %s", result)
if result != ValidationResult.VALIDATED:
raise RequireValidation
try:
await august_gateway.async_authenticate()
except RequireValidation:
_LOGGER.debug(
"Requesting new verification code for %s via %s",
data.get(CONF_USERNAME),
data.get(CONF_LOGIN_METHOD),
)
if code is None:
await august_gateway.authenticator.async_send_verification_code()
raise
return {
"title": data.get(CONF_USERNAME),
"data": august_gateway.config_entry(),
}
@dataclass(slots=True)
class ValidateResult:
"""Result from validation."""
validation_required: bool
info: dict[str, Any]
errors: dict[str, str]
description_placeholders: dict[str, str]
class AugustConfigFlow(ConfigFlow, domain=DOMAIN):
class AugustConfigFlow(
config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
):
"""Handle a config flow for August."""
VERSION = 1
DOMAIN = DOMAIN
def __init__(self) -> None:
"""Store an AugustGateway()."""
self._august_gateway: AugustGateway | None = None
self._aiohttp_session: aiohttp.ClientSession | None = None
self._user_auth_details: dict[str, Any] = {}
self._needs_reset = True
super().__init__()
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle the initial step."""
return await self.async_step_user_validate()
async def async_step_user_validate(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle authentication."""
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
if user_input is not None:
self._user_auth_details.update(user_input)
validate_result = await self._async_auth_or_validate()
description_placeholders = validate_result.description_placeholders
if validate_result.validation_required:
return await self.async_step_validation()
if not (errors := validate_result.errors):
return await self._async_update_or_create_entry(validate_result.info)
return self.async_show_form(
step_id="user_validate",
data_schema=vol.Schema(
{
vol.Required(
CONF_BRAND,
default=self._user_auth_details.get(CONF_BRAND, DEFAULT_BRAND),
): vol.In(AVAILABLE_BRANDS),
vol.Required(
CONF_LOGIN_METHOD,
default=self._user_auth_details.get(
CONF_LOGIN_METHOD, DEFAULT_LOGIN_METHOD
),
): vol.In(LOGIN_METHODS),
vol.Required(
CONF_USERNAME,
default=self._user_auth_details.get(CONF_USERNAME),
): str,
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
description_placeholders=description_placeholders,
)
async def async_step_validation(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle validation (2fa) step."""
if user_input:
if self.source == SOURCE_REAUTH:
return await self.async_step_reauth_validate(user_input)
return await self.async_step_user_validate(user_input)
previously_failed = VERIFICATION_CODE_KEY in self._user_auth_details
return self.async_show_form(
step_id="validation",
data_schema=vol.Schema(
{vol.Required(VERIFICATION_CODE_KEY): vol.All(str, vol.Strip)}
),
errors={"base": "invalid_verification_code"} if previously_failed else None,
description_placeholders={
CONF_BRAND: self._user_auth_details[CONF_BRAND],
CONF_USERNAME: self._user_auth_details[CONF_USERNAME],
CONF_LOGIN_METHOD: self._user_auth_details[CONF_LOGIN_METHOD],
},
)
@callback
def _async_get_gateway(self) -> AugustGateway:
"""Set up the gateway."""
if self._august_gateway is not None:
return self._august_gateway
self._aiohttp_session = async_create_august_clientsession(self.hass)
self._august_gateway = AugustGateway(
Path(self.hass.config.config_dir), self._aiohttp_session
)
return self._august_gateway
@callback
def _async_shutdown_gateway(self) -> None:
"""Shutdown the gateway."""
if self._aiohttp_session is not None:
self._aiohttp_session.detach()
self._august_gateway = None
@property
def logger(self) -> logging.Logger:
"""Return logger."""
return _LOGGER
async def async_step_reauth(
self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
"""Handle configuration by re-auth."""
self._user_auth_details = dict(entry_data)
return await self.async_step_reauth_validate()
return await self.async_step_user()
async def async_step_reauth_validate(
self, user_input: dict[str, Any] | None = None
def _async_decode_jwt(self, encoded: str) -> dict[str, Any]:
"""Decode JWT token."""
return jwt.decode(
encoded,
"",
verify=False,
options={"verify_signature": False},
algorithms=["HS256"],
)
async def _async_handle_reauth(
self, data: dict, decoded: dict[str, Any], user_id: str
) -> ConfigFlowResult:
"""Handle reauth and validation."""
errors: dict[str, str] = {}
description_placeholders: dict[str, str] = {}
if user_input is not None:
self._user_auth_details.update(user_input)
validate_result = await self._async_auth_or_validate()
description_placeholders = validate_result.description_placeholders
if validate_result.validation_required:
return await self.async_step_validation()
if not (errors := validate_result.errors):
return await self._async_update_or_create_entry(validate_result.info)
"""Handle reauth flow."""
reauth_entry = self._get_reauth_entry()
assert reauth_entry.unique_id is not None
# Check if this is a migration from username (contains @) to user ID
if "@" not in reauth_entry.unique_id:
# This is a normal oauth reauth, enforce ID matching for security
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_mismatch(reason="reauth_invalid_user")
return self.async_update_reload_and_abort(reauth_entry, data=data)
return self.async_show_form(
step_id="reauth_validate",
data_schema=vol.Schema(
{
vol.Required(
CONF_BRAND,
default=self._user_auth_details.get(CONF_BRAND, DEFAULT_BRAND),
): vol.In(BRANDS_WITHOUT_OAUTH),
vol.Required(CONF_PASSWORD): str,
}
),
errors=errors,
description_placeholders=description_placeholders
| {
CONF_USERNAME: self._user_auth_details[CONF_USERNAME],
},
# This is a one-time migration from username to user ID
# Only validate if the account has emails
emails: list[str]
if emails := decoded.get("email", []):
# Validate that the email matches before allowing migration
email_to_check_lower = reauth_entry.unique_id.casefold()
if not any(email.casefold() == email_to_check_lower for email in emails):
# Email doesn't match - this is a different account
return self.async_abort(reason="reauth_invalid_user")
# Email matches or no emails on account, update with new unique ID
return self.async_update_reload_and_abort(
reauth_entry, data=data, unique_id=user_id
)
async def _async_reset_access_token_cache_if_needed(
self, gateway: AugustGateway, username: str, access_token_cache_file: str | None
) -> None:
"""Reset the access token cache if needed."""
# We need to configure the access token cache file before we setup the gateway
# since we need to reset it if the brand changes BEFORE we setup the gateway
gateway.async_configure_access_token_cache_file(
username, access_token_cache_file
)
if self._needs_reset:
self._needs_reset = False
await gateway.async_reset_authentication()
async def async_oauth_create_entry(self, data: dict) -> ConfigFlowResult:
"""Create an entry for the flow."""
# Decode JWT once
access_token = data["token"]["access_token"]
decoded = self._async_decode_jwt(access_token)
user_id = decoded["userId"]
async def _async_auth_or_validate(self) -> ValidateResult:
"""Authenticate or validate."""
user_auth_details = self._user_auth_details
gateway = self._async_get_gateway()
assert gateway is not None
await self._async_reset_access_token_cache_if_needed(
gateway,
user_auth_details[CONF_USERNAME],
user_auth_details.get(CONF_ACCESS_TOKEN_CACHE_FILE),
)
await gateway.async_setup(user_auth_details)
if self.source == SOURCE_REAUTH:
return await self._async_handle_reauth(data, decoded, user_id)
errors: dict[str, str] = {}
info: dict[str, Any] = {}
description_placeholders: dict[str, str] = {}
validation_required = False
try:
info = await async_validate_input(user_auth_details, gateway)
except CannotConnect:
errors["base"] = "cannot_connect"
except InvalidAuth:
errors["base"] = "invalid_auth"
except RequireValidation:
validation_required = True
except Exception as ex:
_LOGGER.exception("Unexpected exception")
errors["base"] = "unhandled"
description_placeholders = {"error": str(ex)}
return ValidateResult(
validation_required, info, errors, description_placeholders
)
async def _async_update_or_create_entry(
self, info: dict[str, Any]
) -> ConfigFlowResult:
"""Update existing entry or create a new one."""
self._async_shutdown_gateway()
existing_entry = await self.async_set_unique_id(
self._user_auth_details[CONF_USERNAME]
)
if not existing_entry:
return self.async_create_entry(title=info["title"], data=info["data"])
return self.async_update_reload_and_abort(existing_entry, data=info["data"])
await self.async_set_unique_id(user_id)
self._abort_if_unique_id_configured()
return await super().async_oauth_create_entry(data)

4
homeassistant/components/august/const.py
View File

@@ -1,5 +1,7 @@
"""Constants for August devices."""
from yalexs.const import Brand
from homeassistant.const import Platform
DEFAULT_TIMEOUT = 25
@@ -9,6 +11,8 @@ CONF_BRAND = "brand"
CONF_LOGIN_METHOD = "login_method"
CONF_INSTALL_ID = "install_id"
DEFAULT_AUGUST_BRAND = Brand.YALE_AUGUST
VERIFICATION_CODE_KEY = "verification_code"
NOTIFICATION_ID = "august_notification"

51
homeassistant/components/august/gateway.py
View File

@@ -1,30 +1,43 @@
"""Handle August connection setup and authentication."""
from typing import Any
import logging
from pathlib import Path
from yalexs.const import DEFAULT_BRAND
from aiohttp import ClientSession
from yalexs.authenticator_common import Authentication, AuthenticationState
from yalexs.manager.gateway import Gateway
from homeassistant.const import CONF_USERNAME
from homeassistant.helpers import config_entry_oauth2_flow
from .const import (
CONF_ACCESS_TOKEN_CACHE_FILE,
CONF_BRAND,
CONF_INSTALL_ID,
CONF_LOGIN_METHOD,
)
_LOGGER = logging.getLogger(__name__)
class AugustGateway(Gateway):
"""Handle the connection to August."""
def config_entry(self) -> dict[str, Any]:
"""Config entry."""
assert self._config is not None
return {
CONF_BRAND: self._config.get(CONF_BRAND, DEFAULT_BRAND),
CONF_LOGIN_METHOD: self._config[CONF_LOGIN_METHOD],
CONF_USERNAME: self._config[CONF_USERNAME],
CONF_INSTALL_ID: self._config.get(CONF_INSTALL_ID),
CONF_ACCESS_TOKEN_CACHE_FILE: self._access_token_cache_file,
}
def __init__(
self,
config_path: Path,
aiohttp_session: ClientSession,
oauth_session: config_entry_oauth2_flow.OAuth2Session,
) -> None:
"""Init the connection."""
super().__init__(config_path, aiohttp_session)
self._oauth_session = oauth_session
async def async_get_access_token(self) -> str:
"""Get access token."""
await self._oauth_session.async_ensure_token_valid()
return self._oauth_session.token["access_token"]
async def async_refresh_access_token_if_needed(self) -> None:
"""Refresh the access token if needed."""
await self._oauth_session.async_ensure_token_valid()
async def async_authenticate(self) -> Authentication:
"""Authenticate with the details provided to setup."""
await self._oauth_session.async_ensure_token_valid()
self.authentication = Authentication(
AuthenticationState.AUTHENTICATED, None, None, None
)
return self.authentication

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

@@ -3,6 +3,7 @@
"name": "August",
"codeowners": ["@bdraco"],
"config_flow": true,
"dependencies": ["application_credentials", "cloud"],
"dhcp": [
{
"hostname": "connect",

56
homeassistant/components/august/strings.json
View File

@@ -6,42 +6,34 @@
}
},
"config": {
"error": {
"unhandled": "Unhandled error: {error}",
"invalid_verification_code": "Invalid verification code",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]"
"step": {
"pick_implementation": {
"title": "[%key:common::config_flow::title::oauth2_pick_implementation%]",
"data": {
"implementation": "[%key:common::config_flow::data::implementation%]"
},
"data_description": {
"implementation": "[%key:common::config_flow::description::implementation%]"
}
}
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"oauth_error": "[%key:common::config_flow::abort::oauth2_error%]",
"missing_configuration": "[%key:common::config_flow::abort::oauth2_missing_configuration%]",
"missing_credentials": "[%key:common::config_flow::abort::oauth2_missing_credentials%]",
"authorize_url_timeout": "[%key:common::config_flow::abort::oauth2_authorize_url_timeout%]",
"no_url_available": "[%key:common::config_flow::abort::oauth2_no_url_available%]",
"user_rejected_authorize": "[%key:common::config_flow::abort::oauth2_user_rejected_authorize%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
"oauth_timeout": "[%key:common::config_flow::abort::oauth2_timeout%]",
"oauth_unauthorized": "[%key:common::config_flow::abort::oauth2_unauthorized%]",
"oauth_failed": "[%key:common::config_flow::abort::oauth2_failed%]",
"reauth_invalid_user": "Reauthenticate must use the same account."
},
"step": {
"validation": {
"title": "Two-factor authentication",
"data": {
"verification_code": "Verification code"
},
"description": "Please check your {login_method} ({username}) and enter the verification code below. Codes may take a few minutes to arrive."
},
"user_validate": {
"description": "It is recommended to use the 'email' login method as some brands may not work with the 'phone' method. If the Login Method is 'email', Username is the email address. If the Login Method is 'phone', Username is the phone number in the format '+NNNNNNNNN'. If you choose the wrong brand, you may be able to authenticate initially; however, you will not be able to operate devices. If you are unsure of the brand, create the integration again and try another brand.",
"data": {
"brand": "Brand",
"login_method": "Login Method",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]"
},
"title": "Set up an August account"
},
"reauth_validate": {
"description": "Choose the correct brand for your device, and enter the password for {username}. If you choose the wrong brand, you may be able to authenticate initially; however, you will not be able to operate devices. If you are unsure of the brand, create the integration again and try another brand.",
"data": {
"brand": "[%key:component::august::config::step::user_validate::data::brand%]",
"password": "[%key:common::config_flow::data::password%]"
},
"title": "Reauthenticate an August account"
}
"create_entry": {
"default": "[%key:common::config_flow::create_entry::authenticated%]"
}
},
"entity": {

1
homeassistant/generated/application_credentials.py generated
View File

@@ -4,6 +4,7 @@ To update, run python3 -m script.hassfest
"""
APPLICATION_CREDENTIALS = [
"august",
"electric_kiwi",
"fitbit",
"geocaching",

115
tests/components/august/conftest.py
View File

@@ -5,6 +5,19 @@ from unittest.mock import patch
import pytest
from yalexs.manager.ratelimit import _RateLimitChecker
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.august.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
USER_ID = "a76c25e5-49aa-4c14-cd0c-48a6931e2081"
CLIENT_ID = "1"
@pytest.fixture(name="mock_discovery", autouse=True)
def mock_discovery_fixture():
@@ -20,3 +33,105 @@ def disable_ratelimit_checks_fixture():
"""Disable rate limit checks."""
with patch.object(_RateLimitChecker, "register_wakeup"):
yield
@pytest.fixture(name="mock_config_entry")
def mock_config_entry_fixture(jwt: str) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
data={
"auth_implementation": "august",
"token": {
"access_token": jwt,
"scope": "any",
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": "mock-user-id",
"expires_at": 1697753347,
},
},
unique_id=USER_ID,
)
@pytest.fixture(name="mock_legacy_config_entry")
def mock_legacy_config_entry_fixture() -> MockConfigEntry:
"""Return a legacy config entry without OAuth data."""
return MockConfigEntry(
domain=DOMAIN,
data={
"login_method": "email",
"username": "my@email.tld",
"password": "test-password",
"install_id": None,
"timeout": 10,
"access_token_cache_file": ".my@email.tld.august.conf",
},
unique_id="my@email.tld",
)
@pytest.fixture(name="jwt")
def load_jwt_fixture() -> str:
"""Load Fixture data."""
return load_fixture("jwt", DOMAIN).strip("\n")
@pytest.fixture(name="legacy_jwt")
def load_legacy_jwt_fixture() -> str:
"""Load legacy JWT fixture data."""
return load_fixture("legacy_jwt", DOMAIN).strip("\n")
@pytest.fixture(name="reauth_jwt")
def load_reauth_jwt_fixture() -> str:
"""Load reauth JWT fixture data."""
return load_fixture("reauth_jwt", DOMAIN).strip("\n")
@pytest.fixture(name="migration_jwt")
def load_migration_jwt_fixture() -> str:
"""Load migration JWT fixture data (has email for legacy migration)."""
return load_fixture("migration_jwt", DOMAIN).strip("\n")
@pytest.fixture(name="reauth_jwt_wrong_account")
def load_reauth_jwt_wrong_account_fixture() -> str:
"""Load JWT fixture data for wrong account during reauth."""
# Different userId, no email match
return "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpbnN0YWxsSWQiOiIiLCJyZWdpb24iOiJpcmVsYW5kLXByb2QtYXdzIiwiYXBwbGljYXRpb25JZCI6IiIsInVzZXJJZCI6ImRpZmZlcmVudC11c2VyLWlkIiwidkluc3RhbGxJZCI6ZmFsc2UsInZQYXNzd29yZCI6dHJ1ZSwidkVtYWlsIjp0cnVlLCJ2UGhvbmUiOnRydWUsImhhc0luc3RhbGxJZCI6ZmFsc2UsImhhc1Bhc3N3b3JkIjpmYWxzZSwiaGFzRW1haWwiOmZhbHNlLCJoYXNQaG9uZSI6ZmFsc2UsImlzTG9ja2VkT3V0IjpmYWxzZSwiY2FwdGNoYSI6IiIsImVtYWlsIjpbImRpZmZlcmVudEBlbWFpbC50bGQiXSwicGhvbmUiOltdLCJleHBpcmVzQXQiOiIyMDI0LTEyLTE4VDEzOjU0OjA1LjEzNFoiLCJ0ZW1wb3JhcnlBY2NvdW50Q3JlYXRpb25QYXNzd29yZExpbmsiOiIiLCJpYXQiOjE3MjQxNjIwNDUsImV4cCI6MTczNDUzMDA0NSwib2F1dGgiOnsiYXBwX25hbWUiOiJIb21lIEFzc2lzdGFudCIsImNsaWVudF9pZCI6ImIzY2QzZjBiLWZiOTctNGQ2Yy1iZWU5LWFmN2FiMDQ3NThjNyIsInJlZGlyZWN0X3VyaSI6Imh0dHBzOi8vYWNjb3VudC1saW5rLm5hYnVjYXNhLmNvbS9hdXRob3JpemVfY2FsbGJhY2siLCJwYXJ0bmVyX2lkIjoiNjU3OTc0ODgxMDY2Y2E0OGM5OWMwODI2In19.mK9nTAv7glYgtpLIkVF_dsrjrkRKYemdKfKMkgnafCU"
@pytest.fixture(name="client_credentials", autouse=True)
async def mock_client_credentials_fixture(hass: HomeAssistant) -> None:
"""Mock client credentials."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, "2"),
DOMAIN,
)
@pytest.fixture(name="skip_cloud", autouse=True)
def skip_cloud_fixture():
"""Skip setting up cloud.
Cloud already has its own tests for account link.
We do not need to test it here as we only need to test our
usage of the oauth2 helpers.
"""
with patch("homeassistant.components.cloud.async_setup", return_value=True):
yield
@pytest.fixture
def mock_setup_entry():
"""Mock setup entry."""
with patch(
"homeassistant.components.august.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry

1
tests/components/august/fixtures/jwt Normal file
View File

@@ -0,0 +1 @@
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpbnN0YWxsSWQiOiIiLCJyZWdpb24iOiJpcmVsYW5kLXByb2QtYXdzIiwiYXBwbGljYXRpb25JZCI6IiIsInVzZXJJZCI6ImE3NmMyNWU1LTQ5YWEtNGMxNC1jZDBjLTQ4YTY5MzFlMjA4MSIsInZJbnN0YWxsSWQiOmZhbHNlLCJ2UGFzc3dvcmQiOnRydWUsInZFbWFpbCI6dHJ1ZSwidlBob25lIjp0cnVlLCJoYXNJbnN0YWxsSWQiOmZhbHNlLCJoYXNQYXNzd29yZCI6ZmFsc2UsImhhc0VtYWlsIjpmYWxzZSwiaGFzUGhvbmUiOmZhbHNlLCJpc0xvY2tlZE91dCI6ZmFsc2UsImNhcHRjaGEiOiIiLCJlbWFpbCI6W10sInBob25lIjpbXSwiZXhwaXJlc0F0IjoiMjAyNC0xMi0xOFQxMzo1NDowNS4xMzRaIiwidGVtcG9yYXJ5QWNjb3VudENyZWF0aW9uUGFzc3dvcmRMaW5rIjoiIiwiaWF0IjoxNzI0MTYyMDQ1LCJleHAiOjE3MzQ1MzAwNDUsIm9hdXRoIjp7ImFwcF9uYW1lIjoiSG9tZSBBc3Npc3RhbnQiLCJjbGllbnRfaWQiOiJiM2NkM2YwYi1mYjk3LTRkNmMtYmVlOS1hZjdhYjA0NzU4YzciLCJyZWRpcmVjdF91cmkiOiJodHRwczovL2FjY291bnQtbGluay5uYWJ1Y2FzYS5jb20vYXV0aG9yaXplX2NhbGxiYWNrIiwicGFydG5lcl9pZCI6IjY1Nzk3NDg4MTA2NmNhNDhjOTljMDgyNiJ9fQ.BdRo-dEr-osbDQGB2XzlI-mIj4gqULtapODt-sj-eA8

1
tests/components/august/fixtures/legacy_jwt Normal file
View File

@@ -0,0 +1 @@
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpbnN0YWxsSWQiOiIiLCJyZWdpb24iOiJpcmVsYW5kLXByb2QtYXdzIiwiYXBwbGljYXRpb25JZCI6IiIsInVzZXJJZCI6ImxlZ2FjeS11c2VyLWlkIiwidkluc3RhbGxJZCI6ZmFsc2UsInZQYXNzd29yZCI6dHJ1ZSwidkVtYWlsIjp0cnVlLCJ2UGhvbmUiOnRydWUsImhhc0luc3RhbGxJZCI6ZmFsc2UsImhhc1Bhc3N3b3JkIjpmYWxzZSwiaGFzRW1haWwiOmZhbHNlLCJoYXNQaG9uZSI6ZmFsc2UsImlzTG9ja2VkT3V0IjpmYWxzZSwiY2FwdGNoYSI6IiIsImVtYWlsIjpbIm15QGVtYWlsLnRsZCJdLCJwaG9uZSI6W10sImV4cGlyZXNBdCI6IjIwMjQtMTItMThUMTM6NTQ6MDUuMTM0WiIsInRlbXBvcmFyeUFjY291bnRDcmVhdGlvblBhc3N3b3JkTGluayI6IiIsImlhdCI6MTcyNDE2MjA0NSwiZXhwIjoxNzM0NTMwMDQ1LCJvYXV0aCI6eyJhcHBfbmFtZSI6IkhvbWUgQXNzaXN0YW50IiwiY2xpZW50X2lkIjoiYjNjZDNmMGItZmI5Ny00ZDZjLWJlZTktYWY3YWIwNDc1OGM3IiwicmVkaXJlY3RfdXJpIjoiaHR0cHM6Ly9hY2NvdW50LWxpbmsubmFidWNhc2EuY29tL2F1dGhvcml6ZV9jYWxsYmFjayIsInBhcnRuZXJfaWQiOiI2NTc5NzQ4ODEwNjZjYTQ4Yzk5YzA4MjYifX0.BdRo-dEr-osbDQGB2XzlI-mIj4gqULtapODt-sj-eA8

1
tests/components/august/fixtures/migration_jwt Normal file
View File

@@ -0,0 +1 @@
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpbnN0YWxsSWQiOiIiLCJyZWdpb24iOiJpcmVsYW5kLXByb2QtYXdzIiwiYXBwbGljYXRpb25JZCI6IiIsInVzZXJJZCI6ImE3NmMyNWU1LTQ5YWEtNGMxNC1jZDBjLTQ4YTY5MzFlMjA4MSIsInZJbnN0YWxsSWQiOmZhbHNlLCJ2UGFzc3dvcmQiOnRydWUsInZFbWFpbCI6dHJ1ZSwidlBob25lIjp0cnVlLCJoYXNJbnN0YWxsSWQiOmZhbHNlLCJoYXNQYXNzd29yZCI6ZmFsc2UsImhhc0VtYWlsIjpmYWxzZSwiaGFzUGhvbmUiOmZhbHNlLCJpc0xvY2tlZE91dCI6ZmFsc2UsImNhcHRjaGEiOiIiLCJlbWFpbCI6WyJteUBlbWFpbC50bGQiXSwicGhvbmUiOltdLCJleHBpcmVzQXQiOiIyMDI0LTEyLTE4VDEzOjU0OjA1LjEzNFoiLCJ0ZW1wb3JhcnlBY2NvdW50Q3JlYXRpb25QYXNzd29yZExpbmsiOiIiLCJpYXQiOjE3MjQxNjIwNDUsImV4cCI6MTczNDUzMDA0NSwib2F1dGgiOnsiYXBwX25hbWUiOiJIb21lIEFzc2lzdGFudCIsImNsaWVudF9pZCI6ImIzY2QzZjBiLWZiOTctNGQ2Yy1iZWU5LWFmN2FiMDQ3NThjNyIsInJlZGlyZWN0X3VyaSI6Imh0dHBzOi8vYWNjb3VudC1saW5rLm5hYnVjYXNhLmNvbS9hdXRob3JpemVfY2FsbGJhY2siLCJwYXJ0bmVyX2lkIjoiNjU3OTc0ODgxMDY2Y2E0OGM5OWMwODI2In19.migration-token

1
tests/components/august/fixtures/reauth_jwt Normal file
View File

@@ -0,0 +1 @@
eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpbnN0YWxsSWQiOiIiLCJyZWdpb24iOiJpcmVsYW5kLXByb2QtYXdzIiwiYXBwbGljYXRpb25JZCI6IiIsInVzZXJJZCI6ImE3NmMyNWU1LTQ5YWEtNGMxNC1jZDBjLTQ4YTY5MzFlMjA4MSIsInZJbnN0YWxsSWQiOmZhbHNlLCJ2UGFzc3dvcmQiOnRydWUsInZFbWFpbCI6dHJ1ZSwidlBob25lIjp0cnVlLCJoYXNJbnN0YWxsSWQiOmZhbHNlLCJoYXNQYXNzd29yZCI6ZmFsc2UsImhhc0VtYWlsIjpmYWxzZSwiaGFzUGhvbmUiOmZhbHNlLCJpc0xvY2tlZG91dCI6ZmFsc2UsImNhcHRjaGEiOiIiLCJlbWFpbCI6WyJteUBlbWFpbC50bGQiXSwicGhvbmUiOltdLCJleHBpcmVzQXQiOiIyMDI0LTEyLTE4VDEzOjU0OjA1LjEzNFoiLCJ0ZW1wb3JhcnlBY2NvdW50Q3JlYXRpb25QYXNzd29yZExpbmsiOiIiLCJpYXQiOjE3MjQxNjIwNDUsImV4cCI6MTczNDUzMDA0NSwib2F1dGgiOnsiYXBwX25hbWUiOiJIb21lIEFzc2lzdGFudCIsImNsaWVudF9pZCI6ImIzY2QzZjBiLWZiOTctNGQ2Yy1iZWU5LWFmN2FiMDQ3NThjNyIsInJlZGlyZWN0X3VyaSI6Imh0dHBzOi8vYWNjb3VudC1saW5rLm5hYnVjYXNhLmNvbS9hdXRob3JpemVfY2FsbGJhY2siLCJwYXJ0bmVyX2lkIjoiNjU3OTc0ODgxMDY2Y2E0OGM5OWMwODI2In19.reauth-updated-token

223
tests/components/august/mocks.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Iterable
from contextlib import contextmanager
import json
import os
import time
@@ -26,98 +27,184 @@ from yalexs.activity import (
DoorOperationActivity,
LockOperationActivity,
)
from yalexs.authenticator_common import AuthenticationState
from yalexs.api_async import ApiAsync
from yalexs.authenticator_common import Authentication, AuthenticationState
from yalexs.const import Brand
from yalexs.doorbell import Doorbell, DoorbellDetail
from yalexs.lock import Lock, LockDetail
from yalexs.manager.ratelimit import _RateLimitChecker
from yalexs.pubnub_async import AugustPubNub
from homeassistant.components.august.const import CONF_BRAND, CONF_LOGIN_METHOD, DOMAIN
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.august.const import DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
USER_ID = "a76c25e5-49aa-4c14-cd0c-48a6931e2081"
def _mock_get_config(brand: Brand = Brand.AUGUST):
def _mock_get_config(
brand: Brand = Brand.YALE_AUGUST, jwt: str | None = None
) -> dict[str, Any]:
"""Return a default august config."""
return {
DOMAIN: {
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "mocked_username",
CONF_PASSWORD: "mocked_password",
CONF_BRAND: brand,
"auth_implementation": "august",
"token": {
"access_token": jwt or "access_token",
"expires_in": 1,
"refresh_token": "refresh_token",
"expires_at": time.time() + 3600,
"service": "august",
},
}
}
def _mock_authenticator(auth_state):
def _mock_authenticator(auth_state: AuthenticationState) -> Authentication:
"""Mock an august authenticator."""
authenticator = MagicMock()
type(authenticator).state = PropertyMock(return_value=auth_state)
return authenticator
def _timetoken():
def _timetoken() -> str:
return str(time.time_ns())[:-2]
@patch("yalexs.manager.gateway.ApiAsync")
@patch("yalexs.manager.gateway.AuthenticatorAsync.async_authenticate")
async def _mock_setup_august(
hass: HomeAssistant, api_instance, pubnub_mock, authenticate_mock, api_mock, brand
async def mock_august_config_entry(
hass: HomeAssistant,
) -> MockConfigEntry:
"""Set up august integration."""
authenticate_mock.side_effect = MagicMock(
return_value=_mock_august_authentication(
"original_token", 1234, AuthenticationState.AUTHENTICATED
)
)
api_mock.return_value = api_instance
entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config(brand)[DOMAIN],
options={},
)
"""Mock august config entry and client credentials."""
entry = mock_config_entry()
entry.add_to_hass(hass)
return entry
def mock_config_entry(jwt: str | None = None) -> MockConfigEntry:
"""Return the default mocked config entry."""
return MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config(jwt=jwt)[DOMAIN],
options={},
unique_id=USER_ID,
)
async def mock_client_credentials(hass: HomeAssistant) -> ClientCredential:
"""Mock client credentials."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential("1", "2"),
DOMAIN,
)
@contextmanager
def patch_august_setup():
"""Patch august setup process."""
with (
patch("yalexs.manager.gateway.ApiAsync") as api_mock,
patch.object(_RateLimitChecker, "register_wakeup") as authenticate_mock,
patch(
"homeassistant.components.august.config_entry_oauth2_flow.async_get_config_entry_implementation"
),
):
yield api_mock, authenticate_mock
async def _mock_setup_august(
hass: HomeAssistant,
api_instance: ApiAsync,
pubnub_mock: AugustPubNub,
authenticate_side_effect: MagicMock,
) -> ConfigEntry:
"""Set up august integration."""
entry = await mock_august_config_entry(hass)
with (
patch_august_setup() as patched_setup,
patch.object(pubnub_mock, "run"),
patch("yalexs.manager.data.AugustPubNub", return_value=pubnub_mock),
):
assert await hass.config_entries.async_setup(entry.entry_id)
api_mock, authenticate_mock = patched_setup
authenticate_mock.side_effect = authenticate_side_effect
api_mock.return_value = api_instance
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
return entry
async def _create_august_with_devices(
hass: HomeAssistant,
devices: Iterable[LockDetail | DoorbellDetail],
devices: Iterable[LockDetail | DoorbellDetail] | None = None,
api_call_side_effects: dict[str, Any] | None = None,
activities: list[Any] | None = None,
pubnub: AugustPubNub | None = None,
brand: Brand = Brand.AUGUST,
) -> ConfigEntry:
entry, _ = await _create_august_api_with_devices(
hass, devices, api_call_side_effects, activities, pubnub, brand
brand: Brand = Brand.YALE_AUGUST,
authenticate_side_effect: MagicMock | None = None,
) -> tuple[ConfigEntry, AugustPubNub]:
entry, _, pubnub_instance = await _create_august_api_with_devices(
hass,
devices,
api_call_side_effects,
activities,
pubnub,
brand,
authenticate_side_effect,
)
return entry
return entry, pubnub_instance
async def _create_august_api_with_devices(
hass: HomeAssistant,
devices: Iterable[LockDetail | DoorbellDetail],
devices: Iterable[LockDetail | DoorbellDetail] | None = None,
api_call_side_effects: dict[str, Any] | None = None,
activities: list[Any] | None = None,
activities: dict[str, Any] | None = None,
pubnub: AugustPubNub | None = None,
brand: Brand = Brand.AUGUST,
) -> tuple[MockConfigEntry, MagicMock]:
brand: Brand = Brand.YALE_AUGUST,
authenticate_side_effect: MagicMock | None = None,
) -> tuple[ConfigEntry, ApiAsync, AugustPubNub]:
if api_call_side_effects is None:
api_call_side_effects = {}
if devices is None:
devices = ()
update_api_call_side_effects(api_call_side_effects, devices, activities)
api_instance = await make_mock_api(api_call_side_effects, brand)
if pubnub is None:
pubnub = AugustPubNub()
pubnub.run = AsyncMock()
entry = await _mock_setup_august(
hass,
api_instance,
pubnub,
authenticate_side_effect=authenticate_side_effect,
)
return entry, api_instance, pubnub
def update_api_call_side_effects(
api_call_side_effects: dict[str, Any],
devices: Iterable[LockDetail | DoorbellDetail],
activities: dict[str, Any] | None = None,
) -> None:
"""Update side effects dict from devices and activities."""
device_data = {"doorbells": [], "locks": []}
for device in devices:
for device in devices or ():
if isinstance(device, LockDetail):
device_data["locks"].append(
{"base": _mock_august_lock(device.device_id), "detail": device}
@@ -127,7 +214,7 @@ async def _create_august_api_with_devices(
{
"base": _mock_august_doorbell(
deviceid=device.device_id,
brand=device._data.get("brand", Brand.AUGUST),
brand=device._data.get("brand", Brand.YALE_AUGUST),
),
"detail": device,
}
@@ -200,24 +287,12 @@ async def _create_august_api_with_devices(
"async_unlatch_return_activities", unlock_return_activities_side_effect
)
api_instance, entry = await _mock_setup_august_with_api_side_effects(
hass, api_call_side_effects, pubnub, brand
)
if device_data["locks"]:
# Ensure we sync status when the integration is loaded if there
# are any locks
assert api_instance.async_status_async.mock_calls
return entry, api_instance
async def _mock_setup_august_with_api_side_effects(
hass: HomeAssistant,
async def make_mock_api(
api_call_side_effects: dict[str, Any],
pubnub: AugustPubNub,
brand: Brand = Brand.AUGUST,
):
brand: Brand = Brand.YALE_AUGUST,
) -> ApiAsync:
"""Make a mock ApiAsync instance."""
api_instance = MagicMock(name="Api", brand=brand)
if api_call_side_effects["get_lock_detail"]:
@@ -267,12 +342,12 @@ async def _mock_setup_august_with_api_side_effects(
api_instance.async_unlatch_async = AsyncMock()
api_instance.async_unlatch = AsyncMock()
return api_instance, await _mock_setup_august(
hass, api_instance, pubnub, brand=brand
)
return api_instance
def _mock_august_authentication(token_text, token_timestamp, state):
def _mock_august_authentication(
token_text: str, token_timestamp: float, state: AuthenticationState
) -> Authentication:
authentication = MagicMock(name="yalexs.authentication")
type(authentication).state = PropertyMock(return_value=state)
type(authentication).access_token = PropertyMock(return_value=token_text)
@@ -282,13 +357,15 @@ def _mock_august_authentication(token_text, token_timestamp, state):
return authentication
def _mock_august_lock(lockid="mocklockid1", houseid="mockhouseid1"):
def _mock_august_lock(
lockid: str = "mocklockid1", houseid: str = "mockhouseid1"
) -> Lock:
return Lock(lockid, _mock_august_lock_data(lockid=lockid, houseid=houseid))
def _mock_august_doorbell(
deviceid="mockdeviceid1", houseid="mockhouseid1", brand=Brand.AUGUST
):
deviceid="mockdeviceid1", houseid="mockhouseid1", brand=Brand.YALE_AUGUST
) -> Doorbell:
return Doorbell(
deviceid,
_mock_august_doorbell_data(deviceid=deviceid, houseid=houseid, brand=brand),
@@ -296,8 +373,10 @@ def _mock_august_doorbell(
def _mock_august_doorbell_data(
deviceid="mockdeviceid1", houseid="mockhouseid1", brand=Brand.AUGUST
):
deviceid: str = "mockdeviceid1",
houseid: str = "mockhouseid1",
brand: Brand = Brand.YALE_AUGUST,
) -> dict[str, Any]:
return {
"_id": deviceid,
"DeviceID": deviceid,
@@ -317,7 +396,9 @@ def _mock_august_doorbell_data(
}
def _mock_august_lock_data(lockid="mocklockid1", houseid="mockhouseid1"):
def _mock_august_lock_data(
lockid: str = "mocklockid1", houseid: str = "mockhouseid1"
) -> dict[str, Any]:
return {
"_id": lockid,
"LockID": lockid,
@@ -366,12 +447,12 @@ async def _mock_lock_from_fixture(hass: HomeAssistant, path: str) -> LockDetail:
return LockDetail(json_dict)
async def _mock_doorbell_from_fixture(hass: HomeAssistant, path: str) -> DoorbellDetail:
async def _mock_doorbell_from_fixture(hass: HomeAssistant, path: str) -> LockDetail:
json_dict = await _load_json_fixture(hass, path)
return DoorbellDetail(json_dict)
async def _load_json_fixture(hass: HomeAssistant, path: str) -> Any:
async def _load_json_fixture(hass: HomeAssistant, path: str) -> dict[str, Any]:
fixture = await hass.async_add_executor_job(
load_fixture, os.path.join("august", path)
)
@@ -390,7 +471,9 @@ async def _mock_lock_with_unlatch(hass: HomeAssistant) -> LockDetail:
return await _mock_lock_from_fixture(hass, "get_lock.online_with_unlatch.json")
def _mock_lock_operation_activity(lock, action, offset):
def _mock_lock_operation_activity(
lock: Lock, action: str, offset: float
) -> LockOperationActivity:
return LockOperationActivity(
SOURCE_LOCK_OPERATE,
{
@@ -402,7 +485,9 @@ def _mock_lock_operation_activity(lock, action, offset):
)
def _mock_door_operation_activity(lock, action, offset):
def _mock_door_operation_activity(
lock: Lock, action: str, offset: float
) -> DoorOperationActivity:
return DoorOperationActivity(
SOURCE_LOCK_OPERATE,
{

2
tests/components/august/test_binary_sensor.py
View File

@@ -249,7 +249,7 @@ async def test_door_sense_update_via_pubnub(hass: HomeAssistant) -> None:
pubnub = AugustPubNub()
activities = await _mock_activities_from_fixture(hass, "get_activity.lock.json")
config_entry = await _create_august_with_devices(
config_entry, _ = await _create_august_with_devices(
hass, [lock_one], activities=activities, pubnub=pubnub
)
states = hass.states

2
tests/components/august/test_button.py
View File

@@ -12,7 +12,7 @@ async def test_wake_lock(hass: HomeAssistant) -> None:
lock_one = await _mock_lock_from_fixture(
hass, "get_lock.online_with_doorsense.json"
)
_, api_instance = await _create_august_api_with_devices(hass, [lock_one])
_, api_instance, _ = await _create_august_api_with_devices(hass, [lock_one])
entity_id = "button.online_with_doorsense_name_wake"
binary_sensor_online_with_doorsense_name = hass.states.get(entity_id)
assert binary_sensor_online_with_doorsense_name is not None

771
tests/components/august/test_config_flow.py
View File

@@ -1,399 +1,456 @@
"""Test the August config flow."""
from unittest.mock import patch
from collections.abc import Generator
from unittest.mock import ANY, Mock, patch
from yalexs.authenticator_common import ValidationResult
from yalexs.manager.exceptions import CannotConnect, InvalidAuth, RequireValidation
import pytest
from homeassistant.components.august.const import (
CONF_ACCESS_TOKEN_CACHE_FILE,
CONF_BRAND,
CONF_INSTALL_ID,
CONF_LOGIN_METHOD,
DOMAIN,
VERIFICATION_CODE_KEY,
from homeassistant.components.august.application_credentials import (
OAUTH2_AUTHORIZE,
OAUTH2_TOKEN,
)
from homeassistant.components.august.const import DOMAIN
from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import CONF_PASSWORD, CONF_TIMEOUT, CONF_USERNAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
CLIENT_ID = "1"
USER_ID = "a76c25e5-49aa-4c14-cd0c-48a6931e2081"
async def test_form(hass: HomeAssistant) -> None:
"""Test we get the form."""
@pytest.fixture
def mock_setup_entry() -> Generator[Mock]:
"""Patch setup entry."""
with patch(
"homeassistant.components.august.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
jwt: str,
) -> None:
"""Check full flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
return_value=True,
),
patch(
"homeassistant.components.august.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BRAND: "august",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
)
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"
aioclient_mock.clear_requests()
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": jwt,
"scope": "any",
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": "mock-user-id",
"expires_at": 1697753347,
},
)
result2 = await hass.config_entries.flow.async_configure(result["flow_id"])
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.unique_id == USER_ID
assert result2["type"] is FlowResultType.CREATE_ENTRY
assert result2["title"] == "my@email.tld"
assert result2["data"] == {
CONF_BRAND: "august",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_INSTALL_ID: None,
CONF_ACCESS_TOKEN_CACHE_FILE: ".my@email.tld.august.conf",
assert result2["result"].unique_id == USER_ID
assert entry.data == {
"auth_implementation": "august",
"token": {
"access_token": jwt,
"expires_at": ANY,
"expires_in": ANY,
"refresh_token": "mock-refresh-token",
"scope": "any",
"user_id": "mock-user-id",
},
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_invalid_auth(hass: HomeAssistant) -> None:
"""Test we handle invalid auth."""
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_full_flow_already_exists(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
jwt: str,
mock_config_entry: MockConfigEntry,
) -> None:
"""Check full flow for a user that already exists."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=InvalidAuth,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BRAND: "august",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "invalid_auth"}
async def test_user_unexpected_exception(hass: HomeAssistant) -> None:
"""Test we handle an unexpected exception."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=ValueError("something exploded"),
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BRAND: "august",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "unhandled"}
assert result2["description_placeholders"] == {"error": "something exploded"}
async def test_form_cannot_connect(hass: HomeAssistant) -> None:
"""Test we handle cannot connect error."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=CannotConnect,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] == {"base": "cannot_connect"}
async def test_form_needs_validate(hass: HomeAssistant) -> None:
"""Test we present validation when we need to validate."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
)
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=RequireValidation,
),
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_send_verification_code",
return_value=True,
) as mock_send_verification_code,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
assert len(mock_send_verification_code.mock_calls) == 1
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] is None
assert result2["step_id"] == "validation"
# Try with the WRONG verification code give us the form back again
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=RequireValidation,
),
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_validate_verification_code",
return_value=ValidationResult.INVALID_VERIFICATION_CODE,
) as mock_validate_verification_code,
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_send_verification_code",
return_value=True,
) as mock_send_verification_code,
):
result3 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{VERIFICATION_CODE_KEY: "incorrect"},
)
# Make sure we do not resend the code again
# so they have a chance to retry
assert len(mock_send_verification_code.mock_calls) == 0
assert len(mock_validate_verification_code.mock_calls) == 1
assert result3["type"] is FlowResultType.FORM
assert result3["errors"] == {"base": "invalid_verification_code"}
assert result3["step_id"] == "validation"
# Try with the CORRECT verification code and we setup
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
return_value=True,
),
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_validate_verification_code",
return_value=ValidationResult.VALIDATED,
) as mock_validate_verification_code,
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_send_verification_code",
return_value=True,
) as mock_send_verification_code,
patch(
"homeassistant.components.august.async_setup_entry", return_value=True
) as mock_setup_entry,
):
result4 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{VERIFICATION_CODE_KEY: "correct"},
)
await hass.async_block_till_done()
assert len(mock_send_verification_code.mock_calls) == 0
assert len(mock_validate_verification_code.mock_calls) == 1
assert result4["type"] is FlowResultType.CREATE_ENTRY
assert result4["title"] == "my@email.tld"
assert result4["data"] == {
CONF_BRAND: "august",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_INSTALL_ID: None,
CONF_ACCESS_TOKEN_CACHE_FILE: ".my@email.tld.august.conf",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_form_reauth(hass: HomeAssistant) -> None:
"""Test reauthenticate."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
CONF_INSTALL_ID: None,
CONF_TIMEOUT: 10,
CONF_ACCESS_TOKEN_CACHE_FILE: ".my@email.tld.august.conf",
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
unique_id="my@email.tld",
)
entry.add_to_hass(hass)
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
assert result["url"] == (
f"{OAUTH2_AUTHORIZE}?response_type=code&client_id={CLIENT_ID}"
"&redirect_uri=https://example.com/auth/external/callback"
f"&state={state}"
)
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
return_value=True,
),
patch(
"homeassistant.components.august.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "new-test-password",
},
)
await hass.async_block_till_done()
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"
aioclient_mock.clear_requests()
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": jwt,
"scope": "any",
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": "mock-user-id",
"expires_at": 1697753347,
},
)
result2 = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
assert result2["reason"] == "already_configured"
async def test_form_reauth_with_2fa(hass: HomeAssistant) -> None:
"""Test reauthenticate with 2fa."""
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reauth(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry: MockConfigEntry,
reauth_jwt: str,
) -> None:
"""Test the reauthentication case updates the existing config entry."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
CONF_INSTALL_ID: None,
CONF_TIMEOUT: 10,
CONF_ACCESS_TOKEN_CACHE_FILE: ".my@email.tld.august.conf",
mock_config_entry.add_to_hass(hass)
mock_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
unique_id="my@email.tld",
)
entry.add_to_hass(hass)
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"
result = await entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
side_effect=RequireValidation,
),
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_send_verification_code",
return_value=True,
) as mock_send_verification_code,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_PASSWORD: "new-test-password",
},
)
await hass.async_block_till_done()
assert len(mock_send_verification_code.mock_calls) == 1
assert result2["type"] is FlowResultType.FORM
assert result2["errors"] is None
assert result2["step_id"] == "validation"
# Try with the CORRECT verification code and we setup
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
return_value=True,
),
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_validate_verification_code",
return_value=ValidationResult.VALIDATED,
) as mock_validate_verification_code,
patch(
"yalexs.manager.gateway.AuthenticatorAsync.async_send_verification_code",
return_value=True,
) as mock_send_verification_code,
patch(
"homeassistant.components.august.async_setup_entry", return_value=True
) as mock_setup_entry,
):
result3 = await hass.config_entries.flow.async_configure(
result2["flow_id"],
{VERIFICATION_CODE_KEY: "correct"},
)
await hass.async_block_till_done()
assert len(mock_validate_verification_code.mock_calls) == 1
assert len(mock_send_verification_code.mock_calls) == 0
assert result3["type"] is FlowResultType.ABORT
assert result3["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
async def test_switching_brands(hass: HomeAssistant) -> None:
"""Test brands can be switched by setting up again."""
entry = MockConfigEntry(
domain=DOMAIN,
data={
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
CONF_INSTALL_ID: None,
CONF_TIMEOUT: 10,
CONF_ACCESS_TOKEN_CACHE_FILE: ".my@email.tld.august.conf",
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": reauth_jwt,
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": USER_ID,
"token_type": "Bearer",
"expires_at": 1697753347,
},
unique_id="my@email.tld",
)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_USER}
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
assert mock_config_entry.unique_id == USER_ID
assert "token" in mock_config_entry.data
# Verify access token is refreshed
assert mock_config_entry.data["token"]["access_token"] == reauth_jwt
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_reauth_wrong_account(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry: MockConfigEntry,
reauth_jwt_wrong_account: str,
jwt: str,
) -> None:
"""Test the reauthentication aborts, if user tries to reauthenticate with another account."""
assert mock_config_entry.data["token"]["access_token"] == jwt
mock_config_entry.add_to_hass(hass)
mock_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
assert result["type"] is FlowResultType.FORM
assert result["errors"] == {}
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"
with (
patch(
"homeassistant.components.august.config_flow.AugustGateway.async_authenticate",
return_value=True,
),
patch(
"homeassistant.components.august.async_setup_entry",
return_value=True,
) as mock_setup_entry,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
{
CONF_BRAND: "yale_access",
CONF_LOGIN_METHOD: "email",
CONF_USERNAME: "my@email.tld",
CONF_PASSWORD: "test-password",
},
)
await hass.async_block_till_done()
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": reauth_jwt_wrong_account,
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"token_type": "Bearer",
"expires_at": 1697753347,
},
)
assert result2["type"] is FlowResultType.ABORT
assert result2["reason"] == "reauth_successful"
assert len(mock_setup_entry.mock_calls) == 1
assert entry.data[CONF_BRAND] == "yale_access"
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_invalid_user"
assert mock_config_entry.unique_id == USER_ID
assert "token" in mock_config_entry.data
# Verify access token is like before
assert mock_config_entry.data["token"]["access_token"] == jwt
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_legacy_migration_with_email_match(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_legacy_config_entry: MockConfigEntry,
migration_jwt: str,
) -> None:
"""Test migration from legacy username/password config to OAuth with email validation."""
mock_legacy_config_entry.add_to_hass(hass)
# Start reauth flow (triggered by ConfigEntryAuthFailed in async_setup_entry)
mock_legacy_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
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"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": migration_jwt, # JWT with email: ["my@email.tld"]
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": USER_ID,
"token_type": "Bearer",
"expires_at": 1697753347,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
# Verify the entry was updated with new unique_id and OAuth data
assert mock_legacy_config_entry.unique_id == USER_ID # Updated from email to userId
assert "token" in mock_legacy_config_entry.data
assert mock_legacy_config_entry.data["auth_implementation"] == "august"
assert mock_legacy_config_entry.data["token"]["access_token"] == migration_jwt
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_legacy_migration_wrong_email(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_legacy_config_entry: MockConfigEntry,
reauth_jwt_wrong_account: str,
) -> None:
"""Test migration from legacy config fails when email doesn't match."""
mock_legacy_config_entry.add_to_hass(hass)
# Start reauth flow
mock_legacy_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
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"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": reauth_jwt_wrong_account, # JWT with email: ["different@email.tld"]
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"token_type": "Bearer",
"expires_at": 1697753347,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_invalid_user"
# Verify the entry was NOT updated
assert mock_legacy_config_entry.unique_id == "my@email.tld" # Still email
assert "token" not in mock_legacy_config_entry.data # Still legacy data
@pytest.mark.usefixtures("client_credentials")
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.usefixtures("mock_setup_entry")
async def test_legacy_migration_no_email_in_jwt(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_legacy_config_entry: MockConfigEntry,
jwt: str, # JWT with empty email array
) -> None:
"""Test migration from legacy config succeeds when JWT has no email (can't validate)."""
mock_legacy_config_entry.add_to_hass(hass)
# Start reauth flow
mock_legacy_config_entry.async_start_reauth(hass)
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
result = flows[0]
assert result["step_id"] == "auth"
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
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"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"access_token": jwt, # JWT with email: []
"expires_in": 86399,
"refresh_token": "mock-refresh-token",
"user_id": USER_ID,
"token_type": "Bearer",
"expires_at": 1697753347,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "reauth_successful"
# Verify the entry was updated (allowed because no email to validate)
assert mock_legacy_config_entry.unique_id == USER_ID # Updated from email to userId
assert "token" in mock_legacy_config_entry.data
assert mock_legacy_config_entry.data["auth_implementation"] == "august"
assert mock_legacy_config_entry.data["token"]["access_token"] == jwt

2
tests/components/august/test_diagnostics.py
View File

@@ -25,7 +25,7 @@ async def test_diagnostics(
)
doorbell_one = await _mock_doorbell_from_fixture(hass, "get_doorbell.json")
entry, _ = await _create_august_api_with_devices(hass, [lock_one, doorbell_one])
entry, _, _ = await _create_august_api_with_devices(hass, [lock_one, doorbell_one])
diag = await get_diagnostics_for_config_entry(hass, hass_client, entry)
assert diag == snapshot

54
tests/components/august/test_gateway.py
View File

@@ -1,54 +0,0 @@
"""The gateway tests for the august platform."""
from pathlib import Path
from unittest.mock import MagicMock, patch
from yalexs.authenticator_common import AuthenticationState
from homeassistant.components.august.const import DOMAIN
from homeassistant.components.august.gateway import AugustGateway
from homeassistant.core import HomeAssistant
from .mocks import _mock_august_authentication, _mock_get_config
async def test_refresh_access_token(hass: HomeAssistant) -> None:
"""Test token refreshes."""
await _patched_refresh_access_token(hass, "new_token", 5678)
@patch("yalexs.manager.gateway.ApiAsync.async_get_operable_locks")
@patch("yalexs.manager.gateway.AuthenticatorAsync.async_authenticate")
@patch("yalexs.manager.gateway.AuthenticatorAsync.should_refresh")
@patch("yalexs.manager.gateway.AuthenticatorAsync.async_refresh_access_token")
async def _patched_refresh_access_token(
hass: HomeAssistant,
new_token: str,
new_token_expire_time: int,
refresh_access_token_mock,
should_refresh_mock,
authenticate_mock,
async_get_operable_locks_mock,
) -> None:
authenticate_mock.side_effect = MagicMock(
return_value=_mock_august_authentication(
"original_token", 1234, AuthenticationState.AUTHENTICATED
)
)
august_gateway = AugustGateway(Path(hass.config.config_dir), MagicMock())
mocked_config = _mock_get_config()
await august_gateway.async_setup(mocked_config[DOMAIN])
await august_gateway.async_authenticate()
should_refresh_mock.return_value = False
await august_gateway.async_refresh_access_token_if_needed()
refresh_access_token_mock.assert_not_called()
should_refresh_mock.return_value = True
refresh_access_token_mock.return_value = _mock_august_authentication(
new_token, new_token_expire_time, AuthenticationState.AUTHENTICATED
)
await august_gateway.async_refresh_access_token_if_needed()
refresh_access_token_mock.assert_called()
assert await august_gateway.async_get_access_token() == new_token
assert august_gateway.authentication.access_token_expires == new_token_expire_time

263
tests/components/august/test_init.py
View File

@@ -1,12 +1,11 @@
"""The tests for the august platform."""
from unittest.mock import Mock, patch
from unittest.mock import Mock
from aiohttp import ClientResponseError
import pytest
from yalexs.authenticator_common import AuthenticationState
from yalexs.const import Brand
from yalexs.exceptions import AugustApiAIOHTTPError
from yalexs.exceptions import AugustApiAIOHTTPError, InvalidAuth
from homeassistant.components.august.const import DOMAIN
from homeassistant.components.lock import DOMAIN as LOCK_DOMAIN, LockState
@@ -29,10 +28,8 @@ from homeassistant.setup import async_setup_component
from .mocks import (
_create_august_with_devices,
_mock_august_authentication,
_mock_doorsense_enabled_august_lock_detail,
_mock_doorsense_missing_august_lock_detail,
_mock_get_config,
_mock_inoperative_august_lock_detail,
_mock_lock_with_offline_key,
_mock_operative_august_lock_detail,
@@ -44,68 +41,36 @@ from tests.typing import WebSocketGenerator
async def test_august_api_is_failing(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_RETRY when august api is failing."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
config_entry, _ = await _create_august_with_devices(
hass,
authenticate_side_effect=AugustApiAIOHTTPError(
"offline", ClientResponseError(None, None, status=500)
),
)
config_entry.add_to_hass(hass)
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
side_effect=ClientResponseError(None, None, status=500),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_august_is_offline(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_RETRY when august is offline."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
config_entry, _ = await _create_august_with_devices(
hass, authenticate_side_effect=TimeoutError
)
config_entry.add_to_hass(hass)
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
side_effect=TimeoutError,
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_august_late_auth_failure(hass: HomeAssistant) -> None:
"""Test we can detect a late auth failure."""
aiohttp_client_response_exception = ClientResponseError(None, None, status=401)
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
side_effect=AugustApiAIOHTTPError(
"This should bubble up as its user consumable",
aiohttp_client_response_exception,
config_entry, _ = await _create_august_with_devices(
hass,
authenticate_side_effect=InvalidAuth(
"authfailed", ClientResponseError(None, None, status=401)
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
)
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert flows[0]["step_id"] == "reauth_validate"
assert flows[0]["step_id"] == "pick_implementation"
async def test_unlock_throws_august_api_http_error(hass: HomeAssistant) -> None:
@@ -210,163 +175,12 @@ async def test_lock_has_doorsense(hass: HomeAssistant) -> None:
assert binary_sensor_missing_doorsense_id_name_open is None
async def test_auth_fails(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_ERROR when auth fails."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
side_effect=ClientResponseError(None, None, status=401),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert flows[0]["step_id"] == "reauth_validate"
async def test_bad_password(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_ERROR when the password has been changed."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
return_value=_mock_august_authentication(
"original_token", 1234, AuthenticationState.BAD_PASSWORD
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert flows[0]["step_id"] == "reauth_validate"
async def test_http_failure(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_RETRY when august is offline."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
side_effect=ClientResponseError(None, None, status=500),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
assert hass.config_entries.flow.async_progress() == []
async def test_unknown_auth_state(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_ERROR when august is in an unknown auth state."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
return_value=_mock_august_authentication("original_token", 1234, None),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert flows[0]["step_id"] == "reauth_validate"
async def test_requires_validation_state(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_ERROR when august requires validation."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
return_value=_mock_august_authentication(
"original_token", 1234, AuthenticationState.REQUIRES_VALIDATION
),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
assert len(hass.config_entries.flow.async_progress()) == 1
assert hass.config_entries.flow.async_progress()[0]["context"]["source"] == "reauth"
async def test_unknown_auth_http_401(hass: HomeAssistant) -> None:
"""Config entry state is SETUP_ERROR when august gets an http."""
config_entry = MockConfigEntry(
domain=DOMAIN,
data=_mock_get_config()[DOMAIN],
title="August august",
)
config_entry.add_to_hass(hass)
assert hass.config_entries.flow.async_progress() == []
with patch(
"yalexs.authenticator_async.AuthenticatorAsync.async_authenticate",
return_value=_mock_august_authentication("original_token", 1234, None),
):
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_ERROR
flows = hass.config_entries.flow.async_progress()
assert flows[0]["step_id"] == "reauth_validate"
async def test_load_unload(hass: HomeAssistant) -> None:
"""Config entry can be unloaded."""
august_operative_lock = await _mock_operative_august_lock_detail(hass)
august_inoperative_lock = await _mock_inoperative_august_lock_detail(hass)
config_entry = await _create_august_with_devices(
config_entry, _ = await _create_august_with_devices(
hass, [august_operative_lock, august_inoperative_lock]
)
@@ -385,7 +199,7 @@ async def test_load_triggers_ble_discovery(
august_lock_with_key = await _mock_lock_with_offline_key(hass)
august_lock_without_key = await _mock_operative_august_lock_detail(hass)
config_entry = await _create_august_with_devices(
config_entry, _ = await _create_august_with_devices(
hass, [august_lock_with_key, august_lock_without_key]
)
await hass.async_block_till_done()
@@ -410,7 +224,7 @@ async def test_device_remove_devices(
"""Test we can only remove a device that no longer exists."""
assert await async_setup_component(hass, "config", {})
august_operative_lock = await _mock_operative_august_lock_detail(hass)
config_entry = await _create_august_with_devices(hass, [august_operative_lock])
config_entry, _ = await _create_august_with_devices(hass, [august_operative_lock])
entity = entity_registry.entities["lock.a6697750d607098bae8d6baa11ef8063_name"]
device_entry = device_registry.async_get(entity.device_id)
@@ -427,21 +241,46 @@ async def test_device_remove_devices(
async def test_brand_migration_issue(hass: HomeAssistant) -> None:
"""Test creating and removing the brand migration issue."""
"""Test removing the brand migration issue."""
august_operative_lock = await _mock_operative_august_lock_detail(hass)
config_entry = await _create_august_with_devices(
config_entry, _ = await _create_august_with_devices(
hass, [august_operative_lock], brand=Brand.YALE_HOME
)
assert config_entry.state is ConfigEntryState.LOADED
issue_reg = ir.async_get(hass)
issue_entry = issue_reg.async_get_issue(DOMAIN, "yale_brand_migration")
assert issue_entry
assert issue_entry.severity == ir.IssueSeverity.CRITICAL
assert issue_entry.translation_placeholders == {
"migrate_url": "https://my.home-assistant.io/redirect/config_flow_start?domain=yale"
}
await hass.config_entries.async_remove(config_entry.entry_id)
assert not issue_reg.async_get_issue(DOMAIN, "yale_brand_migration")
async def test_oauth_migration_on_legacy_entry(hass: HomeAssistant) -> None:
"""Test that legacy config entry triggers OAuth migration."""
# Create a legacy config entry without auth_implementation
legacy_entry = MockConfigEntry(
domain=DOMAIN,
data={
"login_method": "email",
"username": "test@example.com",
"password": "test-password",
"install_id": None,
"timeout": 10,
"access_token_cache_file": ".test@example.com.august.conf",
},
unique_id="test@example.com",
)
legacy_entry.add_to_hass(hass)
# Try to setup the entry - should fail with auth error and trigger reauth
await hass.config_entries.async_setup(legacy_entry.entry_id)
await hass.async_block_till_done()
# Entry should be in setup_error state
assert legacy_entry.state is ConfigEntryState.SETUP_ERROR
# A reauth flow should be started
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
assert flows[0]["step_id"] == "pick_implementation"
assert flows[0]["context"]["source"] == "reauth"

2
tests/components/august/test_lock.py
View File

@@ -374,7 +374,7 @@ async def test_lock_update_via_pubnub(hass: HomeAssistant) -> None:
pubnub = AugustPubNub()
activities = await _mock_activities_from_fixture(hass, "get_activity.lock.json")
config_entry = await _create_august_with_devices(
config_entry, _ = await _create_august_with_devices(
hass, [lock_one], activities=activities, pubnub=pubnub
)
pubnub.connected = True