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

3011 lines
95 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,
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 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
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_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()
req = await cloud_client.post(
"/api/cloud/register_auto_login",
json={"email": "hello@bla.com", "password": "falcon42"},
)
assert req.status == HTTPStatus.BAD_REQUEST
cloud.register_and_auto_login.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)
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 cloud.register_and_auto_login.return_value.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"
controller = cloud.register_and_auto_login.return_value
assert getattr(controller, controller_method).call_count == 0
@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 cloud.events.publish(LoginFailedEvent(auto=True, reason=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 cloud.events.publish(
LoginFailedEvent(auto=True, reason=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 cloud.events.publish(
LoginFailedEvent(auto=True, reason=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)
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 cloud.register_and_auto_login.return_value.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 = cloud.register_and_auto_login.return_value
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
controller = cloud.register_and_auto_login.return_value
controller.resend.side_effect = UnknownError
await register_auto_login(hass_client)
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_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)
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 cloud.register_and_auto_login.return_value.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_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)
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 cloud.register_and_auto_login.return_value.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