starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-28 18:24:50 -05:00
Code Issues Packages Projects Releases Wiki Activity
Files
rc
core/tests/components/cloud/test_http_api.py
T

3198 lines
101 KiB
Python
Raw Permalink Blame History

"""Tests for the HTTP API for the cloud component."""
from collections.abc import Callable, Coroutine
from copy import deepcopy
import datetime
from http import HTTPStatus
import logging
from operator import attrgetter
from typing import Any
from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch
import aiohttp
from freezegun.api import FrozenDateTimeFactory
from hass_nabucasa import (
AlreadyConnectedError,
AuthTimeoutError,
AutoLoginController,
LoginFailedEvent,
LoginFailedReason,
)
from hass_nabucasa.auth import (
InvalidTotpCode,
MFARequired,
Unauthenticated,
UnknownError,
)
from hass_nabucasa.const import STATE_CONNECTED
from hass_nabucasa.payments_api import PaymentsApiError
from hass_nabucasa.remote import CertificateStatus
import pytest
from syrupy.assertion import SnapshotAssertion
from webrtc_models import RTCIceServer
from homeassistant.components import system_health
from homeassistant.components.alexa import errors as alexa_errors
# pylint: disable-next=home-assistant-component-root-import
from homeassistant.components.alexa.entities import LightCapabilities
from homeassistant.components.assist_pipeline.pipeline import ( # pylint: disable=home-assistant-component-root-import
STORAGE_KEY,
)
from homeassistant.components.cloud.const import (
DATA_PENDING_AUTO_LOGIN,
DEFAULT_EXPOSED_DOMAINS,
DOMAIN,
)
from homeassistant.components.cloud.http_api import validate_language_voice
from homeassistant.components.frontend import DATA_THEMES
from homeassistant.components.google_assistant.helpers import ( # pylint: disable=home-assistant-component-root-import
GoogleEntity,
)
from homeassistant.components.homeassistant import exposed_entities
from homeassistant.components.websocket_api import ERR_INVALID_FORMAT
from homeassistant.core import HomeAssistant, State
from homeassistant.helpers import entity_registry as er
from homeassistant.loader import async_get_loaded_integration
from homeassistant.setup import async_setup_component
from homeassistant.util import dt as dt_util
from homeassistant.util.location import LocationInfo
from tests.common import mock_platform
from tests.components.google_assistant import MockConfig
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import (
ClientSessionGenerator,
MockHAClientWebSocket,
WebSocketGenerator,
)
PIPELINE_DATA_LEGACY = {
"items": [
{
"conversation_engine": "conversation.home_assistant",
"conversation_language": "language_1",
"id": "12345",
"language": "language_1",
"name": "Home Assistant Cloud",
"stt_engine": "cloud",
"stt_language": "language_1",
"tts_engine": "cloud",
"tts_language": "language_1",
"tts_voice": "Arnold Schwarzenegger",
"wake_word_entity": None,
"wake_word_id": None,
},
],
"preferred_item": "12345",
}
PIPELINE_DATA = {
"items": [
{
"conversation_engine": "conversation.home_assistant",
"conversation_language": "language_1",
"id": "12345",
"language": "language_1",
"name": "Home Assistant Cloud",
"stt_engine": "stt.home_assistant_cloud",
"stt_language": "language_1",
"tts_engine": "cloud",
"tts_language": "language_1",
"tts_voice": "Arnold Schwarzenegger",
"wake_word_entity": None,
"wake_word_id": None,
},
],
"preferred_item": "12345",
}
PIPELINE_DATA_OTHER = {
"items": [
{
"conversation_engine": "other",
"conversation_language": "language_1",
"id": "12345",
"language": "language_1",
"name": "Home Assistant",
"stt_engine": "stt.other",
"stt_language": "language_1",
"tts_engine": "other",
"tts_language": "language_1",
"tts_voice": "Arnold Schwarzenegger",
"wake_word_entity": None,
"wake_word_id": None,
},
],
"preferred_item": "12345",
}
SUBSCRIPTION_INFO_URL = "https://api-test.hass.io/payments/subscription_info"
@pytest.fixture(name="setup_cloud")
async def setup_cloud_fixture(hass: HomeAssistant, cloud: MagicMock) -> None:
"""Fixture that sets up cloud."""
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
"mode": "development",
"cognito_client_id": "cognito_client_id",
"user_pool_id": "user_pool_id",
"region": "region",
"relayer_server": "relayer",
"acme_server": "cert-server",
"api_server": "api-test.example.com",
"google_actions": {"filter": {"include_domains": "light"}},
"alexa": {
"filter": {"include_entities": ["light.kitchen", "switch.ac"]}
},
},
},
)
await hass.async_block_till_done()
await cloud.login("test-user", "test-pass")
cloud.login.reset_mock()
async def test_google_actions_sync(
setup_cloud: None,
hass_client: ClientSessionGenerator,
cloud: MagicMock,
) -> None:
"""Test syncing Google Actions."""
cloud_client = await hass_client()
cloud.google_report_state.request_sync = AsyncMock(
return_value=Mock(status=HTTPStatus.OK)
)
req = await cloud_client.post("/api/cloud/google_actions/sync")
assert req.status == HTTPStatus.OK
assert len(cloud.google_report_state.request_sync.mock_calls) == 1
async def test_google_actions_sync_fails(
setup_cloud: None,
hass_client: ClientSessionGenerator,
cloud: MagicMock,
) -> None:
"""Test syncing Google Actions gone bad."""
cloud_client = await hass_client()
cloud.google_report_state.request_sync = AsyncMock(
return_value=Mock(status=HTTPStatus.INTERNAL_SERVER_ERROR)
)
req = await cloud_client.post("/api/cloud/google_actions/sync")
assert req.status == HTTPStatus.INTERNAL_SERVER_ERROR
assert len(cloud.google_report_state.request_sync.mock_calls) == 1
@pytest.mark.parametrize(
"entity_id", ["stt.home_assistant_cloud", "tts.home_assistant_cloud"]
)
@pytest.mark.usefixtures("setup_cloud")
async def test_login_view_missing_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_client: ClientSessionGenerator,
entity_id: str,
) -> None:
"""Test logging in when a cloud assist pipeline needed entity is missing."""
assert await async_setup_component(hass, "assist_pipeline", {})
# Make sure that the cloud entity does not exist.
entity_registry.async_remove(entity_id)
await hass.async_block_till_done()
cloud_client = await hass_client()
# We assume the user needs to login again for some reason.
with patch(
"homeassistant.components.cloud.assist_pipeline.async_create_default_pipeline",
) as create_pipeline_mock:
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.OK
assert await req.json() == {"success": True}
create_pipeline_mock.assert_not_awaited()
@pytest.mark.parametrize("pipeline_data", [PIPELINE_DATA, PIPELINE_DATA_LEGACY])
@pytest.mark.usefixtures("cloud")
async def test_login_view_existing_pipeline(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_storage: dict[str, Any],
pipeline_data: dict[str, Any],
) -> None:
"""Test logging in when an assist pipeline is available."""
hass_storage[STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": STORAGE_KEY,
"data": deepcopy(pipeline_data),
}
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(hass, "assist_pipeline", {})
assert await async_setup_component(hass, DOMAIN, {"cloud": {}})
await hass.async_block_till_done()
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.assist_pipeline.async_create_default_pipeline",
) as create_pipeline_mock:
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.OK
assert await req.json() == {"success": True}
create_pipeline_mock.assert_not_awaited()
@pytest.mark.usefixtures("cloud")
async def test_login_view_create_pipeline(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_storage: dict[str, Any],
) -> None:
"""Test logging in when no existing cloud assist pipeline is available."""
hass_storage[STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": STORAGE_KEY,
"data": deepcopy(PIPELINE_DATA_OTHER),
}
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(hass, "assist_pipeline", {})
assert await async_setup_component(hass, DOMAIN, {"cloud": {}})
await hass.async_block_till_done()
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.assist_pipeline.async_create_default_pipeline",
return_value=AsyncMock(id="12345"),
) as create_pipeline_mock:
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.OK
assert await req.json() == {"success": True}
create_pipeline_mock.assert_awaited_once_with(
hass,
stt_engine_id="stt.home_assistant_cloud",
tts_engine_id="tts.home_assistant_cloud",
pipeline_name="Home Assistant Cloud",
)
async def test_auto_login_creates_pipeline(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_storage: dict[str, Any],
) -> None:
"""Test the auto-login creates the cloud assist pipeline too."""
hass_storage[STORAGE_KEY] = {
"version": 1,
"minor_version": 1,
"key": STORAGE_KEY,
"data": deepcopy(PIPELINE_DATA_OTHER),
}
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(hass, "assist_pipeline", {})
assert await async_setup_component(hass, DOMAIN, {"cloud": {}})
await hass.async_block_till_done()
await register_auto_login(hass_client)
with patch(
"homeassistant.components.cloud.assist_pipeline.async_create_default_pipeline",
return_value=AsyncMock(id="12345"),
) as create_pipeline_mock:
# Simulate the library retry task logging in.
await cloud.login("hello@bla.com", "falcon42", auto=True)
create_pipeline_mock.assert_awaited_once_with(
hass,
stt_engine_id="stt.home_assistant_cloud",
tts_engine_id="tts.home_assistant_cloud",
pipeline_name="Home Assistant Cloud",
)
async def test_login_view_random_exception(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Try logging in with random exception."""
cloud_client = await hass_client()
cloud.login.side_effect = ValueError("Boom")
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
resp = await req.json()
assert resp == {"code": "valueerror", "message": "Unexpected error: Boom"}
async def test_login_view_invalid_json(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Try logging in with invalid JSON."""
cloud_client = await hass_client()
mock_login = cloud.login
req = await cloud_client.post("/api/cloud/login", data="Not JSON")
assert req.status == HTTPStatus.BAD_REQUEST
assert mock_login.call_count == 0
async def test_login_view_invalid_schema(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Try logging in with invalid schema."""
cloud_client = await hass_client()
mock_login = cloud.login
req = await cloud_client.post("/api/cloud/login", json={"invalid": "schema"})
assert req.status == HTTPStatus.BAD_REQUEST
assert mock_login.call_count == 0
async def test_login_view_request_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test request timeout while trying to log in."""
cloud_client = await hass_client()
cloud.login.side_effect = TimeoutError
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert cloud.login.call_args[1]["check_connection"] is False
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_login_view_request_auth_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test authentication timeout while trying to log in."""
cloud_client = await hass_client()
cloud.login.side_effect = AuthTimeoutError
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert cloud.login.call_args[1]["check_connection"] is False
assert req.status == HTTPStatus.GATEWAY_TIMEOUT
async def test_login_view_with_already_existing_connection(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test request timeout while trying to log in."""
cloud_client = await hass_client()
cloud.login.side_effect = AlreadyConnectedError(
details={"remote_ip_address": "127.0.0.1", "connected_at": "1"}
)
req = await cloud_client.post(
"/api/cloud/login",
json={
"email": "my_username",
"password": "my_password",
"check_connection": True,
},
)
assert cloud.login.call_args[1]["check_connection"] is True
assert req.status == HTTPStatus.CONFLICT
resp = await req.json()
assert resp == {
"code": "alreadyconnectederror",
"message": '{"remote_ip_address": "127.0.0.1", "connected_at": "1"}',
}
async def test_login_view_invalid_credentials(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging in with invalid credentials."""
cloud_client = await hass_client()
cloud.login.side_effect = Unauthenticated
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.UNAUTHORIZED
async def test_login_view_mfa_required(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging in when MFA is required."""
cloud_client = await hass_client()
cloud.login.side_effect = MFARequired(mfa_tokens={"session": "tokens"})
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.UNAUTHORIZED
res = await req.json()
assert res["code"] == "mfarequired"
async def test_login_view_mfa_required_tokens_missing(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test MFA login when code is given but tokens are missing."""
cloud_client = await hass_client()
cloud.login.side_effect = MFARequired(mfa_tokens={})
# Login with password and get MFA required error
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.UNAUTHORIZED
res = await req.json()
assert res["code"] == "mfarequired"
# Login with TOTP code and get MFA expired error
req = await cloud_client.post(
"/api/cloud/login",
json={"email": "my_username", "code": "123346"},
)
assert req.status == HTTPStatus.BAD_REQUEST
res = await req.json()
assert res["code"] == "mfaexpiredornotstarted"
async def test_login_view_mfa_password_and_totp_provided(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging in when password and TOTP code provided at once."""
cloud_client = await hass_client()
req = await cloud_client.post(
"/api/cloud/login",
json={"email": "my_username", "password": "my_password", "code": "123346"},
)
assert req.status == HTTPStatus.BAD_REQUEST
async def test_login_view_invalid_totp_code(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging in when MFA is required and invalid code is provided."""
cloud_client = await hass_client()
cloud.login.side_effect = MFARequired(mfa_tokens={"session": "tokens"})
cloud.login_verify_totp.side_effect = InvalidTotpCode
# Login with password and get MFA required error
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.UNAUTHORIZED
res = await req.json()
assert res["code"] == "mfarequired"
# Login with TOTP code and get invalid TOTP code error
req = await cloud_client.post(
"/api/cloud/login",
json={"email": "my_username", "code": "123346"},
)
assert req.status == HTTPStatus.BAD_REQUEST
res = await req.json()
assert res["code"] == "invalidtotpcode"
async def test_login_view_valid_totp_provided(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging in with valid TOTP code."""
cloud_client = await hass_client()
cloud.login.side_effect = MFARequired(mfa_tokens={"session": "tokens"})
# Login with password and get MFA required error
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.UNAUTHORIZED
res = await req.json()
assert res["code"] == "mfarequired"
# Login with TOTP code and get success response
req = await cloud_client.post(
"/api/cloud/login",
json={"email": "my_username", "code": "123346"},
)
assert req.status == HTTPStatus.OK
result = await req.json()
assert result == {"success": True}
async def test_login_view_unknown_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test unknown error while logging in."""
cloud_client = await hass_client()
cloud.login.side_effect = UnknownError
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_logout_view(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test logging out."""
cloud_client = await hass_client()
req = await cloud_client.post("/api/cloud/logout")
assert req.status == HTTPStatus.OK
data = await req.json()
assert data == {"message": "ok"}
assert cloud.logout.call_count == 1
async def test_logout_view_request_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test timeout while logging out."""
cloud_client = await hass_client()
cloud.logout.side_effect = TimeoutError
req = await cloud_client.post("/api/cloud/logout")
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_logout_view_unknown_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test unknown error while logging out."""
cloud_client = await hass_client()
cloud.logout.side_effect = UnknownError
req = await cloud_client.post("/api/cloud/logout")
assert req.status == HTTPStatus.BAD_GATEWAY
REGISTER_VIEW_PARAMS = [
pytest.param(
"/api/cloud/register",
"auth.async_register",
"register_and_auto_login",
id="register",
),
pytest.param(
"/api/cloud/register_auto_login",
"register_and_auto_login",
"auth.async_register",
id="register_auto_login",
),
]
@pytest.mark.parametrize(
("endpoint", "register_method", "unused_register_method"), REGISTER_VIEW_PARAMS
)
async def test_register_view_no_location(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
endpoint: str,
register_method: str,
unused_register_method: str,
) -> None:
"""Test register without location."""
cloud.id_token = None
cloud_client = await hass_client()
register_mock = attrgetter(register_method)(cloud)
with patch(
"homeassistant.components.cloud.http_api.async_detect_location_info",
return_value=None,
):
req = await cloud_client.post(
endpoint,
json={"email": "hello@bla.com", "password": "falcon42"},
)
assert req.status == HTTPStatus.OK
assert register_mock.call_count == 1
call = register_mock.mock_calls[0]
result_email, result_pass = call.args
assert result_email == "hello@bla.com"
assert result_pass == "falcon42"
assert call.kwargs["client_metadata"] is None
assert attrgetter(unused_register_method)(cloud).call_count == 0
@pytest.mark.parametrize(
("endpoint", "register_method", "unused_register_method"), REGISTER_VIEW_PARAMS
)
async def test_register_view_with_location(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
endpoint: str,
register_method: str,
unused_register_method: str,
) -> None:
"""Test register with location."""
cloud.id_token = None
cloud_client = await hass_client()
register_mock = attrgetter(register_method)(cloud)
with patch(
"homeassistant.components.cloud.http_api.async_detect_location_info",
return_value=LocationInfo(
country_code="XX",
zip_code="12345",
region_code="GH",
ip="1.2.3.4",
city="Gotham",
region_name="Gotham",
time_zone="Earth/Gotham",
currency="XXX",
latitude="12.34567",
longitude="12.34567",
use_metric=True,
),
):
req = await cloud_client.post(
endpoint,
json={"email": "hello@bla.com", "password": "falcon42"},
)
assert req.status == HTTPStatus.OK
assert register_mock.call_count == 1
call = register_mock.mock_calls[0]
result_email, result_pass = call.args
assert result_email == "hello@bla.com"
assert result_pass == "falcon42"
assert call.kwargs["client_metadata"] == {
"NC_COUNTRY_CODE": "XX",
"NC_REGION_CODE": "GH",
"NC_ZIP_CODE": "12345",
}
assert attrgetter(unused_register_method)(cloud).call_count == 0
@pytest.mark.parametrize(
("endpoint", "register_method"),
[params.values[:2] for params in REGISTER_VIEW_PARAMS],
ids=[params.id for params in REGISTER_VIEW_PARAMS],
)
async def test_register_view_bad_data(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
endpoint: str,
register_method: str,
) -> None:
"""Test register bad data."""
cloud_client = await hass_client()
req = await cloud_client.post(
endpoint, json={"email": "hello@bla.com", "not_password": "falcon"}
)
assert req.status == HTTPStatus.BAD_REQUEST
assert attrgetter(register_method)(cloud).call_count == 0
@pytest.mark.parametrize(
("endpoint", "register_method"),
[params.values[:2] for params in REGISTER_VIEW_PARAMS],
ids=[params.id for params in REGISTER_VIEW_PARAMS],
)
async def test_register_view_request_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
endpoint: str,
register_method: str,
) -> None:
"""Test timeout while registering."""
cloud.id_token = None
cloud_client = await hass_client()
attrgetter(register_method)(cloud).side_effect = TimeoutError
req = await cloud_client.post(
endpoint, json={"email": "hello@bla.com", "password": "falcon42"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
@pytest.mark.parametrize(
("endpoint", "register_method"),
[params.values[:2] for params in REGISTER_VIEW_PARAMS],
ids=[params.id for params in REGISTER_VIEW_PARAMS],
)
async def test_register_view_unknown_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
endpoint: str,
register_method: str,
) -> None:
"""Test unknown error while registering."""
cloud.id_token = None
cloud_client = await hass_client()
attrgetter(register_method)(cloud).side_effect = UnknownError
req = await cloud_client.post(
endpoint, json={"email": "hello@bla.com", "password": "falcon42"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def register_auto_login(
hass_client: ClientSessionGenerator, email: str = "hello@bla.com"
) -> None:
"""Register with auto-login via the HTTP API."""
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.http_api.async_detect_location_info",
return_value=None,
):
req = await cloud_client.post(
"/api/cloud/register_auto_login",
json={"email": email, "password": "falcon42"},
)
assert req.status == HTTPStatus.OK
def pending_controller(hass: HomeAssistant) -> AutoLoginController:
"""Return the controller core is holding for the pending registration."""
controller = hass.data[DATA_PENDING_AUTO_LOGIN]
assert controller is not None
return controller
async def fail_auto_login(
hass: HomeAssistant, cloud: MagicMock, reason: LoginFailedReason
) -> None:
"""Give up on the pending auto-login the way hass_nabucasa does."""
controller = pending_controller(hass)
controller.failed_reason = reason
controller.active = False
await cloud.events.publish(LoginFailedEvent(auto=True, reason=reason))
async def get_cloud_status(
client: MockHAClientWebSocket, msg_id: int
) -> dict[str, Any]:
"""Fetch the cloud status over the websocket."""
await client.send_json({"id": msg_id, "type": "cloud/status"})
response = await client.receive_json()
return response["result"]
@pytest.mark.usefixtures("setup_cloud")
async def test_register_auto_login_status(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a pending auto-login is reported while it waits for a confirmation."""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] == {"email": "hello@bla.com", "failed": None}
@pytest.mark.usefixtures("setup_cloud")
async def test_register_auto_login_reports_normalized_email(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the reported email matches the one hass_nabucasa registers with."""
cloud.id_token = None
await register_auto_login(hass_client, email="Hello@BLA.com")
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"]["email"] == "hello@bla.com"
@pytest.mark.usefixtures("setup_cloud")
async def test_register_auto_login_already_given_up(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a retry loop that gave up before the view stored its controller.
The controller is the single source of truth, so the reason survives even though
the LOGIN_FAILED event fired before there was anything to store.
"""
cloud.id_token = None
def give_up_immediately(*args: Any, **kwargs: Any) -> AutoLoginController:
"""Return a controller whose retry loop has already given up."""
return AutoLoginController(
email="hello@bla.com",
cancel=MagicMock(),
attempt_now=MagicMock(),
resend=AsyncMock(),
active=False,
failed_reason=LoginFailedReason.TIMEOUT,
)
cloud.register_and_auto_login.side_effect = give_up_immediately
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] == {
"email": "hello@bla.com",
"failed": "auto_login_failed_timeout",
}
await client.send_json({"id": 6, "type": "cloud/attempt_auto_login_now"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["translation_key"] == "no_pending_auto_login"
@pytest.mark.usefixtures("setup_cloud")
async def test_register_auto_login_replaces_pending(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a second registration takes over from the first."""
cloud.id_token = None
await register_auto_login(hass_client)
first = pending_controller(hass)
await register_auto_login(hass_client, email="second@bla.com")
second = pending_controller(hass)
assert second is not first
assert first.email == "hello@bla.com"
assert second.email == "second@bla.com"
# hass_nabucasa cancels the superseded retry loop itself.
assert first.cancel.call_count == 0
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] == {"email": "second@bla.com", "failed": None}
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_hidden_while_logged_in(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a leftover pending registration is not reported once signed in."""
id_token = cloud.id_token
cloud.id_token = None
await register_auto_login(hass_client)
cloud.id_token = id_token
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["logged_in"] is True
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_register_auto_login_while_logged_in(
cloud: MagicMock,
hass_client: ClientSessionGenerator,
) -> None:
"""Test registering with auto-login is refused while already logged in."""
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.http_api.async_detect_location_info",
return_value=None,
):
req = await cloud_client.post(
"/api/cloud/register_auto_login",
json={"email": "hello@bla.com", "password": "falcon42"},
)
assert req.status == HTTPStatus.BAD_REQUEST
cloud.auth.async_register.assert_not_called()
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_hidden_from_non_admin(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
hass_read_only_access_token: str,
) -> None:
"""Test the pending registration email is not exposed to non-admin users."""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass, hass_read_only_access_token)
status = await get_cloud_status(client, 5)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_cancel_auto_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test cancelling a pending auto-login."""
cloud.id_token = None
await register_auto_login(hass_client)
controller = pending_controller(hass)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/cancel_auto_login"})
response = await client.receive_json()
assert response["success"]
assert controller.cancel.call_count == 1
status = await get_cloud_status(client, 6)
assert status["auto_login"] is None
@pytest.mark.parametrize(
"command",
[
pytest.param("cloud/cancel_auto_login", id="cancel"),
pytest.param("cloud/remove_data", id="remove_data"),
],
)
@pytest.mark.usefixtures("setup_cloud")
async def test_clearing_auto_login_pushes_event(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
command: str,
) -> None:
"""Test other subscribers are told a pending auto-login is gone."""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscribe_events"})
response = await client.receive_json()
assert response["success"]
await client.send_json({"id": 6, "type": command})
messages = [await client.receive_json(), await client.receive_json()]
event = next(message for message in messages if message["id"] == 5)
assert event["event"] == {"type": "auto_login_cancelled"}
@pytest.mark.parametrize(
("command", "controller_method"),
[
pytest.param("cloud/cancel_auto_login", "cancel", id="cancel"),
pytest.param("cloud/attempt_auto_login_now", "attempt_now", id="attempt_now"),
pytest.param("cloud/resend_auto_login_confirm", "resend", id="resend"),
],
)
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_command_without_pending(
hass: HomeAssistant,
cloud: MagicMock,
hass_ws_client: WebSocketGenerator,
command: str,
controller_method: str,
) -> None:
"""Test the auto-login commands report when there is nothing pending."""
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": command})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "not_found"
# The frontend renders the message from the translation key, not the text.
assert response["error"]["translation_domain"] == DOMAIN
assert response["error"]["translation_key"] == "no_pending_auto_login"
assert hass.data[DATA_PENDING_AUTO_LOGIN] is None
cloud.register_and_auto_login.assert_not_called()
@pytest.mark.parametrize(
("reason", "translation_key"),
[
pytest.param(
LoginFailedReason.TIMEOUT, "auto_login_failed_timeout", id="timeout"
),
pytest.param(
LoginFailedReason.CLOUD_ERROR,
"auto_login_failed_cloud_error",
id="cloud_error",
),
pytest.param(
LoginFailedReason.UNEXPECTED_ERROR,
"auto_login_failed_unexpected_error",
id="unexpected_error",
),
],
)
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_failure_pushed(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
reason: LoginFailedReason,
translation_key: str,
) -> None:
"""Test hass_nabucasa giving up on an auto-login reaches the frontend."""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscribe_events"})
response = await client.receive_json()
assert response["success"]
await fail_auto_login(hass, cloud, reason)
event = await client.receive_json()
assert event["id"] == 5
assert event["event"] == {
"type": "auto_login_failed",
"translation_key": translation_key,
}
# The key is kept, so a client that missed the push can still read it.
status = await get_cloud_status(client, 6)
assert status["auto_login"] == {
"email": "hello@bla.com",
"failed": translation_key,
}
@pytest.mark.parametrize(
"command",
[
pytest.param("cloud/attempt_auto_login_now", id="attempt_now"),
pytest.param("cloud/resend_auto_login_confirm", id="resend"),
],
)
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_command_after_failure(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
command: str,
) -> None:
"""Test the retry commands are refused once the retry loop gave up."""
cloud.id_token = None
await register_auto_login(hass_client)
await fail_auto_login(hass, cloud, LoginFailedReason.TIMEOUT)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": command})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "not_found"
assert response["error"]["translation_key"] == "no_pending_auto_login"
@pytest.mark.usefixtures("setup_cloud")
async def test_cancel_auto_login_after_failure(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the failed registration can be dismissed."""
cloud.id_token = None
await register_auto_login(hass_client)
await fail_auto_login(hass, cloud, LoginFailedReason.TIMEOUT)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/cancel_auto_login"})
response = await client.receive_json()
assert response["success"]
status = await get_cloud_status(client, 6)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_interactive_login_failure_not_pushed(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a failed interactive login leaves a pending auto-login alone.
The login view answers that caller itself, so there is nothing to push.
"""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
await cloud.events.publish(LoginFailedEvent(reason=LoginFailedReason.CLOUD_ERROR))
status = await get_cloud_status(client, 5)
assert status["auto_login"] == {"email": "hello@bla.com", "failed": None}
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_cleared_on_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a pending auto-login is cleared when the login succeeds."""
cloud.id_token = None
await register_auto_login(hass_client)
with patch(
"homeassistant.components.cloud.async_dispatcher_send"
) as async_dispatcher_send_mock:
# Simulate the library retry task logging in.
await cloud.login("hello@bla.com", "falcon42")
assert async_dispatcher_send_mock.call_count == 1
assert async_dispatcher_send_mock.mock_calls[0][1][1] == "cloud_event"
assert async_dispatcher_send_mock.mock_calls[0][1][2] == {"type": "login"}
cloud.id_token = None
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_login_view_while_auto_login_pending(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test logging in interactively while an auto-login is pending."""
cloud.id_token = None
await register_auto_login(hass_client)
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.async_dispatcher_send"
) as async_dispatcher_send_mock:
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.OK
assert await req.json() == {"success": True}
assert async_dispatcher_send_mock.call_count == 1
cloud.id_token = None
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_login_dispatches_event_once(
hass: HomeAssistant,
cloud: MagicMock,
) -> None:
"""Test a login outside the login view dispatches exactly one event."""
with patch(
"homeassistant.components.cloud.async_dispatcher_send"
) as async_dispatcher_send_mock:
await cloud.login("hello@bla.com", "falcon42")
assert async_dispatcher_send_mock.call_count == 1
assert async_dispatcher_send_mock.mock_calls[0][1][2] == {"type": "login"}
@pytest.mark.usefixtures("setup_cloud")
async def test_logout_clears_auto_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test logging out clears a pending auto-login."""
cloud.id_token = None
await register_auto_login(hass_client)
controller = pending_controller(hass)
cloud_client = await hass_client()
req = await cloud_client.post("/api/cloud/logout")
assert req.status == HTTPStatus.OK
# The library cancels the auto-login task on logout itself.
assert controller.cancel.call_count == 0
client = await hass_ws_client(hass)
status = await get_cloud_status(client, 5)
assert status["auto_login"] is None
@pytest.mark.parametrize(
("command", "controller_method"),
[
pytest.param("cloud/attempt_auto_login_now", "attempt_now", id="attempt_now"),
pytest.param("cloud/resend_auto_login_confirm", "resend", id="resend"),
],
)
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_command_keeps_pending(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
command: str,
controller_method: str,
) -> None:
"""Test forcing an attempt or resending the mail leaves the retry loop alone."""
cloud.id_token = None
await register_auto_login(hass_client)
controller = pending_controller(hass)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": command})
response = await client.receive_json()
assert response["success"]
assert getattr(controller, controller_method).call_count == 1
assert controller.cancel.call_count == 0
status = await get_cloud_status(client, 6)
assert status["auto_login"] == {"email": "hello@bla.com", "failed": None}
@pytest.mark.usefixtures("setup_cloud")
async def test_resend_auto_login_confirm_error(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a failing resend of the confirmation email."""
cloud.id_token = None
await register_auto_login(hass_client)
pending_controller(hass).resend.side_effect = UnknownError
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/resend_auto_login_confirm"})
response = await client.receive_json()
assert not response["success"]
# The registration stays pending, so the user can try again.
status = await get_cloud_status(client, 6)
assert status["auto_login"]["email"] == "hello@bla.com"
@pytest.mark.usefixtures("setup_cloud")
async def test_auto_login_failed_pushed_without_pending(
hass: HomeAssistant,
cloud: MagicMock,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the give-up push does not depend on core still tracking a registration."""
cloud.id_token = None
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscribe_events"})
assert (await client.receive_json())["success"]
await cloud.events.publish(
LoginFailedEvent(auto=True, reason=LoginFailedReason.TIMEOUT)
)
event = await client.receive_json()
assert event["id"] == 5
assert event["event"] == {
"type": "auto_login_failed",
"translation_key": "auto_login_failed_timeout",
}
status = await get_cloud_status(client, 6)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_resend_auto_login_confirm_timeout(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a resend that never comes back is reported as a gateway timeout."""
cloud.id_token = None
await register_auto_login(hass_client)
pending_controller(hass).resend.side_effect = TimeoutError
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/resend_auto_login_confirm"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == str(HTTPStatus.BAD_GATEWAY)
status = await get_cloud_status(client, 6)
assert status["auto_login"]["email"] == "hello@bla.com"
@pytest.mark.usefixtures("setup_cloud")
async def test_remove_data_cancels_auto_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test removing the cloud data cancels a pending auto-login."""
cloud.id_token = None
await register_auto_login(hass_client)
controller = pending_controller(hass)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/remove_data"})
response = await client.receive_json()
assert response["success"]
assert controller.cancel.call_count == 1
status = await get_cloud_status(client, 6)
assert status["auto_login"] is None
@pytest.mark.usefixtures("setup_cloud")
async def test_remove_data_logged_in_keeps_auto_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the logged-in refusal returns before anything is cancelled."""
id_token = cloud.id_token
cloud.id_token = None
await register_auto_login(hass_client)
controller = pending_controller(hass)
cloud.id_token = id_token
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/remove_data"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "logged_in"
cloud.remove_data.assert_not_called()
assert controller.cancel.call_count == 0
@pytest.mark.usefixtures("setup_cloud")
async def test_remove_data_failure_still_cancels_auto_login(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test the auto-login is cancelled up front, even if the removal then fails."""
cloud.id_token = None
cloud.remove_data.side_effect = ValueError("Cloud not stopped")
await register_auto_login(hass_client)
controller = pending_controller(hass)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/remove_data"})
response = await client.receive_json()
assert not response["success"]
# Cancelled before the wipe starts, so no login can land on erased data.
assert controller.cancel.call_count == 1
status = await get_cloud_status(client, 6)
assert status["auto_login"] is None
async def test_forgot_password_view(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test forgot password."""
cloud_client = await hass_client()
mock_cognito = cloud.auth
req = await cloud_client.post(
"/api/cloud/forgot_password", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.OK
assert mock_cognito.async_forgot_password.call_count == 1
async def test_forgot_password_view_bad_data(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test forgot password bad data."""
cloud_client = await hass_client()
mock_cognito = cloud.auth
req = await cloud_client.post(
"/api/cloud/forgot_password", json={"not_email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_REQUEST
assert mock_cognito.async_forgot_password.call_count == 0
async def test_forgot_password_view_request_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test timeout while forgot password."""
cloud_client = await hass_client()
cloud.auth.async_forgot_password.side_effect = TimeoutError
req = await cloud_client.post(
"/api/cloud/forgot_password", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_forgot_password_view_unknown_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test unknown error while forgot password."""
cloud_client = await hass_client()
cloud.auth.async_forgot_password.side_effect = UnknownError
req = await cloud_client.post(
"/api/cloud/forgot_password", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_forgot_password_view_aiohttp_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test unknown error while forgot password."""
cloud_client = await hass_client()
cloud.auth.async_forgot_password.side_effect = aiohttp.ClientResponseError(
Mock(), Mock()
)
req = await cloud_client.post(
"/api/cloud/forgot_password", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.INTERNAL_SERVER_ERROR
async def test_resend_confirm_view(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test resend confirm."""
cloud_client = await hass_client()
mock_cognito = cloud.auth
req = await cloud_client.post(
"/api/cloud/resend_confirm", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.OK
assert mock_cognito.async_resend_email_confirm.call_count == 1
async def test_resend_confirm_view_bad_data(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test resend confirm bad data."""
cloud_client = await hass_client()
mock_cognito = cloud.auth
req = await cloud_client.post(
"/api/cloud/resend_confirm", json={"not_email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_REQUEST
assert mock_cognito.async_resend_email_confirm.call_count == 0
async def test_resend_confirm_view_request_timeout(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test timeout while resend confirm."""
cloud_client = await hass_client()
cloud.auth.async_resend_email_confirm.side_effect = TimeoutError
req = await cloud_client.post(
"/api/cloud/resend_confirm", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_resend_confirm_view_unknown_error(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test unknown error while resend confirm."""
cloud_client = await hass_client()
cloud.auth.async_resend_email_confirm.side_effect = UnknownError
req = await cloud_client.post(
"/api/cloud/resend_confirm", json={"email": "hello@bla.com"}
)
assert req.status == HTTPStatus.BAD_GATEWAY
async def test_websocket_remove_data(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test removing cloud data."""
cloud.id_token = None
client = await hass_ws_client(hass)
with patch.object(cloud.client.prefs, "async_erase_config") as mock_erase_config:
await client.send_json_auto_id({"type": "cloud/remove_data"})
response = await client.receive_json()
assert response["success"]
cloud.remove_data.assert_awaited_once_with()
mock_erase_config.assert_awaited_once_with()
async def test_websocket_remove_data_logged_in(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test removing cloud data."""
cloud.iot.state = STATE_CONNECTED
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "cloud/remove_data"})
response = await client.receive_json()
assert not response["success"]
assert response["error"] == {
"code": "logged_in",
"message": "Can't remove data when logged in.",
}
async def test_websocket_status(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test querying the status."""
cloud.iot.state = STATE_CONNECTED
client = await hass_ws_client(hass)
with (
patch.dict(
"homeassistant.components.google_assistant.const.DOMAIN_TO_GOOGLE_TYPES",
{"light": None},
clear=True,
),
patch.dict(
"homeassistant.components.alexa.entities.ENTITY_ADAPTERS",
{"switch": None},
clear=True,
),
):
await client.send_json({"id": 5, "type": "cloud/status"})
response = await client.receive_json()
assert response["result"] == {
"auto_login": None,
"logged_in": True,
"email": "hello@home-assistant.io",
"cloud": "connected",
"cloud_last_disconnect_reason": None,
"prefs": {
"alexa_enabled": True,
"cloudhooks": {},
"google_enabled": True,
"google_secure_devices_pin": None,
"google_default_expose": DEFAULT_EXPOSED_DOMAINS,
"alexa_default_expose": DEFAULT_EXPOSED_DOMAINS,
"alexa_report_state": True,
"google_report_state": True,
"onboarded_items": [],
"onboarding_postponed_until": None,
"remote_allow_remote_enable": True,
"remote_enabled": False,
"cloud_ice_servers_enabled": True,
"tts_default_voice": ["en-US", "JennyNeural"],
},
"alexa_entities": {
"include_domains": [],
"include_entity_globs": [],
"include_entities": ["light.kitchen", "switch.ac"],
"exclude_domains": [],
"exclude_entity_globs": [],
"exclude_entities": [],
},
"alexa_registered": False,
"google_entities": {
"include_domains": ["light"],
"include_entity_globs": [],
"include_entities": [],
"exclude_domains": [],
"exclude_entity_globs": [],
"exclude_entities": [],
},
"google_registered": False,
"google_local_connected": False,
"remote_domain": None,
"remote_connected": False,
"remote_certificate_status": None,
"remote_certificate": None,
"http_use_ssl": False,
"active_subscription": True,
"onboarding_completed": False,
"onboarding_postponed": False,
}
async def test_websocket_status_not_logged_in(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test querying the status not logged in."""
cloud.id_token = None
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/status"})
response = await client.receive_json()
assert response["result"] == {
"logged_in": False,
"cloud": "disconnected",
"http_use_ssl": False,
"auto_login": None,
}
async def test_websocket_subscription_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
aioclient_mock: AiohttpClientMocker,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test subscription info."""
cloud.payments.subscription_info.return_value = {"provider": "stripe"}
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscription"})
response = await client.receive_json()
assert response["result"] == {"provider": "stripe"}
async def test_websocket_subscription_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
aioclient_mock: AiohttpClientMocker,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test subscription info fail."""
cloud.payments.subscription_info.side_effect = PaymentsApiError(
"Failed to fetch subscription information"
)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscription"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "request_failed"
async def test_websocket_subscription_not_logged_in(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test subscription info not logged in."""
cloud.id_token = None
client = await hass_ws_client(hass)
with patch(
"hass_nabucasa.payments_api.PaymentsApi.subscription_info",
return_value={"return": "value"},
):
await client.send_json({"id": 5, "type": "cloud/subscription"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "not_logged_in"
async def test_websocket_update_preferences(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test updating preference."""
assert cloud.client.prefs.google_enabled
assert cloud.client.prefs.alexa_enabled
assert cloud.client.prefs.google_secure_devices_pin is None
assert cloud.client.prefs.remote_allow_remote_enable is True
assert cloud.client.prefs.cloud_ice_servers_enabled is True
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "cloud/update_prefs",
"alexa_enabled": False,
"google_enabled": False,
"google_secure_devices_pin": "1234",
"tts_default_voice": ["en-GB", "RyanNeural"],
"remote_allow_remote_enable": False,
"cloud_ice_servers_enabled": False,
}
)
response = await client.receive_json()
assert response["success"]
assert not cloud.client.prefs.google_enabled
assert not cloud.client.prefs.alexa_enabled
assert cloud.client.prefs.google_secure_devices_pin == "1234"
assert cloud.client.prefs.remote_allow_remote_enable is False
assert cloud.client.prefs.cloud_ice_servers_enabled is False
assert cloud.client.prefs.tts_default_voice == ("en-GB", "RyanNeural")
@pytest.mark.parametrize(
("language", "voice"), [("en-GB", "bad_voice"), ("bad_language", "RyanNeural")]
)
async def test_websocket_update_preferences_bad_voice(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
language: str,
voice: str,
) -> None:
"""Test updating preference."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "cloud/update_prefs",
"tts_default_voice": [language, voice],
}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == ERR_INVALID_FORMAT
assert cloud.client.prefs.tts_default_voice == ("en-US", "JennyNeural")
async def test_websocket_update_preferences_alexa_report_state(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test updating alexa_report_state sets alexa authorized."""
client = await hass_ws_client(hass)
with (
patch(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig.async_sync_entities"
),
patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_get_access_token"
),
),
patch(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig.set_authorized"
) as set_authorized_mock,
):
set_authorized_mock.assert_not_called()
await client.send_json(
{"id": 5, "type": "cloud/update_prefs", "alexa_report_state": True}
)
response = await client.receive_json()
set_authorized_mock.assert_called_once_with(True)
await hass.async_block_till_done()
assert response["success"]
async def test_websocket_update_preferences_require_relink(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test updating preference requires relink."""
client = await hass_ws_client(hass)
with (
patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_get_access_token"
),
side_effect=alexa_errors.RequireRelink,
),
patch(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig.set_authorized"
) as set_authorized_mock,
):
set_authorized_mock.assert_not_called()
await client.send_json(
{"id": 5, "type": "cloud/update_prefs", "alexa_report_state": True}
)
response = await client.receive_json()
set_authorized_mock.assert_called_once_with(False)
assert not response["success"]
assert response["error"]["code"] == "alexa_relink"
async def test_websocket_update_preferences_no_token(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test updating preference no token available."""
client = await hass_ws_client(hass)
with (
patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_get_access_token"
),
side_effect=alexa_errors.NoTokenAvailable,
),
patch(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig.set_authorized"
) as set_authorized_mock,
):
set_authorized_mock.assert_not_called()
await client.send_json(
{"id": 5, "type": "cloud/update_prefs", "alexa_report_state": True}
)
response = await client.receive_json()
set_authorized_mock.assert_called_once_with(False)
assert not response["success"]
assert response["error"]["code"] == "alexa_relink"
async def test_websocket_cloud_onboarding_postpone(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test postponing onboarding."""
client = await hass_ws_client(hass)
assert cloud.client.prefs.onboarding_postponed is False
await client.send_json_auto_id({"type": "cloud/onboarding/postpone"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["onboarding_postponed"] is True
assert cloud.client.prefs.onboarding_postponed_until is not None
freezer.tick(datetime.timedelta(hours=25))
await client.send_json_auto_id({"type": "cloud/status"})
response = await client.receive_json()
assert response["result"]["onboarding_postponed"] is False
async def test_websocket_cloud_onboarding_complete(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test completing onboarding items."""
client = await hass_ws_client(hass)
assert cloud.client.prefs.onboarded_items == []
assert cloud.client.prefs.onboarding_completed is False
# Complete a subset of items
await client.send_json_auto_id(
{"type": "cloud/onboarding/complete", "items": ["remote", "backup"]}
)
response = await client.receive_json()
assert response["success"]
assert cloud.client.prefs.onboarded_items == ["remote", "backup"]
assert response["result"]["onboarding_completed"] is False
# Already-completed items are ignored, only new ones are added
await client.send_json_auto_id(
{
"type": "cloud/onboarding/complete",
"items": ["remote", "voice", "streaming"],
}
)
response = await client.receive_json()
assert response["success"]
assert cloud.client.prefs.onboarding_completed is True
assert response["result"]["onboarding_completed"] is True
assert cloud.client.prefs.onboarded_items == [
"remote",
"backup",
"voice",
"streaming",
]
# Completing already-completed items is a no-op
await client.send_json_auto_id(
{"type": "cloud/onboarding/complete", "items": ["remote", "backup"]}
)
response = await client.receive_json()
assert response["success"]
assert cloud.client.prefs.onboarded_items == [
"remote",
"backup",
"voice",
"streaming",
]
async def test_websocket_cloud_onboarding_complete_invalid_item(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test completing an invalid onboarding item."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{"type": "cloud/onboarding/complete", "items": ["remote", "invalid"]}
)
response = await client.receive_json()
assert not response["success"]
async def test_enabling_webhook(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test we call right code to enable webhooks."""
client = await hass_ws_client(hass)
mock_enable = cloud.cloudhooks.async_create
mock_enable.return_value = {}
await client.send_json(
{"id": 5, "type": "cloud/cloudhook/create", "webhook_id": "mock-webhook-id"}
)
response = await client.receive_json()
assert response["success"]
assert mock_enable.call_count == 1
assert mock_enable.mock_calls[0][1][0] == "mock-webhook-id"
async def test_disabling_webhook(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test we call right code to disable webhooks."""
client = await hass_ws_client(hass)
mock_disable = cloud.cloudhooks.async_delete
await client.send_json(
{"id": 5, "type": "cloud/cloudhook/delete", "webhook_id": "mock-webhook-id"}
)
response = await client.receive_json()
assert response["success"]
assert mock_disable.call_count == 1
assert mock_disable.mock_calls[0][1][0] == "mock-webhook-id"
async def test_enabling_remote(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test we call right code to enable remote UI."""
client = await hass_ws_client(hass)
mock_connect = cloud.remote.connect
assert not cloud.client.remote_autostart
await client.send_json({"id": 5, "type": "cloud/remote/connect"})
response = await client.receive_json()
assert response["success"]
assert cloud.client.remote_autostart
assert mock_connect.call_count == 1
mock_disconnect = cloud.remote.disconnect
await client.send_json({"id": 6, "type": "cloud/remote/disconnect"})
response = await client.receive_json()
assert response["success"]
assert not cloud.client.remote_autostart
assert mock_disconnect.call_count == 1
async def test_enabling_remote_remote_activation_not_allowed(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test we can enable remote UI locally when blocked remotely."""
client = await hass_ws_client(hass)
mock_connect = cloud.remote.connect
assert not cloud.client.remote_autostart
await cloud.client.prefs.async_update(remote_allow_remote_enable=False)
await client.send_json({"id": 5, "type": "cloud/remote/connect"})
response = await client.receive_json()
assert response["success"]
assert cloud.client.remote_autostart
assert mock_connect.call_count == 1
mock_disconnect = cloud.remote.disconnect
await client.send_json({"id": 6, "type": "cloud/remote/disconnect"})
response = await client.receive_json()
assert response["success"]
assert not cloud.client.remote_autostart
assert mock_disconnect.call_count == 1
async def test_list_google_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can list Google entities."""
client = await hass_ws_client(hass)
entity = GoogleEntity(
hass, MockConfig(should_expose=lambda *_: False), State("light.kitchen", "on")
)
entity2 = GoogleEntity(
hass,
MockConfig(should_expose=lambda *_: True, should_2fa=lambda *_: False),
State("cover.garage", "open", {"device_class": "garage"}),
)
with patch(
"homeassistant.components.google_assistant.helpers.async_get_entities",
return_value=[entity, entity2],
):
await client.send_json_auto_id({"type": "cloud/google_assistant/entities"})
response = await client.receive_json()
assert response["success"]
assert len(response["result"]) == 2
assert response["result"][0] == {
"entity_id": "light.kitchen",
"might_2fa": False,
"traits": ["action.devices.traits.OnOff"],
}
assert response["result"][1] == {
"entity_id": "cover.garage",
"might_2fa": True,
"traits": ["action.devices.traits.OpenClose"],
}
# Add the entities to the entity registry
entity_registry.async_get_or_create(
"light", "test", "unique", suggested_object_id="kitchen"
)
entity_registry.async_get_or_create(
"cover", "test", "unique", suggested_object_id="garage"
)
with patch(
"homeassistant.components.google_assistant.helpers.async_get_entities",
return_value=[entity, entity2],
):
await client.send_json_auto_id({"type": "cloud/google_assistant/entities"})
response = await client.receive_json()
assert response["success"]
assert len(response["result"]) == 2
assert response["result"][0] == {
"entity_id": "light.kitchen",
"might_2fa": False,
"traits": ["action.devices.traits.OnOff"],
}
assert response["result"][1] == {
"entity_id": "cover.garage",
"might_2fa": True,
"traits": ["action.devices.traits.OpenClose"],
}
async def test_get_google_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can get a Google entity."""
client = await hass_ws_client(hass)
# Test getting an unknown entity
await client.send_json_auto_id(
{"type": "cloud/google_assistant/entities/get", "entity_id": "light.kitchen"}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"] == {
"code": "not_found",
"message": "light.kitchen unknown",
}
entity_registry.async_get_or_create(
"light", "test", "unique", suggested_object_id="kitchen"
)
hass.states.async_set("light.kitchen", "on")
hass.states.async_set("cover.garage", "open", {"device_class": "garage"})
await client.send_json_auto_id(
{"type": "cloud/google_assistant/entities/get", "entity_id": "light.kitchen"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"disable_2fa": None,
"entity_id": "light.kitchen",
"might_2fa": False,
"traits": ["action.devices.traits.OnOff"],
}
await client.send_json_auto_id(
{"type": "cloud/google_assistant/entities/get", "entity_id": "cover.garage"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"disable_2fa": None,
"entity_id": "cover.garage",
"might_2fa": True,
"traits": ["action.devices.traits.OpenClose"],
}
# Set the disable 2fa flag
await client.send_json_auto_id(
{
"type": "cloud/google_assistant/entities/update",
"entity_id": "cover.garage",
"disable_2fa": True,
}
)
response = await client.receive_json()
assert response["success"]
await client.send_json_auto_id(
{"type": "cloud/google_assistant/entities/get", "entity_id": "cover.garage"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"disable_2fa": True,
"entity_id": "cover.garage",
"might_2fa": True,
"traits": ["action.devices.traits.OpenClose"],
}
async def test_update_google_entity(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can update config of a Google entity."""
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "cloud/google_assistant/entities/update",
"entity_id": "light.kitchen",
"disable_2fa": False,
}
)
response = await client.receive_json()
assert response["success"]
await client.send_json_auto_id(
{
"type": "homeassistant/expose_entity",
"assistants": ["cloud.google_assistant"],
"entity_ids": ["light.kitchen"],
"should_expose": False,
}
)
response = await client.receive_json()
assert response["success"]
assert exposed_entities.async_get_entity_settings(hass, "light.kitchen") == {
"cloud.google_assistant": {"disable_2fa": False, "should_expose": False}
}
async def test_list_alexa_entities(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can list Alexa entities."""
client = await hass_ws_client(hass)
entity = LightCapabilities(
hass, MagicMock(entity_config={}), State("light.kitchen", "on")
)
with patch(
"homeassistant.components.alexa.entities.async_get_entities",
return_value=[entity],
):
await client.send_json_auto_id({"id": 5, "type": "cloud/alexa/entities"})
response = await client.receive_json()
assert response["success"]
assert len(response["result"]) == 1
assert response["result"][0] == {
"entity_id": "light.kitchen",
"display_categories": ["LIGHT"],
"interfaces": ["Alexa.PowerController", "Alexa.EndpointHealth", "Alexa"],
}
with (
patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_get_access_token"
),
),
patch(
"homeassistant.components.cloud.alexa_config.alexa_state_report.async_send_add_or_update_message"
),
):
# Add the entity to the entity registry
entity_registry.async_get_or_create(
"light", "test", "unique", suggested_object_id="kitchen"
)
await hass.async_block_till_done()
with patch(
"homeassistant.components.alexa.entities.async_get_entities",
return_value=[entity],
):
await client.send_json_auto_id({"type": "cloud/alexa/entities"})
response = await client.receive_json()
assert response["success"]
assert len(response["result"]) == 1
assert response["result"][0] == {
"entity_id": "light.kitchen",
"display_categories": ["LIGHT"],
"interfaces": ["Alexa.PowerController", "Alexa.EndpointHealth", "Alexa"],
}
async def test_get_alexa_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can get an Alexa entity."""
client = await hass_ws_client(hass)
# Test getting an unknown entity
await client.send_json_auto_id(
{"type": "cloud/alexa/entities/get", "entity_id": "light.kitchen"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] is None
# Test getting an unknown sensor
await client.send_json_auto_id(
{"type": "cloud/alexa/entities/get", "entity_id": "sensor.temperature"}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"] == {
"code": "not_supported",
"message": "sensor.temperature not supported by Alexa",
}
entity_registry.async_get_or_create(
"light", "test", "unique", suggested_object_id="kitchen"
)
entity_registry.async_get_or_create(
"water_heater", "test", "unique", suggested_object_id="basement"
)
await client.send_json_auto_id(
{"type": "cloud/alexa/entities/get", "entity_id": "light.kitchen"}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] is None
await client.send_json_auto_id(
{"type": "cloud/alexa/entities/get", "entity_id": "water_heater.basement"}
)
response = await client.receive_json()
assert not response["success"]
assert response["error"] == {
"code": "not_supported",
"message": "water_heater.basement not supported by Alexa",
}
async def test_update_alexa_entity(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can update config of an Alexa entity."""
entry = entity_registry.async_get_or_create(
"light", "test", "unique", suggested_object_id="kitchen"
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "homeassistant/expose_entity",
"assistants": ["cloud.alexa"],
"entity_ids": [entry.entity_id],
"should_expose": False,
}
)
response = await client.receive_json()
assert response["success"]
assert exposed_entities.async_get_entity_settings(hass, entry.entity_id) == {
"cloud.alexa": {"should_expose": False}
}
async def test_sync_alexa_entities_timeout(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that timeout syncing Alexa entities."""
client = await hass_ws_client(hass)
with patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_sync_entities"
),
side_effect=TimeoutError,
):
await client.send_json({"id": 5, "type": "cloud/alexa/sync"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "timeout"
async def test_sync_alexa_entities_no_token(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test sync Alexa entities when we have no token."""
client = await hass_ws_client(hass)
with patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_sync_entities"
),
side_effect=alexa_errors.NoTokenAvailable,
):
await client.send_json({"id": 5, "type": "cloud/alexa/sync"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "alexa_relink"
async def test_enable_alexa_state_report_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test enable Alexa entities state reporting when no token available."""
client = await hass_ws_client(hass)
with patch(
(
"homeassistant.components.cloud.alexa_config.CloudAlexaConfig"
".async_sync_entities"
),
side_effect=alexa_errors.NoTokenAvailable,
):
await client.send_json({"id": 5, "type": "cloud/alexa/sync"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "alexa_relink"
async def test_tts_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
setup_cloud: None,
) -> None:
"""Test that we can get TTS info."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "cloud/tts/info"})
response = await client.receive_json()
assert response["success"]
assert "languages" in response["result"]
assert all(len(lang) for lang in response["result"]["languages"])
assert len(response["result"]["languages"]) > 300
assert (
len([lang for lang in response["result"]["languages"] if "||" in lang[1]]) > 100
)
for lang in response["result"]["languages"]:
assert validate_language_voice(lang[:2])
@pytest.mark.parametrize(
("endpoint", "data"),
[
("/api/cloud/forgot_password", {"email": "fake@example.com"}),
("/api/cloud/google_actions/sync", None),
("/api/cloud/login", {"email": "fake@example.com", "password": "secret"}),
("/api/cloud/logout", None),
("/api/cloud/register", {"email": "fake@example.com", "password": "secret"}),
(
"/api/cloud/register_auto_login",
{"email": "fake@example.com", "password": "secret"},
),
("/api/cloud/resend_confirm", {"email": "fake@example.com"}),
],
)
async def test_api_calls_require_admin(
hass: HomeAssistant,
setup_cloud: None,
hass_client: ClientSessionGenerator,
hass_read_only_access_token: str,
endpoint: str,
data: dict[str, Any] | None,
) -> None:
"""Test cloud APIs endpoints do not work as a normal user."""
client = await hass_client(hass_read_only_access_token)
resp = await client.post(endpoint, json=data)
assert resp.status == HTTPStatus.UNAUTHORIZED
async def test_support_package_requires_admin(
setup_cloud: None,
hass_client: ClientSessionGenerator,
hass_read_only_access_token: str,
) -> None:
"""Test the support package download is restricted to admins."""
client = await hass_client(hass_read_only_access_token)
resp = await client.get("/api/cloud/support_package")
assert resp.status == HTTPStatus.UNAUTHORIZED
@pytest.mark.parametrize(
"msg",
[
{"type": "cloud/subscription"},
{"type": "cloud/update_prefs", "alexa_report_state": True},
{"type": "cloud/cloudhook/create", "webhook_id": "mock-webhook-id"},
{"type": "cloud/cloudhook/delete", "webhook_id": "mock-webhook-id"},
{"type": "cloud/onboarding/postpone"},
{"type": "cloud/onboarding/complete", "items": ["remote"]},
{"type": "cloud/subscribe_events"},
{"type": "cloud/attempt_auto_login_now"},
{"type": "cloud/resend_auto_login_confirm"},
{"type": "cloud/cancel_auto_login"},
],
)
async def test_ws_commands_require_admin(
hass: HomeAssistant,
setup_cloud: None,
hass_ws_client: WebSocketGenerator,
hass_read_only_access_token: str,
msg: dict[str, Any],
) -> None:
"""Test cloud WebSocket commands do not work as a normal user."""
client = await hass_ws_client(hass, hass_read_only_access_token)
await client.send_json({"id": 5, **msg})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "unauthorized"
async def test_login_view_dispatch_event(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
) -> None:
"""Test dispatching event while logging in."""
assert await async_setup_component(hass, "homeassistant", {})
assert await async_setup_component(hass, DOMAIN, {"cloud": {}})
await hass.async_block_till_done()
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.async_dispatcher_send"
) as async_dispatcher_send_mock:
await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert async_dispatcher_send_mock.call_count == 1
assert async_dispatcher_send_mock.mock_calls[0][1][1] == "cloud_event"
assert async_dispatcher_send_mock.mock_calls[0][1][2] == {"type": "login"}
async def test_logout_view_dispatch_event(
cloud: MagicMock,
setup_cloud: None,
hass_client: ClientSessionGenerator,
) -> None:
"""Test dispatching event while logging out."""
cloud_client = await hass_client()
with patch(
"homeassistant.components.cloud.http_api.async_dispatcher_send"
) as async_dispatcher_send_mock:
await cloud_client.post("/api/cloud/logout")
assert async_dispatcher_send_mock.call_count == 1
assert async_dispatcher_send_mock.mock_calls[0][1][1] == "cloud_event"
assert async_dispatcher_send_mock.mock_calls[0][1][2] == {"type": "logout"}
@pytest.mark.usefixtures("cloud", "setup_cloud")
async def test_cloud_events_subscription(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test subscribing to cloud events over the websocket."""
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscribe_events"})
response = await client.receive_json()
assert response["success"]
cloud_client = await hass_client()
req = await cloud_client.post(
"/api/cloud/login", json={"email": "my_username", "password": "my_password"}
)
assert req.status == HTTPStatus.OK
event = await client.receive_json()
assert event["id"] == 5
assert event["event"] == {"type": "login"}
req = await cloud_client.post("/api/cloud/logout")
assert req.status == HTTPStatus.OK
event = await client.receive_json()
assert event["id"] == 5
assert event["event"] == {"type": "logout"}
@pytest.mark.usefixtures("setup_cloud")
async def test_cloud_events_auto_login_push(
hass: HomeAssistant,
cloud: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
) -> None:
"""Test a successful auto-login is pushed to cloud events subscribers."""
cloud.id_token = None
await register_auto_login(hass_client)
client = await hass_ws_client(hass)
await client.send_json({"id": 5, "type": "cloud/subscribe_events"})
response = await client.receive_json()
assert response["success"]
# Simulate the library retry task logging in.
await cloud.login("hello@bla.com", "falcon42")
event = await client.receive_json()
assert event["id"] == 5
assert event["event"] == {"type": "login"}
@patch("homeassistant.components.cloud.helpers.FixedSizeQueueLogHandler.MAX_RECORDS", 3)
@pytest.mark.usefixtures("enable_custom_integrations")
async def test_download_support_package(
hass: HomeAssistant,
cloud: MagicMock,
set_cloud_prefs: Callable[[dict[str, Any]], Coroutine[Any, Any, None]],
hass_client: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test downloading a support package file."""
aioclient_mock.get("https://cloud.bla.com/status", text="")
aioclient_mock.get(
"https://cert-server/directory", exc=Exception("Unexpected exception")
)
aioclient_mock.get(
"https://cognito-idp.us-east-1.amazonaws.com/AAAA/.well-known/jwks.json",
exc=aiohttp.ClientError,
)
def async_register_mock_platform(
hass: HomeAssistant, register: system_health.SystemHealthRegistration
) -> None:
async def mock_empty_info(hass: HomeAssistant) -> dict[str, Any]:
return {}
register.async_register_info(mock_empty_info, "/config/mock_integration")
mock_platform(
hass,
"mock_no_info_integration.system_health",
MagicMock(async_register=async_register_mock_platform),
)
hass.config.components.add("mock_no_info_integration")
# Add mock custom integration for testing
hass.config.components.add("test") # This is a custom integration from the fixture
assert await async_setup_component(hass, "system_health", {})
# Register custom themes so the support package surfaces them
hass.data[DATA_THEMES] = {
"midnight": {"primary-color": "#000000"},
"solarized-dark": {"primary-color": "#002b36"},
}
with patch("uuid.UUID.hex", new_callable=PropertyMock) as hexmock:
hexmock.return_value = "12345678901234567890"
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
"user_pool_id": "AAAA",
"region": "us-east-1",
"acme_server": "cert-server",
"relayer_server": "cloud.bla.com",
},
},
)
await hass.async_block_till_done()
await cloud.login("test-user", "test-pass")
cloud.remote.snitun_server = "us-west-1"
cloud.remote.certificate_status = CertificateStatus.READY
cloud.remote.latency_by_location = {
"Earth": {"avg": 13.37},
"Moon": {"avg": None},
}
cloud.expiration_date = dt_util.parse_datetime("2025-01-17T11:19:31.0+00:00")
await cloud.client.async_system_message({"region": "xx-earth-616"})
await set_cloud_prefs(
{
"alexa_enabled": True,
"google_enabled": False,
"remote_enabled": True,
"cloud_ice_servers_enabled": True,
}
)
now = dt_util.utcnow()
# The logging is done with local time according to the system timezone. Set the
# fake time to 12:00 local time
tz = now.astimezone().tzinfo
freezer.move_to(datetime.datetime(2025, 2, 10, 12, 0, 0, tzinfo=tz))
logging.getLogger("hass_nabucasa.iot").info(
"This message will be dropped since this test patches MAX_RECORDS"
)
logging.getLogger("hass_nabucasa.iot").info("Hass nabucasa log")
logging.getLogger("snitun.utils.aiohttp_client").warning("Snitun log")
logging.getLogger("homeassistant.components.cloud.client").error("Cloud log")
freezer.move_to(now) # Reset time otherwise hass_client auth fails
cloud_client = await hass_client()
with (
patch.object(hass.config, "config_dir", new="config"),
patch(
"homeassistant.components.homeassistant.system_health.system_info.async_get_system_info",
return_value={
"installation_type": "Home Assistant Core",
"version": "2025.2.0",
"dev": False,
"hassio": False,
"virtualenv": False,
"python_version": "3.13.1",
"docker": False,
"container_arch": None,
"arch": "x86_64",
"timezone": "US/Pacific",
"os_name": "Linux",
"os_version": "6.12.9",
"user": "hass",
},
),
patch(
"homeassistant.components.cloud.http_api.async_get_installed_packages",
return_value=[
{"name": "homeassistant", "version": "3.2.1"},
{"name": "hass-nabucasa", "version": "1.2.3"},
],
),
):
req = await cloud_client.get("/api/cloud/support_package")
assert req.status == HTTPStatus.OK
assert await req.text() == snapshot
@pytest.mark.usefixtures("enable_custom_integrations")
async def test_download_support_package_custom_components_error(
hass: HomeAssistant,
cloud: MagicMock,
set_cloud_prefs: Callable[[dict[str, Any]], Coroutine[Any, Any, None]],
hass_client: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test download support package when async_get_custom_components fails."""
aioclient_mock.get("https://cloud.bla.com/status", text="")
aioclient_mock.get(
"https://cert-server/directory", exc=Exception("Unexpected exception")
)
aioclient_mock.get(
"https://cognito-idp.us-east-1.amazonaws.com/AAAA/.well-known/jwks.json",
exc=aiohttp.ClientError,
)
def async_register_mock_platform(
hass: HomeAssistant, register: system_health.SystemHealthRegistration
) -> None:
async def mock_empty_info(hass: HomeAssistant) -> dict[str, Any]:
return {}
register.async_register_info(mock_empty_info, "/config/mock_integration")
mock_platform(
hass,
"mock_no_info_integration.system_health",
MagicMock(async_register=async_register_mock_platform),
)
hass.config.components.add("mock_no_info_integration")
assert await async_setup_component(hass, "system_health", {})
with patch("uuid.UUID.hex", new_callable=PropertyMock) as hexmock:
hexmock.return_value = "12345678901234567890"
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
"user_pool_id": "AAAA",
"region": "us-east-1",
"acme_server": "cert-server",
"relayer_server": "cloud.bla.com",
},
},
)
await hass.async_block_till_done()
await cloud.login("test-user", "test-pass")
cloud.remote.snitun_server = "us-west-1"
cloud.remote.certificate_status = CertificateStatus.READY
cloud.expiration_date = dt_util.parse_datetime("2025-01-17T11:19:31.0+00:00")
await cloud.client.async_system_message({"region": "xx-earth-616"})
await set_cloud_prefs(
{
"alexa_enabled": True,
"google_enabled": False,
"remote_enabled": True,
"cloud_ice_servers_enabled": True,
}
)
now = dt_util.utcnow()
tz = now.astimezone().tzinfo
freezer.move_to(datetime.datetime(2025, 2, 10, 12, 0, 0, tzinfo=tz))
logging.getLogger("hass_nabucasa.iot").info(
"This message will be dropped since this test patches MAX_RECORDS"
)
logging.getLogger("hass_nabucasa.iot").info("Hass nabucasa log")
logging.getLogger("snitun.utils.aiohttp_client").warning("Snitun log")
logging.getLogger("homeassistant.components.cloud.client").error("Cloud log")
freezer.move_to(now)
cloud_client = await hass_client()
with (
patch.object(hass.config, "config_dir", new="config"),
patch(
"homeassistant.components.homeassistant.system_health.system_info.async_get_system_info",
return_value={
"installation_type": "Home Assistant Core",
"version": "2025.2.0",
"dev": False,
"hassio": False,
"virtualenv": False,
"python_version": "3.13.1",
"docker": False,
"container_arch": None,
"arch": "x86_64",
"timezone": "US/Pacific",
"os_name": "Linux",
"os_version": "6.12.9",
"user": "hass",
},
),
patch(
"homeassistant.components.cloud.http_api.async_get_custom_components",
side_effect=Exception("Custom components error"),
),
patch(
"homeassistant.components.cloud.http_api.async_get_installed_packages",
return_value=[
{"name": "homeassistant", "version": "3.2.1"},
{"name": "hass-nabucasa", "version": "1.2.3"},
],
),
):
req = await cloud_client.get("/api/cloud/support_package")
assert req.status == HTTPStatus.OK
assert await req.text() == snapshot
@pytest.mark.usefixtures("enable_custom_integrations")
async def test_download_support_package_integration_load_error(
hass: HomeAssistant,
cloud: MagicMock,
set_cloud_prefs: Callable[[dict[str, Any]], Coroutine[Any, Any, None]],
hass_client: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
freezer: FrozenDateTimeFactory,
snapshot: SnapshotAssertion,
) -> None:
"""Test download support package when async_get_loaded_integration fails."""
aioclient_mock.get("https://cloud.bla.com/status", text="")
aioclient_mock.get(
"https://cert-server/directory", exc=Exception("Unexpected exception")
)
aioclient_mock.get(
"https://cognito-idp.us-east-1.amazonaws.com/AAAA/.well-known/jwks.json",
exc=aiohttp.ClientError,
)
def async_register_mock_platform(
hass: HomeAssistant, register: system_health.SystemHealthRegistration
) -> None:
async def mock_empty_info(hass: HomeAssistant) -> dict[str, Any]:
return {}
register.async_register_info(mock_empty_info, "/config/mock_integration")
mock_platform(
hass,
"mock_no_info_integration.system_health",
MagicMock(async_register=async_register_mock_platform),
)
hass.config.components.add("mock_no_info_integration")
# Add a component that will fail to load integration info
hass.config.components.add("test") # This is a custom integration from the fixture
hass.config.components.add("failing_integration")
assert await async_setup_component(hass, "system_health", {})
with patch("uuid.UUID.hex", new_callable=PropertyMock) as hexmock:
hexmock.return_value = "12345678901234567890"
assert await async_setup_component(
hass,
DOMAIN,
{
DOMAIN: {
"user_pool_id": "AAAA",
"region": "us-east-1",
"acme_server": "cert-server",
"relayer_server": "cloud.bla.com",
},
},
)
await hass.async_block_till_done()
await cloud.login("test-user", "test-pass")
cloud.remote.snitun_server = "us-west-1"
cloud.remote.certificate_status = CertificateStatus.READY
cloud.expiration_date = dt_util.parse_datetime("2025-01-17T11:19:31.0+00:00")
await cloud.client.async_system_message({"region": "xx-earth-616"})
await set_cloud_prefs(
{
"alexa_enabled": True,
"google_enabled": False,
"remote_enabled": True,
"cloud_ice_servers_enabled": True,
}
)
now = dt_util.utcnow()
tz = now.astimezone().tzinfo
freezer.move_to(datetime.datetime(2025, 2, 10, 12, 0, 0, tzinfo=tz))
logging.getLogger("hass_nabucasa.iot").info(
"This message will be dropped since this test patches MAX_RECORDS"
)
logging.getLogger("hass_nabucasa.iot").info("Hass nabucasa log")
logging.getLogger("snitun.utils.aiohttp_client").warning("Snitun log")
logging.getLogger("homeassistant.components.cloud.client").error("Cloud log")
freezer.move_to(now)
cloud_client = await hass_client()
with (
patch.object(hass.config, "config_dir", new="config"),
patch(
"homeassistant.components.homeassistant.system_health.system_info.async_get_system_info",
return_value={
"installation_type": "Home Assistant Core",
"version": "2025.2.0",
"dev": False,
"hassio": False,
"virtualenv": False,
"python_version": "3.13.1",
"docker": False,
"container_arch": None,
"arch": "x86_64",
"timezone": "US/Pacific",
"os_name": "Linux",
"os_version": "6.12.9",
"user": "hass",
},
),
patch(
"homeassistant.components.cloud.http_api.async_get_loaded_integration",
side_effect=lambda hass, domain: (
Exception("Integration load error")
if domain == "failing_integration"
else async_get_loaded_integration(hass, domain)
),
),
patch(
"homeassistant.components.cloud.http_api.async_get_installed_packages",
return_value=[
{"name": "homeassistant", "version": "3.2.1"},
{"name": "hass-nabucasa", "version": "1.2.3"},
],
),
):
req = await cloud_client.get("/api/cloud/support_package")
assert req.status == HTTPStatus.OK
assert await req.text() == snapshot
async def test_websocket_ice_servers(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
cloud: MagicMock,
setup_cloud: None,
) -> None:
"""Test getting ICE servers."""
cloud.client._ice_servers = [
RTCIceServer(urls="stun:stun.l.bla.com:19302"),
RTCIceServer(
urls="turn:turn.example.com:3478", username="user", credential="pass"
),
]
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "cloud/webrtc/ice_servers"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == [
{"urls": "stun:stun.l.bla.com:19302"},
{
"urls": "turn:turn.example.com:3478",
"username": "user",
"credential": "pass",
},
]
cloud.id_token = None
await client.send_json_auto_id({"type": "cloud/webrtc/ice_servers"})
response = await client.receive_json()
assert not response["success"]
assert response["error"]["code"] == "not_logged_in"
Reference in New Issue View Git Blame Copy Permalink