mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 13:17:32 +00:00
Move cloud onboarding API to an onboarding platform (#141978)
* Move cloud onboarding API to an onboarding platform * Address review comments * Add tests * Move cloud onboarding tests to the cloud integration * Address review comments * Don't wait for platforms * Add test * Remove useless check for CLOUD_DATA
This commit is contained in:
Erik Montnemery
GitHub
parent
a4234bf80e
commit
4aca9cd66b
110
homeassistant/components/cloud/onboarding.py
Normal file
110
homeassistant/components/cloud/onboarding.py
Normal file
@ -0,0 +1,110 @@
|
||||
"""Cloud onboarding views."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Coroutine
|
||||
from functools import wraps
|
||||
from typing import TYPE_CHECKING, Any, Concatenate
|
||||
|
||||
from aiohttp import web
|
||||
from aiohttp.web_exceptions import HTTPUnauthorized
|
||||
|
||||
from homeassistant.components.http import KEY_HASS
|
||||
from homeassistant.components.onboarding import (
|
||||
BaseOnboardingView,
|
||||
NoAuthBaseOnboardingView,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from . import http_api as cloud_http
|
||||
from .const import DATA_CLOUD
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from homeassistant.components.onboarding import OnboardingStoreData
|
||||
|
||||
|
||||
async def async_setup_views(hass: HomeAssistant, data: OnboardingStoreData) -> None:
|
||||
"""Set up the cloud views."""
|
||||
|
||||
hass.http.register_view(CloudForgotPasswordView(data))
|
||||
hass.http.register_view(CloudLoginView(data))
|
||||
hass.http.register_view(CloudLogoutView(data))
|
||||
hass.http.register_view(CloudStatusView(data))
|
||||
|
||||
|
||||
def ensure_not_done[_ViewT: BaseOnboardingView, **_P](
|
||||
func: Callable[
|
||||
Concatenate[_ViewT, web.Request, _P],
|
||||
Coroutine[Any, Any, web.Response],
|
||||
],
|
||||
) -> Callable[Concatenate[_ViewT, web.Request, _P], Coroutine[Any, Any, web.Response]]:
|
||||
"""Home Assistant API decorator to check onboarding and cloud."""
|
||||
|
||||
@wraps(func)
|
||||
async def _ensure_not_done(
|
||||
self: _ViewT,
|
||||
request: web.Request,
|
||||
*args: _P.args,
|
||||
**kwargs: _P.kwargs,
|
||||
) -> web.Response:
|
||||
"""Check onboarding status, cloud and call function."""
|
||||
if self._data["done"]:
|
||||
# If at least one onboarding step is done, we don't allow accessing
|
||||
# the cloud onboarding views.
|
||||
raise HTTPUnauthorized
|
||||
|
||||
return await func(self, request, *args, **kwargs)
|
||||
|
||||
return _ensure_not_done
|
||||
|
||||
|
||||
class CloudForgotPasswordView(
|
||||
NoAuthBaseOnboardingView, cloud_http.CloudForgotPasswordView
|
||||
):
|
||||
"""View to start Forgot Password flow."""
|
||||
|
||||
url = "/api/onboarding/cloud/forgot_password"
|
||||
name = "api:onboarding:cloud:forgot_password"
|
||||
|
||||
@ensure_not_done
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle forgot password request."""
|
||||
return await super()._post(request)
|
||||
|
||||
|
||||
class CloudLoginView(NoAuthBaseOnboardingView, cloud_http.CloudLoginView):
|
||||
"""Login to Home Assistant Cloud."""
|
||||
|
||||
url = "/api/onboarding/cloud/login"
|
||||
name = "api:onboarding:cloud:login"
|
||||
|
||||
@ensure_not_done
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle login request."""
|
||||
return await super()._post(request)
|
||||
|
||||
|
||||
class CloudLogoutView(NoAuthBaseOnboardingView, cloud_http.CloudLogoutView):
|
||||
"""Log out of the Home Assistant cloud."""
|
||||
|
||||
url = "/api/onboarding/cloud/logout"
|
||||
name = "api:onboarding:cloud:logout"
|
||||
|
||||
@ensure_not_done
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle logout request."""
|
||||
return await super()._post(request)
|
||||
|
||||
|
||||
class CloudStatusView(NoAuthBaseOnboardingView):
|
||||
"""Get cloud status view."""
|
||||
|
||||
url = "/api/onboarding/cloud/status"
|
||||
name = "api:onboarding:cloud:status"
|
||||
|
||||
@ensure_not_done
|
||||
async def get(self, request: web.Request) -> web.Response:
|
||||
"""Return cloud status."""
|
||||
hass = request.app[KEY_HASS]
|
||||
cloud = hass.data[DATA_CLOUD]
|
||||
return self.json({"logged_in": cloud.is_logged_in})
|
||||
@ -21,6 +21,7 @@ from .const import (
|
||||
STEP_USER,
|
||||
STEPS,
|
||||
)
|
||||
from .views import BaseOnboardingView, NoAuthBaseOnboardingView # noqa: F401
|
||||
|
||||
STORAGE_KEY = DOMAIN
|
||||
STORAGE_VERSION = 4
|
||||
|
||||
@ -6,7 +6,8 @@ import asyncio
|
||||
from collections.abc import Callable, Coroutine
|
||||
from functools import wraps
|
||||
from http import HTTPStatus
|
||||
from typing import TYPE_CHECKING, Any, Concatenate, cast
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any, Concatenate, Protocol, cast
|
||||
|
||||
from aiohttp import web
|
||||
from aiohttp.web_exceptions import HTTPUnauthorized
|
||||
@ -27,16 +28,11 @@ from homeassistant.components.http.data_validator import RequestDataValidator
|
||||
from homeassistant.components.http.view import HomeAssistantView
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
from homeassistant.helpers import area_registry as ar
|
||||
from homeassistant.helpers import area_registry as ar, integration_platform
|
||||
from homeassistant.helpers.backup import async_get_manager as async_get_backup_manager
|
||||
from homeassistant.helpers.system_info import async_get_system_info
|
||||
from homeassistant.helpers.translation import async_get_translations
|
||||
from homeassistant.setup import (
|
||||
SetupPhases,
|
||||
async_pause_setup,
|
||||
async_setup_component,
|
||||
async_wait_component,
|
||||
)
|
||||
from homeassistant.setup import async_setup_component, async_wait_component
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from . import OnboardingData, OnboardingStorage, OnboardingStoreData
|
||||
@ -51,11 +47,14 @@ from .const import (
|
||||
STEPS,
|
||||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup(
|
||||
hass: HomeAssistant, data: OnboardingStoreData, store: OnboardingStorage
|
||||
) -> None:
|
||||
"""Set up the onboarding view."""
|
||||
await async_process_onboarding_platforms(hass)
|
||||
hass.http.register_view(OnboardingStatusView(data, store))
|
||||
hass.http.register_view(InstallationTypeOnboardingView(data))
|
||||
hass.http.register_view(UserOnboardingView(data, store))
|
||||
@ -66,10 +65,38 @@ async def async_setup(
|
||||
hass.http.register_view(RestoreBackupView(data))
|
||||
hass.http.register_view(UploadBackupView(data))
|
||||
hass.http.register_view(WaitIntegrationOnboardingView(data))
|
||||
await setup_cloud_views(hass, data)
|
||||
|
||||
|
||||
class _BaseOnboardingView(HomeAssistantView):
|
||||
class OnboardingPlatformProtocol(Protocol):
|
||||
"""Define the format of onboarding platforms."""
|
||||
|
||||
async def async_setup_views(
|
||||
self, hass: HomeAssistant, data: OnboardingStoreData
|
||||
) -> None:
|
||||
"""Set up onboarding views."""
|
||||
|
||||
|
||||
async def async_process_onboarding_platforms(hass: HomeAssistant) -> None:
|
||||
"""Start processing onboarding platforms."""
|
||||
await integration_platform.async_process_integration_platforms(
|
||||
hass, DOMAIN, _register_onboarding_platform, wait_for_platforms=False
|
||||
)
|
||||
|
||||
|
||||
async def _register_onboarding_platform(
|
||||
hass: HomeAssistant, integration_domain: str, platform: OnboardingPlatformProtocol
|
||||
) -> None:
|
||||
"""Register a onboarding platform."""
|
||||
if not hasattr(platform, "async_setup_views"):
|
||||
_LOGGER.debug(
|
||||
"'%s.onboarding' is not a valid onboarding platform",
|
||||
integration_domain,
|
||||
)
|
||||
return
|
||||
await platform.async_setup_views(hass, hass.data[DOMAIN].steps)
|
||||
|
||||
|
||||
class BaseOnboardingView(HomeAssistantView):
|
||||
"""Base class for onboarding views."""
|
||||
|
||||
def __init__(self, data: OnboardingStoreData) -> None:
|
||||
@ -77,13 +104,13 @@ class _BaseOnboardingView(HomeAssistantView):
|
||||
self._data = data
|
||||
|
||||
|
||||
class _NoAuthBaseOnboardingView(_BaseOnboardingView):
|
||||
class NoAuthBaseOnboardingView(BaseOnboardingView):
|
||||
"""Base class for unauthenticated onboarding views."""
|
||||
|
||||
requires_auth = False
|
||||
|
||||
|
||||
class OnboardingStatusView(_NoAuthBaseOnboardingView):
|
||||
class OnboardingStatusView(NoAuthBaseOnboardingView):
|
||||
"""Return the onboarding status."""
|
||||
|
||||
url = "/api/onboarding"
|
||||
@ -101,7 +128,7 @@ class OnboardingStatusView(_NoAuthBaseOnboardingView):
|
||||
)
|
||||
|
||||
|
||||
class InstallationTypeOnboardingView(_NoAuthBaseOnboardingView):
|
||||
class InstallationTypeOnboardingView(NoAuthBaseOnboardingView):
|
||||
"""Return the installation type during onboarding."""
|
||||
|
||||
url = "/api/onboarding/installation_type"
|
||||
@ -117,7 +144,7 @@ class InstallationTypeOnboardingView(_NoAuthBaseOnboardingView):
|
||||
return self.json({"installation_type": info["installation_type"]})
|
||||
|
||||
|
||||
class _BaseOnboardingStepView(_BaseOnboardingView):
|
||||
class _BaseOnboardingStepView(BaseOnboardingView):
|
||||
"""Base class for an onboarding step."""
|
||||
|
||||
step: str
|
||||
@ -304,7 +331,7 @@ class IntegrationOnboardingView(_BaseOnboardingStepView):
|
||||
return self.json({"auth_code": auth_code})
|
||||
|
||||
|
||||
class WaitIntegrationOnboardingView(_NoAuthBaseOnboardingView):
|
||||
class WaitIntegrationOnboardingView(NoAuthBaseOnboardingView):
|
||||
"""Get backup info view."""
|
||||
|
||||
url = "/api/onboarding/integration/wait"
|
||||
@ -350,7 +377,7 @@ class AnalyticsOnboardingView(_BaseOnboardingStepView):
|
||||
return self.json({})
|
||||
|
||||
|
||||
def with_backup_manager[_ViewT: _BaseOnboardingView, **_P](
|
||||
def with_backup_manager[_ViewT: BaseOnboardingView, **_P](
|
||||
func: Callable[
|
||||
Concatenate[_ViewT, BackupManager, web.Request, _P],
|
||||
Coroutine[Any, Any, web.Response],
|
||||
@ -382,7 +409,7 @@ def with_backup_manager[_ViewT: _BaseOnboardingView, **_P](
|
||||
return with_backup
|
||||
|
||||
|
||||
class BackupInfoView(_NoAuthBaseOnboardingView):
|
||||
class BackupInfoView(NoAuthBaseOnboardingView):
|
||||
"""Get backup info view."""
|
||||
|
||||
url = "/api/onboarding/backup/info"
|
||||
@ -401,7 +428,7 @@ class BackupInfoView(_NoAuthBaseOnboardingView):
|
||||
)
|
||||
|
||||
|
||||
class RestoreBackupView(_NoAuthBaseOnboardingView):
|
||||
class RestoreBackupView(NoAuthBaseOnboardingView):
|
||||
"""Restore backup view."""
|
||||
|
||||
url = "/api/onboarding/backup/restore"
|
||||
@ -446,7 +473,7 @@ class RestoreBackupView(_NoAuthBaseOnboardingView):
|
||||
return web.Response(status=HTTPStatus.OK)
|
||||
|
||||
|
||||
class UploadBackupView(_NoAuthBaseOnboardingView, backup_http.UploadBackupView):
|
||||
class UploadBackupView(NoAuthBaseOnboardingView, backup_http.UploadBackupView):
|
||||
"""Upload backup view."""
|
||||
|
||||
url = "/api/onboarding/backup/upload"
|
||||
@ -458,116 +485,6 @@ class UploadBackupView(_NoAuthBaseOnboardingView, backup_http.UploadBackupView):
|
||||
return await self._post(request)
|
||||
|
||||
|
||||
async def setup_cloud_views(hass: HomeAssistant, data: OnboardingStoreData) -> None:
|
||||
"""Set up the cloud views."""
|
||||
|
||||
with async_pause_setup(hass, SetupPhases.WAIT_IMPORT_PACKAGES):
|
||||
# Import the cloud integration in an executor to avoid blocking the
|
||||
# event loop.
|
||||
def import_cloud() -> None:
|
||||
"""Import the cloud integration."""
|
||||
# pylint: disable-next=import-outside-toplevel
|
||||
from homeassistant.components.cloud import http_api # noqa: F401
|
||||
|
||||
await hass.async_add_import_executor_job(import_cloud)
|
||||
|
||||
# The cloud integration is imported locally to avoid cloud being imported by
|
||||
# bootstrap.py and to avoid circular imports.
|
||||
|
||||
# pylint: disable-next=import-outside-toplevel
|
||||
from homeassistant.components.cloud import http_api as cloud_http
|
||||
|
||||
# pylint: disable-next=import-outside-toplevel,hass-component-root-import
|
||||
from homeassistant.components.cloud.const import DATA_CLOUD
|
||||
|
||||
def with_cloud[_ViewT: _BaseOnboardingView, **_P](
|
||||
func: Callable[
|
||||
Concatenate[_ViewT, web.Request, _P],
|
||||
Coroutine[Any, Any, web.Response],
|
||||
],
|
||||
) -> Callable[
|
||||
Concatenate[_ViewT, web.Request, _P], Coroutine[Any, Any, web.Response]
|
||||
]:
|
||||
"""Home Assistant API decorator to check onboarding and cloud."""
|
||||
|
||||
@wraps(func)
|
||||
async def _with_cloud(
|
||||
self: _ViewT,
|
||||
request: web.Request,
|
||||
*args: _P.args,
|
||||
**kwargs: _P.kwargs,
|
||||
) -> web.Response:
|
||||
"""Check onboarding status, cloud and call function."""
|
||||
if self._data["done"]:
|
||||
# If at least one onboarding step is done, we don't allow accessing
|
||||
# the cloud onboarding views.
|
||||
raise HTTPUnauthorized
|
||||
|
||||
hass = request.app[KEY_HASS]
|
||||
if DATA_CLOUD not in hass.data:
|
||||
return self.json(
|
||||
{"code": "cloud_disabled"},
|
||||
status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
|
||||
)
|
||||
|
||||
return await func(self, request, *args, **kwargs)
|
||||
|
||||
return _with_cloud
|
||||
|
||||
class CloudForgotPasswordView(
|
||||
_NoAuthBaseOnboardingView, cloud_http.CloudForgotPasswordView
|
||||
):
|
||||
"""View to start Forgot Password flow."""
|
||||
|
||||
url = "/api/onboarding/cloud/forgot_password"
|
||||
name = "api:onboarding:cloud:forgot_password"
|
||||
|
||||
@with_cloud
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle forgot password request."""
|
||||
return await super()._post(request)
|
||||
|
||||
class CloudLoginView(_NoAuthBaseOnboardingView, cloud_http.CloudLoginView):
|
||||
"""Login to Home Assistant Cloud."""
|
||||
|
||||
url = "/api/onboarding/cloud/login"
|
||||
name = "api:onboarding:cloud:login"
|
||||
|
||||
@with_cloud
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle login request."""
|
||||
return await super()._post(request)
|
||||
|
||||
class CloudLogoutView(_NoAuthBaseOnboardingView, cloud_http.CloudLogoutView):
|
||||
"""Log out of the Home Assistant cloud."""
|
||||
|
||||
url = "/api/onboarding/cloud/logout"
|
||||
name = "api:onboarding:cloud:logout"
|
||||
|
||||
@with_cloud
|
||||
async def post(self, request: web.Request) -> web.Response:
|
||||
"""Handle logout request."""
|
||||
return await super()._post(request)
|
||||
|
||||
class CloudStatusView(_NoAuthBaseOnboardingView):
|
||||
"""Get cloud status view."""
|
||||
|
||||
url = "/api/onboarding/cloud/status"
|
||||
name = "api:onboarding:cloud:status"
|
||||
|
||||
@with_cloud
|
||||
async def get(self, request: web.Request) -> web.Response:
|
||||
"""Return cloud status."""
|
||||
hass = request.app[KEY_HASS]
|
||||
cloud = hass.data[DATA_CLOUD]
|
||||
return self.json({"logged_in": cloud.is_logged_in})
|
||||
|
||||
hass.http.register_view(CloudForgotPasswordView(data))
|
||||
hass.http.register_view(CloudLoginView(data))
|
||||
hass.http.register_view(CloudLogoutView(data))
|
||||
hass.http.register_view(CloudStatusView(data))
|
||||
|
||||
|
||||
@callback
|
||||
def _async_get_hass_provider(hass: HomeAssistant) -> HassAuthProvider:
|
||||
"""Get the Home Assistant auth provider."""
|
||||
|
||||
@ -173,11 +173,10 @@ IGNORE_VIOLATIONS = {
|
||||
"logbook",
|
||||
# Temporary needed for migration until 2024.10
|
||||
("conversation", "assist_pipeline"),
|
||||
# The onboarding integration provides limited backup and cloud APIs for use
|
||||
# The onboarding integration provides limited backup for use
|
||||
# during onboarding. The onboarding integration waits for the backup manager
|
||||
# and cloud to be ready before calling any backup or cloud functionality.
|
||||
# and to be ready before calling any backup functionality.
|
||||
("onboarding", "backup"),
|
||||
("onboarding", "cloud"),
|
||||
}
|
||||
|
||||
|
||||
|
||||
165
tests/components/cloud/test_onboarding.py
Normal file
165
tests/components/cloud/test_onboarding.py
Normal file
@ -0,0 +1,165 @@
|
||||
"""Test the onboarding views."""
|
||||
|
||||
from http import HTTPStatus
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import onboarding
|
||||
from homeassistant.components.cloud import DOMAIN
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from tests.common import register_auth_provider
|
||||
from tests.typing import ClientSessionGenerator
|
||||
|
||||
|
||||
def mock_onboarding_storage(hass_storage, data):
|
||||
"""Mock the onboarding storage."""
|
||||
hass_storage[onboarding.STORAGE_KEY] = {
|
||||
"version": onboarding.STORAGE_VERSION,
|
||||
"data": data,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def auth_active(hass: HomeAssistant) -> None:
|
||||
"""Ensure auth is always active."""
|
||||
await register_auth_provider(hass, {"type": "homeassistant"})
|
||||
|
||||
|
||||
@pytest.fixture(name="setup_cloud", autouse=True)
|
||||
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, {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("method", "view", "kwargs"),
|
||||
[
|
||||
(
|
||||
"post",
|
||||
"cloud/forgot_password",
|
||||
{"json": {"email": "hello@bla.com"}},
|
||||
),
|
||||
(
|
||||
"post",
|
||||
"cloud/login",
|
||||
{"json": {"email": "my_username", "password": "my_password"}},
|
||||
),
|
||||
("post", "cloud/logout", {}),
|
||||
("get", "cloud/status", {}),
|
||||
],
|
||||
)
|
||||
async def test_onboarding_view_after_done(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
method: str,
|
||||
view: str,
|
||||
kwargs: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test raising after onboarding."""
|
||||
mock_onboarding_storage(hass_storage, {"done": [onboarding.const.STEP_USER]})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
|
||||
resp = await client.request(method, f"/api/onboarding/{view}", **kwargs)
|
||||
|
||||
assert resp.status == 401
|
||||
|
||||
|
||||
async def test_onboarding_cloud_forgot_password(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test cloud forgot password."""
|
||||
mock_onboarding_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
|
||||
mock_cognito = cloud.auth
|
||||
|
||||
req = await client.post(
|
||||
"/api/onboarding/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_onboarding_cloud_login(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_onboarding_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.post(
|
||||
"/api/onboarding/cloud/login",
|
||||
json={"email": "my_username", "password": "my_password"},
|
||||
)
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"cloud_pipeline": None, "success": True}
|
||||
assert cloud.login.call_count == 1
|
||||
|
||||
|
||||
async def test_onboarding_cloud_logout(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_onboarding_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.post("/api/onboarding/cloud/logout")
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"message": "ok"}
|
||||
assert cloud.logout.call_count == 1
|
||||
|
||||
|
||||
async def test_onboarding_cloud_status(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_onboarding_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.get("/api/onboarding/cloud/status")
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"logged_in": False}
|
||||
@ -6,16 +6,12 @@ from http import HTTPStatus
|
||||
from io import StringIO
|
||||
import os
|
||||
from typing import Any
|
||||
from unittest.mock import ANY, DEFAULT, AsyncMock, MagicMock, Mock, patch
|
||||
from unittest.mock import ANY, AsyncMock, Mock, patch
|
||||
|
||||
from hass_nabucasa.auth import CognitoAuth
|
||||
from hass_nabucasa.const import STATE_CONNECTED
|
||||
from hass_nabucasa.iot import CloudIoT
|
||||
import pytest
|
||||
from syrupy import SnapshotAssertion
|
||||
|
||||
from homeassistant.components import backup, onboarding
|
||||
from homeassistant.components.cloud import DOMAIN as CLOUD_DOMAIN, CloudClient
|
||||
from homeassistant.components.onboarding import const, views
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
@ -31,6 +27,7 @@ from tests.common import (
|
||||
MockModule,
|
||||
MockUser,
|
||||
mock_integration,
|
||||
mock_platform,
|
||||
register_auth_provider,
|
||||
)
|
||||
from tests.test_util.aiohttp import AiohttpClientMocker
|
||||
@ -1073,142 +1070,6 @@ async def test_onboarding_backup_upload(
|
||||
mock_receive.assert_called_once_with(agent_ids=["backup.local"], contents=ANY)
|
||||
|
||||
|
||||
@pytest.fixture(name="cloud")
|
||||
async def cloud_fixture() -> AsyncGenerator[MagicMock]:
|
||||
"""Mock the cloud object.
|
||||
|
||||
See the real hass_nabucasa.Cloud class for how to configure the mock.
|
||||
"""
|
||||
with patch(
|
||||
"homeassistant.components.cloud.Cloud", autospec=True
|
||||
) as mock_cloud_class:
|
||||
mock_cloud = mock_cloud_class.return_value
|
||||
|
||||
mock_cloud.auth = MagicMock(spec=CognitoAuth)
|
||||
mock_cloud.iot = MagicMock(
|
||||
spec=CloudIoT, last_disconnect_reason=None, state=STATE_CONNECTED
|
||||
)
|
||||
|
||||
def set_up_mock_cloud(
|
||||
cloud_client: CloudClient, mode: str, **kwargs: Any
|
||||
) -> DEFAULT:
|
||||
"""Set up mock cloud with a mock constructor."""
|
||||
|
||||
# Attributes set in the constructor with parameters.
|
||||
mock_cloud.client = cloud_client
|
||||
|
||||
return DEFAULT
|
||||
|
||||
mock_cloud_class.side_effect = set_up_mock_cloud
|
||||
|
||||
# Attributes that we mock with default values.
|
||||
mock_cloud.id_token = None
|
||||
mock_cloud.is_logged_in = False
|
||||
|
||||
yield mock_cloud
|
||||
|
||||
|
||||
@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, CLOUD_DOMAIN, {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("setup_cloud")
|
||||
async def test_onboarding_cloud_forgot_password(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test cloud forgot password."""
|
||||
mock_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
|
||||
mock_cognito = cloud.auth
|
||||
|
||||
req = await client.post(
|
||||
"/api/onboarding/cloud/forgot_password", json={"email": "hello@bla.com"}
|
||||
)
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
assert mock_cognito.async_forgot_password.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("setup_cloud")
|
||||
async def test_onboarding_cloud_login(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.post(
|
||||
"/api/onboarding/cloud/login",
|
||||
json={"email": "my_username", "password": "my_password"},
|
||||
)
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"cloud_pipeline": None, "success": True}
|
||||
assert cloud.login.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("setup_cloud")
|
||||
async def test_onboarding_cloud_logout(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.post("/api/onboarding/cloud/logout")
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"message": "ok"}
|
||||
assert cloud.logout.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("setup_cloud")
|
||||
async def test_onboarding_cloud_status(
|
||||
hass: HomeAssistant,
|
||||
hass_storage: dict[str, Any],
|
||||
hass_client: ClientSessionGenerator,
|
||||
cloud: MagicMock,
|
||||
) -> None:
|
||||
"""Test logging out from cloud."""
|
||||
mock_storage(hass_storage, {"done": []})
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client()
|
||||
req = await client.get("/api/onboarding/cloud/status")
|
||||
|
||||
assert req.status == HTTPStatus.OK
|
||||
data = await req.json()
|
||||
assert data == {"logged_in": False}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("domain", "expected_result"),
|
||||
[
|
||||
@ -1286,3 +1147,59 @@ async def test_wait_integration_startup(
|
||||
|
||||
# The component has been loaded
|
||||
assert "test" in hass.config.components
|
||||
|
||||
|
||||
async def test_not_setup_platform_if_onboarded(
|
||||
hass: HomeAssistant, hass_storage: dict[str, Any]
|
||||
) -> None:
|
||||
"""Test if onboarding is done, we don't setup platforms."""
|
||||
mock_storage(hass_storage, {"done": onboarding.STEPS})
|
||||
|
||||
platform_mock = Mock(async_setup_views=AsyncMock(), spec=["async_setup_views"])
|
||||
mock_platform(hass, "test.onboarding", platform_mock)
|
||||
assert await async_setup_component(hass, "test", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(platform_mock.async_setup_views.mock_calls) == 0
|
||||
|
||||
|
||||
async def test_setup_platform_if_not_onboarded(
|
||||
hass: HomeAssistant, hass_storage: dict[str, Any]
|
||||
) -> None:
|
||||
"""Test if onboarding is not done, we setup platforms."""
|
||||
platform_mock = Mock(async_setup_views=AsyncMock(), spec=["async_setup_views"])
|
||||
mock_platform(hass, "test.onboarding", platform_mock)
|
||||
assert await async_setup_component(hass, "test", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
platform_mock.async_setup_views.assert_awaited_once_with(hass, {"done": []})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"platform_mock",
|
||||
[
|
||||
Mock(some_method=AsyncMock(), spec=["some_method"]),
|
||||
Mock(spec=[]),
|
||||
],
|
||||
)
|
||||
async def test_bad_platform(
|
||||
hass: HomeAssistant,
|
||||
platform_mock: Mock,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test loading onboarding platform which doesn't have the expected methods."""
|
||||
mock_platform(hass, "test.onboarding", platform_mock)
|
||||
assert await async_setup_component(hass, "test", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert await async_setup_component(hass, "onboarding", {})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert platform_mock.mock_calls == []
|
||||
assert "'test.onboarding' is not a valid onboarding platform" in caplog.text
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user